diff --git a/source/doc.go b/source/doc.go index 83bb96b..25ad95f 100644 --- a/source/doc.go +++ b/source/doc.go @@ -30,7 +30,9 @@ // restarts and rebalances, backends whose redelivery rides persisted offsets // (Kafka) resume from the last committed position, which a concurrently // committed higher offset can advance past a nacked record; those backends -// therefore redeliver exactly within a session and best-effort across restarts, +// therefore redeliver within the live session subject to each adapter's +// documented constraints (Kafka: a concurrent commit can pass a nacked record +// before its re-seek lands) and best-effort across restarts, // and each adapter documents the precise semantics in its own module. A handler // returns a [Result] — [Ack], [Nak], [Term], [InProgress], or [Manual] — and // the Hopper applies it to the backend. Backends differ (Kafka commits offsets diff --git a/source/inlet.go b/source/inlet.go index 818db84..aa61e0e 100644 --- a/source/inlet.go +++ b/source/inlet.go @@ -64,11 +64,14 @@ type Subscription interface { // 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 + // after a Nak marks past the record, though a re-seek already issued may + // still yield one final redelivery (at-least-once); a Nak after an Ack does + // not unmark, so the persisted position stays past the record while any + // in-session re-seek may still redeliver it 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. + // 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). 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 485b0ea..d23a2d7 100644 --- a/source/kafka/README.md +++ b/source/kafka/README.md @@ -122,6 +122,11 @@ track franz-go releases. ## Setup notes +- Configure `WithDLQTopic` before processing poison-prone streams. Without it, + every `Term` fails with `ErrNoDLQTopic` and the record stays unmarked — a + poison record redelivers on every fetch, stalling its partition until the + dead-letter topic is configured or the DLQ produce recovers. + - The dead-letter topic must exist before the first `Term` settles: the adapter's DLQ producer does not enable topic auto-creation. Create it up front (as the integration suite does with `CreateTopics`). diff --git a/source/kafka/capability.go b/source/kafka/capability.go index b359696..e1af782 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -428,7 +428,10 @@ func toRecordHeaders(hs source.Headers) []kgo.RecordHeader { // for m. A direct Settle for m during an open Begin, or after Begin returned, // is a caller error with undefined results — it is not detected or coordinated // with the transaction: at best it duplicates an already-committed mark -// (harmless at-least-once), at worst it races End's commit/abort. +// (harmless at-least-once), at worst it races End's commit/abort. fn must not +// panic: a panic is not recovered; it propagates to the caller of Begin, +// leaving the transaction open until a later End, the session closes, or a +// rebalance fences the producer (which aborts it). // // It is available only when the inlet was built with [WithTransactional]; // otherwise it reports the capability is absent rather than silently running fn