...

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

...

```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  Consumer / Share Consumer 

```

...

The checkpoint execution pseudo code: 

class FlinkKafkaSink {

    // Phase: process()
    TransactionSession session = new TransactionSession(

...

cfg);

...

    session.initialize();

...

    KafkaProducer<K,V> producer = new KafkaProducer<>(

...

cfg, session);

...

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

    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);
        for (SinkRecord record : convertMessages(records)) {
            producer.send(new ProducerRecord<>(outputTopic, record.key(), record.value()));
        }

     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);   // 2. Share consumer acknowledges input within the SAME session
        session.addShareAcksToTransaction(NEW

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

      shareConsumersession.groupMetadatabeginTransaction().groupId(),;
    try {
        for ShareAcknowledgements.fromRecords(records, AcknowledgeType.ACCEPT)var r : records) producer.send(transform(r));
        session.commitTransaction();

   // offsets auto-pulled from  session.commitTransaction();bound consumer
    } catch (ExceptionKafkaException e) {
        session.abortTransaction();    // consumer auto-rewinds
    }
}

Architectural comparison:

...

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.

...

•  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 = share.poll(Duration.ofSeconds(1));
    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

...