...

Internal Topic Record Schemas

__transaction_state:

...

unchanged

share-partition participation in transactions is tracked via the existing TransactionMetadata.topicPartitions field

(mechanism: broker-internal AddPartitionsToTxn triggered by TxnShareAcknowledge handler, mirroring how __consumer_offsets-N is registered via AddOffsetsToTxn for regular consumer groups)

Flow — Regular Consumer Group 

...

[Note: Only TV 2 supporting - broker-side auto-registration and proper epoch fencing, do not want to maintain TV1 code. TV2 is already default option since 4.0]

State machine additions:

__share_group_state schema unchanged 

 no ShareCoordinatorShard changes; no snapshot replay changes; no compatibility risk.

Inter-broker protocol unchanged

 existing handleAddPartitionsToTransaction accepts arbitrary TopicPartition; no new code path on TxnCoord side

State machine additions:

• New transient state TX_PENDING 
• On WriteTxnMarkers commit: TX_PENDING(ACCEPT)
• New transient state TX_PENDING 
• On WriteTxnMarkers commit: TX_PENDING(ACCEPT) → ACKNOWLEDGED; same for RELEASE and REJECT.
• On WriteTxnMarkers abort: TX_PENDING(*) → back to ACQUIRED (lock continues; consumer can retry the work).

Pseudo code

// KafkaOne-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,   time setup
shareConsumer.subscribe(List.of("source-topic"));
producer.initTransactions();   // exactly once per producer instance

// CTP loop
while (running) {
    ConsumerRecords<K, V> records = shareConsumer.poll(Duration.ofSeconds(5));
    if (records.isEmpty()) continue;

    try {
        producer.beginTransaction();

        Map<TopicIdPartition, Acknowledgements> acks = new HashMap<>();

        for (ConsumerRecord<K, //V> record→ACCEPTrecord for: processedrecords) records{
        shareConsumer.shareGroupMetadata()
    )
V output   producer.commitTransaction()
= process(record);    // 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)producer.send(new ProducerRecord<>("destination-topic", output));    
    // user task does this
    consumeracks.acknowledgeAsync(record, ACCEPT)
computeIfAbsent(
                   .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.

...

 new TopicIdPartition(record.topicId(), 
                            new TopicPartition(record.topic(), record.partition())),
                    k -> Acknowledgements.empty()
                ).add(record.offset(), AcknowledgeType.ACCEPT);                  // accumulate
        }

        // Compress acks to AcknowledgementBatch wire form and stage in the txn.
        producer.sendShareAcknowledgementsToTransaction(
            toBatches(acks),                                                      // Map<TopicIdPartition, List<AcknowledgementBatch>>
            shareConsumer.shareGroupMetadata()                                    
        );

        producer.commitTransaction();    // BLOCKS — true synchronisation point
        // Records are now both produced AND acknowledged ATOMICALLY.
        // Do NOT call shareConsumer.acknowledge(...) on these records.
		//    if called mistakenly then non-transactional acknowledge is rejected (TX_PENDING to ACKNOWLEDGED - InvalidRecordStateException); the transactional path works.

    } catch (ProducerFencedException | UnsupportedVersionException fatal) {
        throw fatal;                     // unrecoverable — kill the process
    } catch (KafkaException abortable) {
        producer.abortTransaction();     // BLOCKS — txn reverted; staged records revert to ACQUIRED on broker
        // Records will be re-delivered on next poll(); retry naturally.
    }
}

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.

Metrics

All new metrics follow the existing Kafka metric conventions (Yammer for broker JMX, KafkaMetric for client-side).

Metric names mirror the established kafka.server:type=group-coordinator-metrics,name=... and kafka.server:type=share-coordinator-metrics,name=... patterns.

Backward-compatible: no existing metrics renamed or removed.

Compatibility, Deprecation, and Migration Plan

Enablement and Rollout Plan

Feature Gating

opt-in, online-upgradable, and zero-disruption for existing workloads - No new feature flag introduced. Uses existing kafka-features.sh machinery.

Rolling Upgrade (Online; No Traffic Disruption)

Phase 1 — Software upgrade (rolling)
  Upgrade brokers one at a time to the binary containing KIP-1289.
  KIP-1289 code is dormant (share.version still 1).
  Standard Kafka rolling-restart

Phase 2 — Soak timing
  Verify cluster health

Phase 3 — Finalise feature
  kafka-features.sh upgrade --feature share.version --version 2
  - Online metadata propagation (update of  __cluster_metadata)
  - No restart, no rebalance, no leader change, no socket disruption
  - Brokers begin advertising apiKey 93

Phase 4 — Client onboarding
  Roll out applications calling sendShareAcknowledgementsToTransaction.
  Add the alerts (esp. TxnPendingRecordsCount) and relevant metrics in dashboard & observability.

Cross-Version Client/Broker Matrix

Downgrade

1. Drain in-flight TX_PENDING:
   - Halt producers calling sendShareAcknowledgementsToTransaction
   - Wait for transaction.timeout.ms (default 60s) for any abandoned txns to clear
2. kafka-features.sh downgrade --feature share.version --version 1
Broker rejects the downgrade RPC if any in-memory TX_PENDING exists, preventing data inconsistency.
Software downgrade (binary): must follow feature downgrade; standard rolling restart.

Test Plan

Rough implementation [The actual implementation will be phasewise with multiple smaller PRs]:
    https://github.com/apache/kafka/pull/22357

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

• Unit Tests: Validates state transitions (e.g., ACQUIRED to ACKNOWLEDGED on commit vs. AVAILABLE on abort), idempotency of operations, and transaction timeout/auto-abort logic.

• Integration Tests: Focuses on end-to-end commit/abort flows, coordinator recovery, and multi-consumer behavior within a single group during network partitions.

• System & Performance Tests: Benchmarks transactional vs. non-transactional modes and verifies exactly-once delivery.Chaos Tests: Simulates broker and coordinator crashes specifically during critical phases like PREPARE_COMMIT to ensure protocol durability. non-transactional modes and verifies exactly-once delivery.

• Chaos Tests: Simulates broker and coordinator crashes specifically during critical phases like PREPARE_COMMIT to ensure protocol durability.

Follow-up KIP (deferred)

• Persistent TX_PENDING in __share_group_state for crash-resilient EOS on broker failover during staging (addresses Corner Cases).
• Per-share-partition metric for TX_PENDING residency time— would add a histogram of "time spent in TX_PENDING" useful for diagnosing slow producers; can be added in the persistence KIP without compatibility concerns.
• External processing engines or database as 2PC participants for write/sink records.

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.