Versions Compared

Old Version 11

changes.mady.by.user Shekhar Rajak

Saved on

compared with

New Version 12

changes.mady.by.user Shekhar Rajak

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

The session identity is not multiplied -- there is one session, one producerId, one epoch, regardless of how many clients use it.

7. Relationship to KIP-939 (Support Participation in 2PC)

KIP-939 (Accepted, authored by Artem Livshits) is the foundational KIP that enables Kafka to participate in externally-coordinated two-phase commit. This KIP  is explicitly designed as a complement to KIP-939, not a replacement. The following table clarifies the boundary:

Concern

KIP-939 (Accepted)

This KIP 

Problem solved

Make Kafka a proper 2PC participant

Provide a lightweight, entity-agnostic client abstraction for transaction lifecycle

Scope

Broker-side + producer API

Client-side refactoring only

Wire protocol changes

InitProducerIdRequest v6: Enable2Pc, KeepPreparedTxn; InitProducerIdResponse: OngoingTxnProducerId, OngoingTxnEpoch

None (reuses KIP-939's RPCs at the same versions)

New producer methods

initTransactions(boolean keepPreparedTxn), prepareTransaction(), completeTransaction(PreparedTxnState)

None (wraps existing methods)

Transaction timeout

enable2Pc=true sets txnTimeoutMs = MAX_INT, transaction never auto-aborted

No change (relies on KIP-939 behavior)

Recovery after crash

initTransactions(true) + completeTransaction(savedState) on a new KafkaProducer instance

TransactionSession.resume(txnId, pid, epoch) + commitTransaction() -- same RPCs, lighter client

Flink reflection hack

Provides public API alternative (initTransactions(true))

Provides lightweight alternative that avoids creating full producer

Multi-entity transactions

Not addressed (predates KIP-1289, KIP-1302)

Core motivation: shared TransactionSession for producer + share consumer

ACL model

Adds TWO_PHASE_COMMIT operation on TRANSACTIONAL_ID resource

No change (reuses KIP-939 ACLs)

What KIP-939 Got Right

  1. Implicit prepare. KIP-939 rejected an explicit "prepare" RPC, keeping Kafka's existing "implicit prepare" (flush-as-prepare). The external coordinator tracks prepared state, not the broker. This avoids duplicating state and an extra synchronous operation on the transaction coordinator topic. TransactionSession.prepareTransaction() wraps this same implicit-prepare mechanism.

  2. keepPreparedTxn flag. Separating "don't abort the ongoing transaction" from "enable 2PC" was correct. Flink needs keepPreparedTxn=true even without enable2Pc=true (for clusters that don't grant 2PC privileges). TransactionSession uses this flag transparently.

  3. PreparedTxnState as serializable {producerId, epoch}. This is the portable transaction identity that can be stored in any database. TransactionSession.resume() accepts the same identity fields.

What KIP-939's Rejected Alternatives Reveal

KIP-939 explicitly rejected a HeartBeat RPC (page 11): "HeartBeat RPC definitely sounds like a 'good thing to do'. It is not clear, though, what would be the cases when we need to handle these situations differently." Since then, KIP-1309

Jira
serverASF JIRA
serverId5aa69414-a9e9-3523-82ec-879b028fb15b
keyKAFKA-20381
has been proposed to add exactly this heartbeat, with strong community support. The evolution from KIP-939's rejection to KIP-1309's proposal demonstrates that the ecosystem's needs have grown beyond KIP-939's original scope. TransactionSession provides the natural home for the heartbeat thread (KIP 1309), which runs independently of the producer's Sender thread.

Why TransactionSession Is Not Redundant With KIP-939

The key question: "If KIP-939 provides initTransactions(true) + completeTransaction(), why do we need TransactionSession?"

Answer: KIP-939 solved the protocol problem. this KIP solves the abstraction problem.

KIP-939 added the correct primitives to the producer API. But the producer API is the wrong abstraction for three emerging use cases that KIP-939 did not anticipate:

  1. KIP-1289 (Transactional Share Acks): A share consumer must send AddShareAcksToTxnRequest using the producer's (producerId, epoch). There is no method for this on KafkaProducer today, and adding one would pollute the record-producer interface with share-consumer semantics.

  2. KIP-1302 (Share Groups in Connect Sink): The exactly-once Kafka-to-Kafka path requires three entities (producer, share consumer, transaction coordinator) to participate in one transaction. The shared identity must be accessible to both producer and consumer without one proxying through the other.

  3. Jira
    serverASF JIRA
    serverId5aa69414-a9e9-3523-82ec-879b028fb15b
    keyKAFKA-20381
    /KIP-1309 (Transaction Heartbeat): The heartbeat thread should belong to the transaction session, not the producer. The producer may be idle (no records to send) while the transaction is active during a long checkpoint. The heartbeat must continue independently.

TransactionSession is the missing abstraction that connects KIP-939's 2PC primitives, KIP-1289's multi-entity transactions, KIP-1302's Connect sink EOS, and KIP 1309's liveness detection into a coherent client-side architecture.

8. Test Plan

8.1 Unit Tests

Test

Description

TransactionSession lifecycle

Verify state transitions: UNINITIALIZED -> INITIALIZING -> READY -> IN_TRANSACTION -> COMMITTING -> READY.

TransactionSession.resume()

Verify that a resumed session starts in IN_TRANSACTION state and can call commitTransaction().

TransactionSession with heartbeat

Verify that heartbeat thread starts when transaction.session.timeout.ms > 0 and stops on commitTransaction().

KafkaProducer with external session

Verify that beginTransaction()/commitTransaction() on the producer throw IllegalStateException when constructed with an external session.

KafkaProducer with internal session

Verify backward compatibility: initTransactions(), beginTransaction(), commitTransaction() work exactly as before.

TransactionSession epoch fencing

Verify that a second session with the same transactionalId fences the first (epoch bump).

TransactionSession identity accessors

Verify producerId(), producerEpoch(), transactionalId() return correct values after initialize().

8.2 Integration Tests

...

Test

...

Description

...

Producer + external session E2E

...

Create TransactionSession, create KafkaProducer with it, produce records, commit via session. Verify read_committed consumer sees the records.

...

Resume and commit from different process

...

Session A begins transaction and produces records. Session B resumes with A's identity and commits. Verify read_committed consumer sees the records.

...

Share consumer transactional ack

...

Producer and share consumer share a TransactionSession. Producer writes output, share consumer acknowledges input, session commits atomically.

...

Streams with TransactionSession

...

StreamsProducer uses TransactionSession instead of direct KafkaProducer transaction calls. Verify exactly-once semantics preserved.

...

2PC with resume

...

Session begins transaction, prepares (KIP-939). Different session resumes and commits. Verify atomic commit.

...

Heartbeat via TransactionSession

...

...

8. Test Plan

8.1 Unit Tests

TestDescription
Session LifecycleVerify transitions: UNINITIALIZED → INITIALIZING → READY → IN_TRANSACTION → COMMITTING → READY.
Session ResumeVerify resume() starts in IN_TRANSACTION and can execute commitTransaction().
Heartbeat LogicVerify heartbeat thread lifecycle based on transaction.session.timeout.ms.
Producer FencingEnsure KafkaProducer throws IllegalStateException on lifecycle calls when using an external session.
Backward CompatibilityVerify standard KafkaProducer transaction methods work via internal session delegation.
Epoch FencingVerify that a new session with the same transactional.id correctly fences the older session.
Identity AccessorsEnsure producerId(), producerEpoch(), and transactionalId() are accurate post-initialization.

8.2 Integration Tests

Producer + external session E2E

Resume and commit from different process

8.3 Compatibility Tests

TestDescription

Old producer behavior unchanged

KafkaProducer with transactional.id config (no external session). Verify all existing transaction tests pass without modification.

Mixed: internal + external sessions

Legacy ProducerRun existing test suite on KafkaProducer without external sessions to ensure zero regression.
Mixed ModeRun internal and external sessions concurrently on the same cluster to verify
One producer uses internal session, another uses external session, both on same cluster. Verify
no interference.

9. Reference 

KIP-98 - Exactly Once Delivery and Transactional Messaging

...