Page History

...

3.2 Extracting TransactionSession from TransactionManager

The existing TransactionManager (1999 lines) is refactored into two components:

TransactionManager retains a reference to TransactionSession for identity information (producerId, epoch) and delegates lifecycle calls to it.

State machine mapping:

...

The state transitions are identical. The extraction is mechanical: move identity fields and lifecycle methods out of TransactionManager, leave produce-path integration in place.

3.3 No Wire Protocol Changes

This KIP does NOT introduce new RPCs or modify existing RPC schemas. TransactionSession sends the same RPCs that TransactionManager inside KafkaProducer sends today:

3.4 Internal Network Client

TransactionSession uses its own lightweight NetworkClient instance for coordinator communication. This is the same pattern used by the consumer's HeartbeatThread and the admin client's internal network handling. The NetworkClient is configured with:

• The same bootstrap.servers and security settings as the producer/consumer.

• A separate connection to the transaction coordinator broker.

• No RecordAccumulator, no batching, no Sender thread -- just request/response for transaction RPCs.

When a KafkaProducer is constructed with an external TransactionSession, the producer's Sender thread does NOT send transaction RPCs. It only sends Produce RPCs. The TransactionSession handles all coordinator communication independently. This clean separation eliminates the priority queue (PriorityQueue<TxnRequestHandler>) in the current TransactionManager that interleaves transaction RPCs with produce RPCs on the same Sender thread.

3.5 TransactionSession.resume() -- Eliminating Reflection

The static factory method TransactionSession.resume() replaces Flink's resumeTransaction() reflection hack:

...

resume() creates a TransactionSession in IN_TRANSACTION state with the given identity, connects to the coordinator, and allows commitTransaction(), abortTransaction(), or completeTransaction() to be called.

Relationship to KIP-939's initTransactions(keepPreparedTxn=true): KIP-939 provides a public API for resuming a prepared transaction: call initTransactions(true), which sends InitProducerId with keepPreparedTxn=true and receives both the new epoch and the ongoing transaction's (OngoingTxnProducerId, OngoingTxnEpoch). This works correctly but requires a full KafkaProducer instance. TransactionSession.resume() achieves the same result -- re-establishing the transaction identity -- without the producer overhead. Internally, resume() may call InitProducerId(keepPreparedTxn=true) under the covers, but it does so through a lightweight network client rather than a full producer infrastructure. The two approaches are functionally equivalent; TransactionSession.resume() is an efficiency optimization, not a semantic change.

4. Use Cases

4.1 Apache Flink: Lightweight Transaction Completion

Flink's exactly-once KafkaSink separates the write path (writer subtask) from the commit path (committer). The committer must complete a transaction that was started by a different entity.

Three approaches compared:

A. Current Flink pattern (reflection -- pre-KIP-939):

```java
// Committer creates a full KafkaProducer, then uses reflection
FlinkKafkaInternalProducer<byte[], byte[]> committer = new FlinkKafkaInternalProducer<>(configs);
committer.resumeTransaction(producerId, epoch);  // Reflection on TransactionManager internals
committer.commitTransaction();
// Problem: fragile, breaks on Kafka client upgrades
```

B. KIP-939 pattern (public API -- no reflection, but heavyweight):

...

KIP-939 eliminates the reflection hack -- this is a significant improvement. But the committer still instantiates a full KafkaProducer with RecordAccumulator (batching infrastructure), Sender thread (network I/O loop), serializers, partitioners, interceptors, and compression codecs. None of these are used. The committer sends exactly one RPC: EndTxn.

C. TransactionSession pattern (this KIP -- lightweight, public API):

```java
// Committer creates a lightweight TransactionSession, no producer infrastructure
TransactionSession session = TransactionSession.resume(txnId, producerId, epoch, configs);
// Internally: connects to transaction coordinator only, no RecordAccumulator, no Sender
session.commitTransaction();  // Sends EndTxn (COMMIT)
session.close();
// ~Simple lines of code vs  TransactionManager + full KafkaProducer
```

