From 8ee1dd1602e22f9f9244afea56fa5d3515af4597 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Sun, 23 Aug 2026 03:52:58 -0400 Subject: [PATCH] docs(source): promote Begin's sole-settle-path and panic rules to the primary doc positions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gate rerun 7 found that the Begin-then-Settle disposition and fn-panic semantics — already defined since the earlier gate rounds — sat in a poison-handling paragraph reviewers consistently missed. This promotes both rules into Begin's opening paragraph, cross-references the rule from inlet.go's Settle doc, documents SeekToCursor's re-delivery convention (cursor offset onward) in the README, and adds Reject/Skip rows to the README ack-model table. --- source/inlet.go | 4 +++- source/kafka/README.md | 6 ++++-- source/kafka/capability.go | 7 ++++++- 3 files changed, 13 insertions(+), 4 deletions(-) diff --git a/source/inlet.go b/source/inlet.go index aa61e0e..44ce6b4 100644 --- a/source/inlet.go +++ b/source/inlet.go @@ -71,7 +71,9 @@ type Subscription interface { // subscription never delivered, or after drain completed, is // backend-defined: where detectable it is rejected (Kafka rejects messages // that are not Kafka records; a record from a sibling subscription of the - // same backend is applied as if delivered). + // same backend is applied as if delivered). For messages settled through a + // transactional [source.Transactional.Begin], that Begin-only-settle-path + // rule governs instead — see its documentation. Settle(ctx context.Context, m Message, r Result) error // Close begins a graceful drain: from then on Next stops yielding fresh // messages — only messages already handed out or buffered before Close may diff --git a/source/kafka/README.md b/source/kafka/README.md index 08f401b..ad483a6 100644 --- a/source/kafka/README.md +++ b/source/kafka/README.md @@ -35,7 +35,8 @@ and the marked offsets are committed on graceful drain and on rebalance | `Ack` | mark the record for commit (commit-after-process) | | `Nak` | never mark; pause, re-seek to the record's offset, resume — the record is fetched again in this session | | `NakAfter(d)` | same as `Nak`, waiting out `d` between pause and re-seek (best-effort) | -| `Term` | produce the record to the dead-letter topic, then mark (transactional subscriptions: rejected with `ErrTermInsideTransaction`, leaving the record unmarked so it is redelivered; route poison through `Begin` instead). If the DLQ produce fails, `Settle` returns the error and the record is **not** marked — it is redelivered; nothing is silently lost | +| `Term` | produce the record to the dead-letter topic, then mark (transactional subscriptions: rejected with `ErrTermInsideTransaction`, leaving the record unmarked so it is redelivered; route poison through `Begin` instead). If the DLQ produce fails, `Settle` returns the error and the record is **not** marked — it is redelivered; nothing is silently lost. `Reject` maps here too (same behavior, `InvalidForState` class) | +| `Skip` | ack and discard (`Drop` class): marked like an Ack, not retried, not dead-lettered | | `InProgress` | no-op (Kafka has no per-message ack deadline) | | `Manual` | no-op (the handler settled via `Message.As` + the client) | @@ -75,7 +76,8 @@ do expose it: - `Seekable` — live offset reposition via `SetOffsets` (and `ListOffsets` for time-based seeks), the basis for replay. Partitions are enumerated from the group's committed offsets when present, else discovered from broker metadata, - so seeking works before the first commit. + so seeking works before the first commit. `SeekToCursor` re-delivers from + the cursor's own offset onward. - `ConsumerGroups` — `GroupID` plus assign/revoke hooks; the adapter drain-and-commits marked offsets on a graceful revoke and skips the commit on an ungraceful loss. diff --git a/source/kafka/capability.go b/source/kafka/capability.go index e1af782..6645468 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -413,7 +413,12 @@ func toRecordHeaders(hs source.Headers) []kgo.RecordHeader { // Begin runs fn inside a Kafka producer transaction so the records fn produces // through the handed [source.Tx] are committed (or aborted) atomically with the -// consumed offset of m — exactly-once consume-process-produce. The choreography +// consumed offset of m — exactly-once consume-process-produce. Begin replaces +// [Subscription.Settle] for m entirely: a direct Settle for m during or after +// Begin is a caller error with undefined results (at best a harmless duplicate +// mark; at worst a race with End's commit/abort), and fn must not panic — a +// panic propagates unrecovered, leaving the transaction open until End, session +// close, or a rebalance fence aborts it. The choreography // is: begin the transaction, run fn (which produces emitted records through the // txHandle), mark m's offset for commit on success, then End with TryCommit if fn // succeeded or TryAbort if it failed. On commit, franz-go flushes the produced