...

2. Public Interfaces

2.1 New Class: TransactionSession

Package: org.apache.kafka.clients.transaction

TransactionSession is a lightweight, thread-safe client focused strictly on transaction coordinator interaction.

It decouples lifecycle management from record production.

```

With this KIP every transactional use case in Kafka becomes the same three-step pattern — construct a TransactionSession, bind clients to it, drive begin/prepare/commit/abort/resume on the session —

with the only differences being which clients you bind (KafkaProducer, KafkaConsumer, KafkaShareConsumer) and which lifecycle methods you use (commit for simple cases, prepare+complete for 2PC, resume for crash recovery.

...

2. Public Interfaces

2.1 New Class: TransactionSession

Package: org.apache.kafka.clients.transaction

TransactionSession is a lightweight, thread-safe client focused strictly on transaction coordinator interaction.

It decouples lifecycle management from record production.

```

public classpublic class TransactionSession implements Closeable {

    public TransactionSession(Map<String, Object> configs);

    public static TransactionSession resume(
        String transactionalId,
        long producerId,
        short producerEpoch,
        Map<String, Object> configs
    );

    // --- Lifecycle ---
    public void initialize();
    public void beginTransaction();
    public PreparedTxnState prepareTransaction();
    public void completeTransaction(PreparedTxnState preparedTxnState);
    public void commitTransaction();
    public void abortTransaction();

    // --- Identity ---
    public String transactionalId();

    // --- Participation ---
    public void addPartitionsToTransaction(Collection<TopicPartition> partitions);
    
    public void sendOffsetsToTransaction(
        Map<TopicPartition, OffsetAndMetadata> offsets,
        ConsumerGroupMetadata groupMetadata
    );

    public void addShareAcksToTransaction(
        String groupId,
        Collection<ShareAcknowledgment> acknowledgments
    );

    // --- Liveness ---
    public void heartbeat();

    @Override
    public void close();
}

...

No code changes are required for existing users.

2.4 Integration with KafkaShareConsumer

...

For KIP-1289 transactional acknowledgments, the share consumer accepts a TransactionSession:

```

...

3. Proposed Changes

3.1 Architecture: Before and After

Before (current):

Image Removed

After (this KIP):

Image Removed

3.2 Extracting TransactionSession from TransactionManager

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). 

...

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

```

...

and Kafka Consumer

 void bindTransactionSession(TransactionSession session);
 void unbindTransactionSession();

...

3. Proposed Changes

3.1 Architecture: Before and After

Before (current):

Image Added

After (this KIP):

Image Added

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();
```

The checkpoint execution pseudo code: 

class FlinkKafkaSink {

    // Phase: process()
    TransactionSession session = new TransactionSession(cfg);
    session.initialize();
    KafkaProducer<K,V> producer = new KafkaProducer<>(cfg, session);

    void onCheckpointBarrier(long checkpointId) {
        session.beginTransaction();
        for (record : bufferedSinceLastBarrier) producer.send(record);

        // Phase 1: snapshotState() in Flink
        PreparedTxnState prepared = session.prepareTransaction();
        flinkCheckpoint.persist(checkpointId, session.transactionalId(),
                                session.producerId(), session.producerEpoch(),
                                prepared);
    }

    void onCheckpointComplete(long checkpointId) {
        // Phase 2: notifyCheckpointComplete() in Flink
        session.commitTransaction();
    }

    // Crash recovery — replaces FlinkKafkaInternalProducer.resumeTransaction(...)
    void recover(CheckpointMeta meta) {
        TransactionSession recovered = TransactionSession.resume(
            meta.txnId, meta.pid, meta.epoch, cfg);
        recovered.completeTransaction(meta.prepared);   // idempotent commit
    }
}

4.2  Consumer / Share Consumer 

CTP - Consumer awared transaction rough idea (future KIP) [may be we can plan to have it as part of this KIP as subtask]

TransactionSession session = new TransactionSession(cfg);
session.initialize();
KafkaProducer<K,V> producer = new KafkaProducer<>(cfg, session);
KafkaConsumer<K,V> consumer = new KafkaConsumer<>(cfg);
consumer.subscribe(List.of("input"));
consumer.bindTransactionSession(session);   // NEW

while (running) {
    var records = consumer.poll(Duration.ofSeconds(1));
    if (records.isEmpty()) continue;

    session.beginTransaction();
    try {

...

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);

...

// 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 (SinkRecordvar recordr : convertMessages(records)) { records) producer.send(transform(r));
        session.commitTransaction();    producer.send(new ProducerRecord<>(outputTopic, record.key(), record.value()));// offsets auto-pulled from bound consumer
    } catch (KafkaException e) }
{
        // 2. Share consumer acknowledges input within the SAME sessionsession.abortTransaction();    // consumer auto-rewinds
        session.addShareAcksToTransaction(
            shareConsumer.groupMetadata().groupId(),
            ShareAcknowledgements.fromRecords(records, AcknowledgeType.ACCEPT)
        );

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

•  consumer.bindTransactionSession(session) makes the consumer a participant.

•  During IN_TRANSACTION: commitSync/seek/subscribe throw; poll permitted.

•  During COMMITTING/ABORTING: poll blocks.

•  On abortTransaction(): consumer's in-memory cursor is rolled back to committed() for affected partitions — no application code required.

•  On commitTransaction(): session pulls latest consumed offsets from each bound consumer and includes them in TxnOffsetCommit automatically.

 Share consumer + producer in same transaction

TransactionSession session = new TransactionSession(cfg);
session.initialize();
KafkaProducer<K,V> producer = new KafkaProducer<>(cfg, session);
KafkaShareConsumer<K,V> share = new KafkaShareConsumer<>(shareCfg);
share.subscribe(List.of("input-queue"));
share.bindTransactionSession(session);

while (running) {
    var records   session.abortTransaction(= share.poll(Duration.ofSeconds(1));
    }
}

Architectural comparison:

...

if (records.isEmpty()) continue;

    session.beginTransaction();
    try {
        for (var r : records) {
            producer.send(transform(r));
            share.acknowledge(r, AcknowledgeType.ACCEPT);   // buffered into txn
        }
        session.commitTransaction();                         // atomic
    } catch (KafkaException e) {
        session.abortTransaction();                          // acks discarded → redeliver
    }
}

•  share.acknowledge(r, ACCEPT) while bound does not issue ShareAcknowledge; it buffers an entry.

•  share.acknowledge(r, RENEW) always issues immediately (liveness, not transactional).

•  session.commitTransaction() emits TxnShareAcknowledge (KIP-1289 wire RPC) stamped with (pid, epoch) + EndTxn(commit) —

 coordinator writes commit markers to producer's data partitions AND share-coordinator's state atomically.

•  On abort: buffered share acks are discarded; broker's share state is unchanged → locks expire → records redeliver.

4.4 Kafka Connect: Exactly-Once Source Tasks

Similar API

4.5 Custom 2PC Coordinators

Similar API

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

Similar API

Architectural comparison:

Image Added

...

5. Compatibility, Deprecation, and Migration Plan

5.1 Full Backward Compatibility

KafkaProducer constructed with transactional.id in the config continues to work exactly as today.

Internally, it creates a TransactionSession and delegates to it, but the public API (initTransactions(), beginTransaction(), commitTransaction(), abortTransaction(), sendOffsetsToTransaction()) is unchanged.

5.2 Deprecation Path

No deprecations. The KafkaProducer convenience methods remain the recommended API for simple produce-and-commit patterns.

TransactionSession is for advanced use cases: 2PC, cross-entity transactions, external coordinators.

We can plan for deprecation in future once this feature is stable.

...

6. Security

6.1 Authorization

TransactionSession requires the same ACLs as the current producer transaction API:

...

7. Test Plan

7.1 Unit Tests

7.2 Integration Tests

Producer + external session E2E

Resume and commit from different process

7.3 Compatibility Tests

5. Compatibility, Deprecation, and Migration Plan

5.1 Full Backward Compatibility

KafkaProducer constructed with transactional.id in the config continues to work exactly as today.

Internally, it creates a TransactionSession and delegates to it, but the public API (initTransactions(), beginTransaction(), commitTransaction(), abortTransaction(), sendOffsetsToTransaction()) is unchanged.

5.2 Deprecation Path

No deprecations. The KafkaProducer convenience methods remain the recommended API for simple produce-and-commit patterns. TransactionSession is for advanced use cases: 2PC, cross-entity transactions, external coordinators.

6. Security

6.1 Authorization

TransactionSession requires the same ACLs as the current producer transaction API:

6.2 Session Sharing

When a TransactionSession is shared between a producer and a share consumer (Use Case 4.2), both entities operate under the same transactionalId and require the same ACLs.

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:

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

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
   server ASF JIRA serverId 5aa69414-a9e9-3523-82ec-879b028fb15b key KAFKA-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

8.2 Integration Tests

8.3 Compatibility Tests

9. Reference 

KIP-98 - Exactly Once Delivery and Transactional Messaging

...

KIP-1309: Improve transaction liveness checking

KIP-1345: Cross-Cluster Atomic Transactions via Global Transaction Coordinator

...