What changes in Flink connector:

• FlinkKafkaInternalProducer.resumeTransaction() reflection code deleted.

• KafkaCommitter uses TransactionSession.resume() instead of creating a KafkaProducer.

• KafkaCommittable stores TransactionSession identity fields (already does: producerId, epoch, transactionalId).

• Writer subtask uses TransactionSession + KafkaProducer(configs, session) for the write path. The session is the shared identity; the producer is used only for send().

Relationship to KIP-939: This KIP does not replace KIP-939. It builds on KIP-939's primitives (keepPreparedTxn, PreparedTxnState, completeTransaction) and provides a lighter-weight client for the recovery/commit path. The underlying RPCs are identical.

4.2 Share Group Consumer: Transactional Acknowledgments (KIP-1289)

KIP-1289 defines AddShareAcksToTxnRequest (API 83) and TxnShareAcknowledgeRequest (API 84), but has no client-side implementation. With TransactionSession, the pattern is clean:

...

Without TransactionSession, the share consumer would need to receive the producerId and epoch from the producer through an out-of-band channel, then construct the AddShareAcksToTxnRequest manually. TransactionSession provides the shared identity object that makes this natural.

4.3 Kafka Streams: Cleaner Task Transaction Management

Current Streams pattern:

...

With TransactionSession:

...

...

The manual transactionInFlight tracking is replaced by TransactionSession's internal state machine. The heartbeat (KIP 1309) is automatic.

4.4 Kafka Connect: Exactly-Once Source Tasks

Current Connect pattern:

...

With TransactionSession:

...

State mapping:

3.3 No Wire Protocol Changes

This KIP reuses existing RPCs and schemas. TransactionSession becomes the new sender for all transaction-related requests:

3.4 Internal Network Client

TransactionSession uses a lightweight NetworkClient for coordinator communication, similar to the consumer's HeartbeatThread.

Dedicated Connection - Maintains its own connection to the coordinator broker.

4. Use Cases

4.1 Apache Flink: Lightweight Transaction Completion

Flink's exactly-once KafkaSink separates the write path (writer subtask) from the commit path (committer). 

```java
// Before (Flink -- reflection, fragile)
FlinkKafkaInternalProducer producer = new FlinkKafkaInternalProducer(configs);
producer.resumeTransaction(producerId, epoch);  // reflection on TransactionManager internals
producer.commitTransaction();

// After (this KIP -- public API, stable)
TransactionSession session = TransactionSession.resume(
    transactionalId, producerId, epoch, configs
);
session.commitTransaction();
session.close();
```

4.2 Share Group Consumer: Transactional Acknowledgments (KIP-1289)

```

// 1. Open shared session
TransactionSession session = new TransactionSession(configs);
session.initialize();
session.beginTransaction();

// 2. Producer writes records using session identity
KafkaProducer<K, V> producer = new KafkaProducer<>(producerConfigs, session);
producer.send(outputRecord);

// 3. Share consumer acknowledges within the SAME transaction
shareConsumer.acknowledgeTransactionally(session, acknowledgments);

// 4. Atomic commit for both entities
session.commitTransaction();

```

4.4 Kafka Connect: Exactly-Once Source Tasks

```java

class ExactlyOnceWorkerSourceTask {
    private TransactionSession session;
    private KafkaProducer<byte[], byte[]> producer;

    void maybeBeginTransaction() { session.beginTransaction(); }
    void commitTransaction()      { session.commitTransaction(); }
}

```

4.5 Custom 2PC Coordinators

```java

// Phase 1: Prepare
kafkaSession.beginTransaction();
producer.send(records);
database.prepareTransaction(dbTxnId); 
kafkaSession.prepareTransaction();    

// Phase 2: Commit (Recovery-friendly)
TransactionSession resumed = TransactionSession.resume(txnId, pid, epoch, configs);
resumed.commitTransaction();          
database.commitTransaction(dbTxnId);

```

4.6 Kafka Connect Sink with Share Groups: Exactly-Once Kafka-to-Kafka (KIP-1302)

// KIP-1302 exactly-once pattern with TransactionSession
void iterationExactlyOnce() {
    ConsumerRecords<byte[], byte[]> records = shareConsumer.poll(pollTimeout);
    
    session.beginTransaction();
    try {
        // 1. Producer writes output records
        for (SinkRecord record : convertMessages(records)) {
            producer.send(new ProducerRecord<>(outputTopic, record.key(), record.value()));
        }

        // 2. Share consumer acknowledges input within the SAME session
        session.addShareAcksToTransaction(
            shareConsumer.groupMetadata().groupId(),
            ShareAcknowledgements.fromRecords(records, AcknowledgeType.ACCEPT)
        );

        session.commitTransaction();
    } catch (Exception e) {
        session.abortTransaction();
    }
}

Architectural comparison:

Image Added

...

4.5 Custom 2PC Coordinators

Any application implementing distributed transactions across Kafka and external systems (databases, object stores) can use TransactionSession as the Kafka leg of the 2PC:

...

4.6 Kafka Connect Sink with Share Groups: Exactly-Once Kafka-to-Kafka (KIP-1302)

KIP-1302 introduces WorkerShareSinkTask -- a Connect runtime class that drives SinkTask using a KafkaShareConsumer instead of a KafkaConsumer. This enables queue semantics (elastic scaling independent of partition count, no rebalances, no head-of-line blocking) for all sink connectors.

KIP-1302 defines two delivery modes:

The exactly-once mode is the critical integration point with this KIP. It requires three entities to participate in a single transaction:

1. KafkaProducer -- writes transformed output records to the output topic.

2. KafkaShareConsumer -- sends transactional acknowledgments (AddShareAcksToTxn, API 83) for the input records.

3. Transaction Coordinator -- coordinates the atomic commit of both output records and source acknowledgments.

The Problem Without KIP:

KIP-1302's exactly-once code path (from the KIP):

...

This works but has fundamental coupling issues:

1. sendShareAcksToTransaction() does not exist yet and would have to be added to KafkaProducer. This method has no business being on a record producer. It is a share consumer operation that happens to need the transaction identity. Without this KIP, KIP-1302's EOS phase has no clean home for this API -- it must either add it to the Producer<K, V> interface (polluting the record producer abstraction) or send AddShareAcksToTxnRequest directly from the worker with credentials extracted from the producer.

2. The transaction identity (producerId, epoch) flows through the producer, making it impossible for the share consumer to participate in the transaction independently.

3. If the Connect runtime wants to separate the poll/process thread from the commit thread (as Flink does), there is no way to hand off the transaction identity without reflection.

4. The heartbeat for transaction liveness (KIP 1309) must run inside the producer, even though the long-running phase is often the task.put() call (controlled by the Connect runtime, not the producer).

The Solution With this KIP:

...

What changes in WorkerShareSinkTask:

...

Architectural comparison:

...

Why this matters for KIP-1302 specifically:

...

Gives addShareAcksToTransaction() a proper home. Without this KIP, KIP-1302's EOS implementation must add this method to KafkaProducer or the Producer<K, V> interface -- neither of which is the right abstraction. TransactionSession.addShareAcksToTransaction() is the correct home: it is a transaction lifecycle operation that happens to originate from a share consumer.

...

Enables async commit. KIP-1302's WorkerShareSinkTask may want to separate the poll/process thread from the commit thread (similar to Flink's writer/committer split). With TransactionSession, the commit thread can call session.commitTransaction() without needing a producer instance.

...

Transaction heartbeat covers the full window. With the session owning the heartbeat (KIP 1309), liveness detection covers the entire transaction duration -- including task.put() latency, share consumer polling, and output record production. Without this KIP, the heartbeat would need to be in the producer, which may be idle during task.put().

...

...
5. Compatibility, Deprecation, and Migration Plan
...