Page History

...

It decouples lifecycle management from record production.

```java```

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

```

2.2 Configuration

TransactionSession accepts a subset of existing producer configs plus one new config:

...

No new broker-side configs. No new wire protocol RPCs. TransactionSession sends the same RPCs that KafkaProducer sends today:

 FindCoordinator, InitProducerId, AddPartitionsToTxn, EndTxn, AddOffsetsToTxn, TxnOffsetCommit, TxnHeartbeat (KIP 1309).

...

KafkaProducer gains a new constructor and method to accept an external TransactionSession:

```java```

public class KafkaProducer<K, V> {
      /**
        * Create aBind producer that usesto an externally-managed transactionexternal session. 
     * Lifecycle  * The producer does NOT own the transaction lifecycle.
     * It uses the session's producerId/epoch for idempotent writes.
     * beginTransaction(), commitTransaction(), abortTransaction()
     * throw IllegalStateException -- use the TransactionSession directly.
     */
    methods (e.g., beginTransaction) throw IllegalStateException.
     */
    public KafkaProducer(Map<String, Object> configs, TransactionSession session);

      /**
        * GetAccess the transaction session for this producer.
     * Returns null if the producer is not transactional.
     * Returns the internal TransactionSession if transactional.id is configured.
     * Returns the external TransactionSession if constructed with one.
     */
   session (internal or external). Returns null if non-transactional.
     */
    public TransactionSession transactionSession();
}

```

Backward compatibility: When KafkaProducer is constructed Compatibility: Constructing a producer with transactional.id in the config (the current pattern), it internally creates a TransactionSession and delegates to it. The existing initTransactions(), beginTransaction(), commitTransaction(), abortTransaction(), sendOffsetsToTransaction() methods continue to work unchanged. They are convenience wrappers around the internal TransactionSessionworks as it does today.

It will internally create a TransactionSession and use its existing methods (initTransactions, beginTransaction, etc.) as convenience wrappers.

No code changes are required for existing users.

2.4 Integration with KafkaShareConsumer

...