diff --git a/source/inlet.go b/source/inlet.go index d9b4244..818db84 100644 --- a/source/inlet.go +++ b/source/inlet.go @@ -61,12 +61,18 @@ type Subscription interface { // ack/commit, schedule redelivery, route to dead-letter, or extend the // deadline, per Result.Action. It is the single point where a delivery // decision reaches the backend. Settling a message more than once is safe: - // every call is applied and recorded (a duplicate mark is idempotent at - // every backend), and in-flight accounting never goes negative, so drain - // semantics are unaffected. + // every call is applied and recorded in arrival order, and in-flight + // accounting never goes negative, so drain semantics are unaffected. A + // later decision supersedes an earlier one for future commits — an Ack + // after a Nak marks past the record — though a re-seek already issued may + // still yield one final redelivery (at-least-once). Settling a message this + // subscription never delivered, or after drain completed, is + // backend-defined: where detectable it is rejected (Kafka rejects non-Kafka + // messages); otherwise it applies as if delivered. Settle(ctx context.Context, m Message, r Result) error - // Close begins a graceful drain: Next stops yielding new messages, and once - // in-flight messages are settled, Next returns ErrDrained. Close is - // idempotent. + // Close begins a graceful drain: from then on Next stops yielding fresh + // messages — only messages already handed out or buffered before Close may + // still be yielded — and once in-flight messages are settled, Next returns + // ErrDrained. Close is idempotent. Close() error } diff --git a/source/kafka/README.md b/source/kafka/README.md index 17acf73..485b0ea 100644 --- a/source/kafka/README.md +++ b/source/kafka/README.md @@ -35,7 +35,7 @@ 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`; 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 | | `InProgress` | no-op (Kafka has no per-message ack deadline) | | `Manual` | no-op (the handler settled via `Message.As` + the client) | diff --git a/source/kafka/capability.go b/source/kafka/capability.go index 03c405d..b9901c1 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -439,10 +439,11 @@ func toRecordHeaders(hs source.Headers) []kgo.RecordHeader { // a loop simply sees the record again on its next fetch — route it through // Begin instead. Inside fn, produce the rejected record to the dead-letter // topic via the handed [source.Tx] and return nil, so the DLQ write and the -// consumed offset commit atomically. Begin is the only settle path for m while -// it runs: mixing a direct [Subscription.Settle] for m into an open Begin is a -// programming error; the direct settle marks independently and can double- -// commit or race the transaction's End. +// consumed offset commit atomically. Begin is the only settle path for m when +// used transactionally: mixing a direct [Subscription.Settle] for m into an +// open Begin is a caller error and its behavior is undefined — the direct +// settle marks independently of the transaction, so it can double-advance the +// partition or race End's commit/abort. func (s *subscription) Begin(ctx context.Context, m source.Message, fn func(ctx context.Context, tx source.Tx) error) error { if s.transactSess == nil { return fmt.Errorf("source/kafka: transactional: %w", errNotTransactional) diff --git a/source/kafka/kafka.go b/source/kafka/kafka.go index 460d9fa..a14d930 100644 --- a/source/kafka/kafka.go +++ b/source/kafka/kafka.go @@ -24,9 +24,10 @@ // not an in-session one. // - Term produces the record to the configured dead-letter topic, then marks // it for commit so it is not re-read — except on a transactional -// subscription, where Term reports [ErrTermInsideTransaction]: route poison -// through Begin's transaction instead, producing to the dead-letter topic -// via the handed [source.Tx] so DLQ write and offset commit are atomic. +// subscription, where Term reports [ErrTermInsideTransaction] and leaves +// the record unmarked, so it is redelivered: route poison through Begin's +// transaction instead, producing to the dead-letter topic via the handed +// [source.Tx] so DLQ write and offset commit are atomic. // - InProgress is a no-op: Kafka has no per-message ack deadline to extend. // - Manual is a no-op: the handler settled the record itself through // [source.Message.As] and the underlying *kgo.Client.