Page History

...

Step 1: Coordinator sends BeginBatchTxn(id) to Broker
Step 2: Executor-1 calls poll() to Broker - gets records
Step 3: Executor-2 calls poll() to Broker - gets records
Step 4: Executor-1 sends ack(txnId, records) to Broker - records become LOCKED
Step 5: Executor-2 sends ack(txnId, records) to Broker - records become LOCKED
Step 6: Coordinator sends PrepareBatchTxn(id) to Broker - state becomes PREPARE
Step 7: Broker responds "prepared" to Coordinator
Step 8: Coordinator sends CommitBatchTxn(id) to Broker - LOCKED records become ACKNOWLEDGED
Step 9: Broker responds "committed" to Coordinator

...

```
{groupId}-{transactionalIdPrefix}-{checkpointId}

Example: my-share-group-flink-job-42
```

4. State Storage

Extend `__share_group_state` topic with transaction records:

```java
public class ShareTransactionState {
    String transactionId;
    String groupId;
    TransactionState state;  // ONGOING, PREPARE, COMMITTED, ABORTED
    long createTimeMs;
    long prepareTimeMs;
    Map<TopicPartition, List<AcknowledgementBatch>> pendingAcks;
}
```

5. Failure Handling

Timeout Behavior

- ONGOING: Abort → Release records

- PREPARE: Coordinator decides (commit/abort)

- COMMITTED/ABORTED: Cleanup after retention

Recovery Protocol

```java
// On coordinator startup
List<PendingTransaction> pending = coordinator.listPendingTransactions(groupId);
for (PendingTransaction txn : pending) {
    if (txn.state == PREPARE && shouldCommit(txn)) {
        coordinator.commitTransaction(txn.id);  // Complete Phase 2
    } else {
        coordinator.abortTransaction(txn.id);   // Rollback
    }
}
```

Idempotency

...

This KIP reuses Kafka's existing two-phase commit protocol:

• EndTxn(COMMIT) transitions the transaction to PREPARE_COMMIT in __transaction_state
• WriteTxnMarkers dispatches commit markers to all partitions (data topics, __consumer_offsets, and now __share_group_state)
• When all markers are confirmed, state transitions to COMPLETE_COMMIT

6. Wire Protocol Details

TxnShareAcknowledgeRequest

...