...

• producer.send("output-A", record)
  • Source Component: TransactionManager (Client)
  • Action: Sends AddPartitionsToTxnRequest(["output-A-0"]).
  • State Change: Updates local state to TransactionMetadata.topicPartitions = {output-A-0}.
• producer.sendShareAcknowledgementsToTransaction(acks, meta)
  • Source Component: TransactionManager (Client)  ---→  KafkaApis (Broker Core)
  • Action: Sends TxnShareAcknowledgeRequest to Broker 5 (the SharePartition leader - lets say it is broker 5).
  • Broker Execution: KafkaApis intercepts the request and performs two background operations:
    1. Resolves the state partition: shareCoordinatorPartition = (__share_group_state, partitionFor(groupId)).
    2. First call txnCoordinator.handleAddPartitionsToTransaction(transactionalId, producerId, producerEpoch, Set.of(shareCoordinatorPartition), callback, TV_2, requestLocal) 
       1. Then callback → Calls txnCoordinator.handleAddPartitionsToTransaction to dynamically register __share_group_state-N in __transaction_state.
       2. persists to __transaction_state via TransactionLogValue: topicPartitions = {output-A-0, __share_group_state-N}
  • State Change: TransactionMetadata now tracks both topicPartitions = {output-A-0, __share_group_state-N}.
  • Staging Phase: Only on success, SharePartitionManager stages the transactional acknowledgements, transitioning the internal InFlightState to TX_PENDING.
• producer.commitTransaction()
  • Source Component: Client  -------> TransactionCoordinator(Broker Core)
  • Action: Sends EndTxnRequest(COMMIT).
  • Execution: TxnCoord reads the topic partitions from state and resolves the physical leaders via MetadataCache:
    • output-A-0 --------> Broker 3
    • __share_group_state-N  -----→ Broker 5
  • Action: Dispatches WriteTxnMarkers to both brokers.
• Each broker handles the transaction markers
  • Execution on Broker 3 (Output Side):
    • Appends the COMMIT control batch to the output-A-0 log.
    • KIP-1289 hook runs (no-op since no TX_PENDING here)
  • Execution on Broker 5 (Share Side):
    • Appends the COMMIT control batch to the __share_group_state-N log.
    • Fires the KIP-1289 hook: Triggers sharePartitionManager.applyTxnMarker(COMMIT).
    • Scans the internal partitionCache, finds the matching TX_PENDING flight states, and commits them to ACKNOWLEDGED.
• Transaction Completion
  • Action: Both brokers return successful ACKs back to TxnCoord.
  • State Change: TxnCoord transitions the state to COMPLETE_COMMIT.
  • Result: The client-side blocking producer.commitTransaction() call returns successfully.

...

// Kafka-as-destination path (true EOS via TxnShareAcknowledge):
batch boundary (or per-record, depending on tx granularity):
    producer.beginTransaction()
    for each (record, output) in batch:
        producer.send(destinationTopic, output)
    producer.sendShareAcknowledgementsToTransaction(
        ackMap,                                  // record→ACCEPT for processed records
        shareConsumer.shareGroupMetadata()
    )
    producer.commitTransaction()
    // No separate consumer.acknowledge() — the ACK is in the transaction.

-------------------

// Pure-external-sink path (at-least-once + idempotent destination):[this KIP is not changin anything of this flow]
worker thread per record:
    output = process(record)
    externalSink.write(output, idempotencyKey)        // user task does this
    consumer.acknowledgeAsync(record, ACCEPT)
        .thenAccept(_ -> markComplete(record))

...

Corner cases 

Here are your corner cases formatted using the same clean, un-phased, human-style layout. The wording, titles, technical specifications, and internal error codes remain exactly as provided.

1. Broker Crash During TX_PENDING (in-memory state)

• Scenario: broker stages TX_PENDING then crashes before marker arrives.
• Behavior: TX_PENDING is in-memory only (__share_group_state schema unchanged in v1); records replay as ACQUIRED/AVAILABLE after broker restart and are redelivered.
• Guarantee: no data loss; at-least-once on broker-crash window (existing share-group baseline).
• Future: persistent TX_PENDING via extending ShareSnapshotValue.DeliveryState deferred to a follow-up KIP.

2. TxnCoord Registration Fails (COORDINATOR_NOT_AVAILABLE, NOT_COORDINATOR)

• Scenario: broker-internal AddPartitionsToTxn returns retriable error.
• Behavior: broker does NOT stage; returns error to producer; producer enters abortable state.
• Recovery: user catches KafkaException → abortTransaction() → retries.

3. Producer Fenced Mid-Stage (PRODUCER_FENCED, INVALID_PRODUCER_EPOCH)

• Scenario: zombie producer A (epoch 1) sends TxnShareAcknowledge after producer B (epoch 2) has been initialized.
• Behavior: TxnCoord registration rejects with fencing error → no staging happens → producer A becomes fatal.
• Late-arriving marker safety: per-InFlightState (stagedProducerId, stagedProducerEpoch) filter rejects mismatched markers; idempotent.

5. Duplicate WriteTxnMarkers Delivery

• Scenario: TxnCoord retries marker after network failure; broker receives same marker twice.
• Behavior: second invocation finds state != TX_PENDING (already resolved) → no-op.
• Guarantee: idempotent by state filter — verified in InFlightStateTxnTest.

6. Acquisition Lock Expiry While TX_PENDING

• Scenario: transaction stages records but stalls; lock duration expires before commit/abort.
• Behavior: lock timer detects TX_PENDING with expired lock → reverts to ACQUIRED → re-acquirable; subsequent marker for the stale staging is filtered out (epoch mismatch).
• Configuration: group.share.record.lock.duration.ms (default 30s; per share group).

7. Transaction Timeout While TX_PENDING (transaction.timeout.ms)

• Scenario: producer crashes after staging; TxnCoord detects abandoned transaction via timeout.
• Behavior: TxnCoord auto-sends WriteTxnMarkers(ABORT); broker reverts TX_PENDING → ACQUIRED → AVAILABLE on next lock cycle. No data loss; eventual redelivery.

8. Stale Member Epoch (STALE_MEMBER_EPOCH, UNKNOWN_MEMBER_ID)

• Scenario: rebalance changes memberEpoch between shareGroupMetadata() snapshot and sendShareAcksToTransaction() call.
• Behavior: broker compares request memberEpoch with current SharePartition owner; mismatch → returns STALE_MEMBER_EPOCH; no staging.
• Recovery: abortable; user retries the entire read-process-write loop with a fresh snapshot.

9. Mixed-Version Cluster During Rolling Upgrade

• Scenario: some brokers lack apiKey 93 (TxnShareAcknowledge).
• Behavior: producer probes ApiVersions; if apiKey 93 absent on any broker → UnsupportedVersionException thrown synchronously, no bytes on wire.
• Recovery: user must abort the transaction (output-side AddPartitionsToTxn still needs cleanup).
• Gating: share.version=2 finalized feature flag controls broker advertisement.

10. Invalid Acknowledge Type Inside Transaction

• Scenario: producer sends AcknowledgeType.RELEASE (2), RENEW (4), or GAP (0) inside sendShareAcksToTransaction.
• Behavior: broker rejects with INVALID_RECORD_STATE; no staging.
• Allowed values: only ACCEPT (1) and REJECT (3) — release/renew/gap have no transactional semantics.

11. Partial Multi-Partition Stage Failure

• Scenario: producer stages on partitions A and B; A succeeds, B fails.
• Behavior: per-partition 2PC rollback — already-staged records on A are reverted to ACQUIRED before the response returns (commit 8 on the branch).
• Result: all-or-nothing per call; user sees error → aborts → retries.
• Guarantee: no partial commit; clean rollback semantics.

12. Leader Change for __share_group_state-N Between Stage and Commit

• Scenario: share-coordinator partition leadership moves from broker 5 to broker 7 between staging and WriteTxnMarkers dispatch.
• Behavior: TxnCoord re-resolves leader at commit time via MetadataCache → marker goes to broker 7; broker 7's in-memory partitionCache does NOT contain the original TX_PENDING state.
• Limitation in this intial version: TX_PENDING lost on share-coordinator leadership change (same root cause as Corner Case 1); lock-timeout reverts records; redelivery.
• Future: persistent TX_PENDING closes this case as well.

13. Transaction Aborted Without Markers Reaching Broker

• Scenario: transaction aborts but broker hosting TX_PENDING is partitioned from TxnCoord.
• Behavior: lock timeout (Corner Case 6) eventually reverts; subsequent late-arriving marker is no-op (state already AVAILABLE or epoch mismatched).
• Convergence guaranteed.

14. Concurrent Transactions From Same Producer (Different Epochs)

• Scenario: producer rapidly bumps epoch; old in-flight TxnShareAcknowledge arrives with stale epoch.
• Behavior: TxnCoord rejects with PRODUCER_FENCED; broker never stages. Per-record fencing on the state machine prevents any stale marker from resolving newer staging.

...

Note: We can persist the record state and get rid of case 1 in follow up KIP.

 Compatibility, Deprecation, and Migration Plan

• Zero Breaking Changes: Current implicit and explicit modes remain the default and continue to function as-is.

• Opt-in Requirement: Users must explicitly enable the new mode.

...

Test Plan

The verification strategy focuses on state machine integrity and fault tolerance under high-concurrency and failure scenarios.

...