Kafka pubsub: transactions and exactly-once consume-transform-produce - #4508
Draft
javier-aliaga wants to merge 3 commits into
Draft
Kafka pubsub: transactions and exactly-once consume-transform-produce#4508javier-aliaga wants to merge 3 commits into
javier-aliaga wants to merge 3 commits into
Conversation
7 tasks
Adds opt-in Kafka transactions support to the shared Kafka component: - producerTransactionsEnabled wraps every Publish/BulkPublish in a Kafka transaction on an idempotent producer. An aborted publish is never visible to read_committed consumers, and bulk publishes become atomic (all entries commit or none do, so per-entry error mapping is bypassed in favor of whole-batch failure). - transactionalIdPrefix controls the producer transactional.id; a random per-instance suffix is always appended so scaled replicas never fence each other. Abandoned transactions are aborted by the broker on transaction timeout. - consumerIsolationLevel (read_uncommitted/read_committed) is exposed as an independent consumer option. Transaction cleanup follows sarama's state machine: abortable states are aborted so the producer stays usable; a fatal transaction state closes the producer and the next publish lazily recreates it with the same transactional.id (epoch bump aborts the stale transaction broker-side). Client creation/invalidation is now guarded by a lock, and the AWS IAM client path shares the producer config application with the static path. Init validates the sarama preconditions: producerRequiredAcks=all, producerRetryMax >= 1 and Kafka >= 0.11. Part of dapr#4301 (transactional consume-transform-produce follows in a separate commit). Signed-off-by: Javier Aliaga <javier@diagrid.io>
…shes Adds consumerTransactionsEnabled: every delivery is processed inside a Kafka transaction, giving consume-transform-produce pipelines the Kafka Streams guarantee boundary through the Dapr sidecar. Mechanism: - Each consumer-group claim (topic-partition) lazily creates its own transactional producer. Deliveries within a claim are serial, so the producer has at most one open transaction by construction and no lock is ever held across the app handler (the deadlock that sank the previous attempt in dapr#4300). - The claim producer's transactional.id is stable (<prefix>-<group>-<topic>-<partition>): after a rebalance the new owner's producer epoch-fences a zombie's, and a producer recreated after a fatal error aborts its own stale transaction the same way. - A per-delivery transaction token (__txnToken) is injected into the event metadata, which the runtime already forwards to the app as HTTP headers / gRPC metadata. Publishes carrying metadata.__txnToken are routed into the correlated open transaction; the token itself is never forwarded as a record header. A token that does not match an open transaction fails loudly rather than silently downgrading to a non-transactional publish. - On handler success the consumer offset joins the transaction with the group member metadata (KIP-447) before commit; deliberately no MarkMessage afterwards, so a stale non-transactional autocommit can never regress a transactional commit. On handler error the transaction aborts and the existing retry path takes over with a fresh transaction per attempt. - Bulk subscribe is all-or-nothing: one transaction per batch, any entry failure aborts and redelivers the whole batch (partial success cannot coexist with atomicity). Delivery to the app itself remains at-least-once and non-Kafka side effects stay outside the transaction, as with Kafka Streams. - Record-less transactions: sarama silently skips offset commits for transactions that produced no records, so a delivery whose handler published nothing ends its (empty) transaction and commits the offset via a synchronous MarkMessage+Commit instead — with no records there is nothing for the offset to be atomic with. Autocommit is disabled in transactional mode so a stale background commit can never regress a transactional offset commit. Part of dapr#4301 (certification tests follow in a separate commit). Signed-off-by: Javier Aliaga <javier@diagrid.io>
Four deterministic scenarios against the real 3-broker cluster, exercising the component API directly (they certify the component's transactional semantics; runtime metadata pass-through is dapr/dapr's contract): 1. Transactional bulk publish is atomic: a batch with an oversized entry aborts as a whole and nothing becomes visible to read_committed consumers; a valid batch is fully visible. 2. consumerIsolationLevel: a record from an aborted transaction is delivered by a read_uncommitted component and never by a read_committed one. 3. Consume-transform-produce: the handler publishes an output carrying the delivery's __txnToken and fails twice before succeeding — exactly one output is visible to read_committed consumers, the aborted attempts' outputs stay hidden, and the input offset commits atomically with the output. 4. Bulk subscribe all-or-nothing: a per-entry failure aborts and redelivers the whole batch; offsets commit only after full success (record-less synchronous path). Closes dapr#4301 Signed-off-by: Javier Aliaga <javier@diagrid.io>
javier-aliaga
force-pushed
the
feat/kafka-exactly-once-semantics
branch
from
August 4, 2026 10:31
97610f8 to
1c254b4
Compare
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #4508 +/- ##
==========================================
+ Coverage 31.94% 32.92% +0.98%
==========================================
Files 353 354 +1
Lines 47723 38277 -9446
==========================================
- Hits 15243 12602 -2641
+ Misses 31277 24453 -6824
- Partials 1203 1222 +19 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Adds Kafka transactions support to the Kafka pubsub component (in the shared
common/component/kafkacode), enabling exactly-once processing between Kafka topics through three independently configurable capabilities:producerTransactionsEnabled): every publish — single or bulk — is wrapped in a Kafka transaction on an idempotent producer. Bulk publishes become atomic (all entries commit or none do), and aborted publishes are never visible to consumers reading withread_committed.consumerIsolationLevel: "read_committed"): subscriptions never receive records from open or aborted transactions.consumerTransactionsEnabled): every delivery is processed inside a Kafka transaction. The component injects a__txnTokenmetadata entry (reaching the app as a header via the runtime's existing metadata pass-through — no dapr/dapr changes needed). Publishes from the handler that echo the token join the delivery's open transaction, and the consumer offset commits atomically with them (KIP-447,AddMessageToTxnWithGroupMetadata). Handler failure aborts the whole transaction and the delivery is redelivered.Design notes
transactional.id=<prefix>-<consumerGroup>-<topic>-<partition>: zombie fencing across rebalances and crashes, and no lock held across the app handler (deliveries are serial within a claim).read_committed: leavingconsumerIsolationLevelunset selects it automatically; an explicitread_uncommittedfails validation. Autocommit is disabled in transactional consume mode, and record-less deliveries commit their offset synchronously (sarama silently skips offset commits for record-less transactions).producerTransactionsEnabledis supported for output;consumerTransactionsEnabledis rejected at init — the bindings model (separate input and output instances) cannot echo the token through the same component instance. Consume-transform-produce is pubsub-only.transactional.id/idempotence are producer configs,isolation.levelis a consumer config). A merged flag would force per-delivery transaction overhead onto consume-and-publish apps that don't need correlation, and serialized transactional publishing onto apps that only need transactional consume.Close()preserves the prompt consumer-group teardown contract from fix: Fix kafka consumer shutdown #3907 — the group closes immediately and never waits on an in-flight transactional publish (the producer is abandoned to process exit in that case).Scope note: the guarantee covers Kafka records and consumer offsets. The app handler itself still runs at-least-once — non-Kafka side effects are outside the transaction.
Issue reference
Closes #4301
Checklist
Please make sure you've completed the relevant tasks for this PR, out of the following list: