...

Abort case: 

6. Corner Cases

Case 1: Crash BEFORE beginTransaction()

poll() -> records [0..9] ACQUIRED
<CRASH>

What happens:

• No transaction was started, no output produced, no acks sent
• ACQUIRED locks on records 0-9 expire (controlled by share.record.lock.duration.ms)
• Broker re-delivers records 0-9 to another member of the share group
• No data loss, no duplicates

Case 2: Crash AFTER send() but BEFORE commitTransaction()

poll() -> records [0..9] ACQUIRED
beginTransaction()
send(output for 0-9)       ← records written to broker, but transactional
<CRASH>

What happens:

• Output records are in the partition log but invisible (read_committed consumers skip uncommitted)
• No EndTxnRequest was sent
• TC detects transaction timeout (transaction.timeout.ms), auto-aborts
• TC sends WriteTxnMarkers(ABORT) -> output records get ABORT marker, permanently invisible
• Share group: no acks were sent, ACQUIRED locks expire, records 0-9 re-delivered
• No data loss, no duplicates

Case 3: Crash AFTER sendShareAcksToTransaction() but BEFORE commitTransaction()

poll() -> records [0..9] ACQUIRED
beginTransaction()
send(output)
sendShareAcksToTransaction()  ← acks written to SGC as PENDING
<CRASH>

What happens:

• Output records: in log, invisible (uncommitted)
• Share acks: stored in SGC as PENDING (not materialized)
• TC detects timeout, sends WriteTxnMarkers(ABORT) to both data partitions AND __share_group_state
• Output: ABORT marker written, records permanently invisible
• Share acks: SGC discards pending acks, locks expire, records 0-9 re-delivered
• No data loss, no duplicates -- this is the critical case that KIP-1289 solves

Case 4: Crash AFTER EndTxn(COMMIT) sent but BEFORE WriteTxnMarkers completes

poll() -> records [0..9] ACQUIRED
beginTransaction()
send(output)
sendShareAcksToTransaction()
commitTransaction()            ← EndTxn sent, PREPARE_COMMIT logged
<TC crashes or broker restart>

What happens:

• PREPARE_COMMIT is durable in __transaction_state log
• When TC recovers (or new leader elected for __transaction_state partition), it replays the log
• TC sees PREPARE_COMMIT, resumes sending WriteTxnMarkers(COMMIT) to all partitions
• Output records become visible, share acks materialized
• No data loss, no duplicates -- the two-phase commit guarantees completion

Case 5: ProducerFencedException (zombie detection)

Instance A: beginTransaction(), send(), ...
Instance B: initTransactions() with same transactional.id
            ← TC bumps epoch, A is now a zombie
Instance A: commitTransaction()
            ← TC rejects: ProducerFencedException

What happens:

• Instance A's transaction is aborted by the TC (epoch fenced)
• Any pending output and share acks from A are discarded
• Instance B takes over, re-processes the records
• No data loss, no duplicates -- fencing prevents split-brain

Case 6: abortTransaction() called explicitly

poll() -> records [0..9] ACQUIRED
beginTransaction()
send(output for 0-4)
record 5 fails validation
abortTransaction()             ← explicit abort

What happens:

• TC writes PREPARE_ABORT, sends WriteTxnMarkers(ABORT)
• Output records 0-4: ABORT marker, permanently invisible
• Share acks (if any sent): discarded
• ACQUIRED locks expire, all records 0-9 re-delivered
• Application gets fresh batch, can retry
• No data loss

Case 7: Partial produce failure (network error to one partition)

beginTransaction()
send("enriched-orders-0", rec1)   ← success
send("enriched-orders-1", rec2)   ← network error, KafkaException

What happens:

• Application catches KafkaException, calls abortTransaction()
• Same as Case 6: everything rolled back, records re-delivered
• No data loss

The Guarantee Matrix

...

Failure Scenarios

• Crash Before/During Txn: If the application crashes before commitTransaction(), the Transaction Coordinator (TC) auto-aborts

• The "Critical Case" (Crash after sendShareAcks): If a crash occurs after acks are sent but before the final commit, the TC sends Abort  signal and ensures pending acks are discarded.

• Zombie Producers: If a new producer instance starts, the old one is fenced. Any in-flight output or acks are discarded by the TC, preventing "split-brain" duplicates.

• Post-Prepare Crash: If the TC crashes after logging PREPARE_COMMIT, the 2PC protocol ensures that the transaction is finalized upon recovery. Both outputs and acks will successfully materialize.

7. Compatibility, Deprecation, and Migration Plan

• Zero Breaking Changes: Current implicit and explicit modes remain the default and continue to function as-is.

• Opt-in Requirement: Users must explicitly enable the new mode via share.acknowledgement.mode = transactional.

• Version Safety: * Clients will "fail fast" if the connected broker does not support the new TxnShareAcknowledgeRequest.

  • Fallback logic is included to issue warnings if a downgrade to non-transactional acks is necessary.

8. Test Plan

The verification strategy focuses on state machine integrity and fault tolerance under high-concurrency and failure scenarios.

• Unit Tests: Validates state transitions (e.g., ACQUIRED to ACKNOWLEDGED on commit vs. AVAILABLE on abort), idempotency of operations, and transaction timeout/auto-abort logic.

• Integration Tests: Focuses on end-to-end commit/abort flows, coordinator recovery, and multi-consumer behavior within a single group during network partitions.

• System & Performance Tests: Benchmarks transactional vs. non-transactional modes and verifies exactly-once delivery within Flink/Spark checkpointing cycles.

• Chaos Tests: Simulates broker and coordinator crashes specifically during critical phases like PREPARE_COMMIT to ensure protocol durability.

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.