Status

Current state: Under DiscussionAccepted

Discussion thread: here

JIRA:

...

• Legacy TV (TV0 / TV1) : accept markerEpoch >= currentProducerEpoch.

• >=TV2: require markerEpoch > currentProducerEpoch (equality indicates a late or duplicate marker and must be rejected).

...

{
  "apiKey": 27,
  "type": "request",
  "listeners": ["broker"],
  "name": "WriteTxnMarkersRequest",
  "validVersions": "1-2",
  "flexibleVersions": "1+",
  "fields": [
    { "name": "Markers", "type": "[]WritableTxnMarker", "versions": "0+",
      "about": "The transaction markers to be written.", "fields": [
      { "name": "ProducerId", "type": "int64", "versions": "0+",
        "entityType": "producerId", "about": "The current producer ID." },
      { "name": "ProducerEpoch", "type": "int16", "versions": "0+",
        "about": "The current epoch associated with the producer ID." },
      { "name": "TransactionResult", "type": "bool", "versions": "0+",
        "about": "The result (false = ABORT, true = COMMIT)." },
      { "name": "Topics", "type": "[]WritableTxnMarkerTopic", "versions": "0+",
        "about": "Each topic to write markers for.", "fields": [
        { "name": "Name", "type": "string", "versions": "0+",
          "entityType": "topicName", "about": "The topic name." },
        { "name": "PartitionIndexes", "type": "[]int32", "versions": "0+",
          "about": "Partition indexes to write markers for." }
      ]},
      { "name": "CoordinatorEpoch", "type": "int32", "versions": "0+",
        "about": "Epoch of the transaction state partition hosting this coordinator." },
      // --------- NEW FIELD (TV) ADDED BELOW --------
      { "name": "TransactionVersion", "type": "int8", "versions": "2+", "ignorable": true,
        "about": "Transaction version: 0/1 = legacy (TV0/TV1), 2 = TV2.", "default": "0" }
    ]}
  ]
}

Proposed Changes

Coordinator Changes

...

When processing an EndTxnRequest, the coordinator already determines the transaction version (legacy or TV2) and stores it in the transaction’s TransactionMetadata.

...

With this KIP, the coordinator will

...

propagate the transaction version to partition leaders by including a new TransactionVersion field

...

when building a WriteTxnMarkersRequest. This field is only included if the target broker supports request version 2; otherwise, the coordinator falls back to version 1, which omits the field. This allows leaders to apply the correct epoch validation rule while maintaining compatibility across mixed-version clusters.

Broker/Leader Changes

Leaders must apply strict validation when TV is known to be >=TV2:

// Pseudocode inside appendEndTxnMarker() or where checkProducerEpoch is called:

short current = updatedEntry.producerEpoch();
short marker = markerProducerEpoch;          
// Check if TransactionVersion field is available (version 2+)
int txnVersion = (request.version >= 2) ? request.transactionVersion() : 1;  

if (txnVersion >= 2) {
    // TV2: coordinator bumps epoch before marker; duplicates carry old epoch.
    // Accept only strictly greater epoch.
    if (marker <= current) {
        throw new InvalidProducerEpochException("Reject late/dup TV2 marker: " +
            "markerEpoch=" + marker + " <= currentEpoch=" + current);
    }
} else {
    // Legacy behavior
    if (marker < current) {
        throw new InvalidProducerEpochException("Marker epoch < current.");
    }
}

...

Clients require no changes. They continue to issue EndTxnRequest as before. With brokers that support this KIP, leaders apply version-aware validation: the legacy rule for TV0/TV1 and the stricter rule for versions greater than or equal to TV2 . This ensures stronger exactly-once guarantees without altering client behavior.

...

• TV0 / TV1 (value = 0 or 1) →                      →   markerEpoch >= currentProducerEpoch (legacy validation)

• TV2 and future TV's (value >= 2)       →               → markerEpoch > currentProducerEpoch (strict validation)

...

If a coordinator supports version 2 but a partition leader only supports version 1, the coordinator will automatically fall back to sending a version 1 request. This ensures smooth interoperability and maintains backward compatibility.

New Transaction Versions

Future transaction versions will also be subjected to a strict validation by default. If a newer version ever changes or removes the epoch bump behavior introduced in TV2, the broker logic should be updated accordingly. Applying strict validation to all versions ≥ TV2 keeps the behavior forward compatible and ensures the issue remains permanently resolved.

Test Plan

Integration testing will be done to test the various writeTxnMarker request scenarios like testing behavior when stale markers arrive with both TV1 and TV2.

...