...

JIRA:

Motivation

Late-arriving or duplicate EndTransaction markers have historically been a correctness risk, as partition leaders could not reliably distinguish them from markers belonging to the active transaction. With TV2’s epoch-bump contract, we now have the ability to close this gap and strengthen exactly-once guarantees and this KIP aims to do just that.

The introduction of KIP-890 brought a protocol change with Transaction Version 2 (TV2), where the transaction coordinator always increments the producer epoch by one (+1) before writing the final transaction marker (commit or abort). This change tightened the contract between the coordinator and partition leaders: a valid TV2 marker must have an epoch strictly greater than the producer’s current epoch at the leader.

To enforce this contract, the marker must be validated at the time it is written to the partition logs. When a client issues an EndTxnRequest, the coordinator determines the outcome (commit or abort) and then sends WriteTxnMarkersRequest messages to all partition leaders involved in the transaction. Upon receiving these requests, each leader performs the validation during the log append step, comparing the marker’s epoch against its local producer state before appending the EndTxn control record. Currently, the check looks like this:

// ProducerAppendInfo.java – appendEndTxnMarker()
private void checkProducerEpoch(short producerEpoch, long offset) {
    if (producerEpoch < updatedEntry.producerEpoch())
        throw new InvalidProducerEpochException(...);
}

This check accepts markers when producerEpoch >= currentProducerEpoch. Under legacy transaction versions (TV0 and TV1) this was the expected behavior: EndTxn markers were written with the same epoch as the transactional records, so equality matched the intended case. However, it also created a correctness gap. Because the coordinator did not bump the epoch at EndTxn time, leaders could not distinguish between a valid marker and a late or duplicate one. If a duplicate marker arrived after a new transaction had already begun with the same epoch, the leader would treat it as valid and could mistakenly commit or abort records from the newer transaction. This threatens our EOS guarantees.

In other words, late-arriving markers have always been a potential problem in TV0/TV1 — the system simply lacked the means to distinguish between them.

With TV2, the semantics change. The coordinator always bumps the producer epoch before sending the final marker, establishing a clear invariant: a valid EndTxn marker must have producerEpoch == current + 1 at the leader. Any marker arriving with equality (producerEpoch == current) can now be safely identified as late or duplicate and rejected. This closes a long-standing gap in transaction handling and prevents scenarios where multiple transactions could be conflated under the same epoch.

The solution requires that leaders know which transaction version applies to each marker. Today they do not — the same relaxed check is applied universally. The core of this KIP is therefore to make the transaction version explicit in the WriteTxnMarkersRequest, enabling leaders to enforce the correct validation rule:

• Legacy TV: accept markerEpoch >= currentProducerEpoch.

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

This approach preserves the legacy behavior and backward compatibility, while ensuring that TV2 leaders enforce the intended epoch invariant and strengthen exactly-once guarantees.

Public Interfaces

We will add a new tagged field, TransactionVersion, to the WriteTxnMarkersRequest so that the transaction coordinator can pass version information to the partition leaders. This approach uses the tagged field mechanism, which allows optional data to be attached to existing message versions without requiring a protocol version bump.

Since WriteTxnMarkersRequest already supports flexible versions (version 1+), we can add the new field as a tagged field. This ensures backward compatibility: older brokers will simply ignore the tagged field and assume legacy behavior, while newer brokers can read the field and apply the appropriate validation logic. The new transaction coordinator will always set this field, and if a new broker receives a request from an old transaction coordinator where the field is not present, it will assume legacy behavior by default.

WriteTxnMarkersRequest (apiKey: 27)

{
  "apiKey": 27,
  "type": "request",
  "listeners": ["broker"],
  "name": "WriteTxnMarkersRequest",
  "validVersions": "1",
  "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 TAGGED FIELD (TV) ADDED BELOW --------
      { "name": "TransactionVersion", "type": "int8", "versions": "1+", "taggedVersions": "1+", "tag": 0,
  "about": "Transaction version. Examples are value 1 = TV1, 2 = TV2, etc", "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.

As part of this KIP, the coordinator will now propagate the transaction version to partition leaders by including a new TransactionVersion field in the WriteTxnMarkersRequest. This ensures that leaders can apply the correct producer epoch validation rule depending on whether the transaction uses the legacy protocol or TV2.

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();      // Producer state epoch at leader
short marker = markerProducerEpoch;                // From request
byte txnVersion = request.hasTransactionVersion() ? request.transactionVersion() : TV_1;

if (txnVersion >= TV_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.");
    }
}

...

Compatibility, Deprecation, and Migration Plan

Clients

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 TV2. This ensures stronger exactly-once guarantees without altering client behavior.

New Brokers

New brokers will support reading the tagged TransactionVersion field in WriteTxnMarkersRequest. When processing a request, brokers apply the appropriate validation rule based on the field value:

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

• TV2 (value = 2) → markerEpoch > currentProducerEpoch (new strict validation)

• Missing tagged field → Default to legacy validation

This allows leaders to distinguish between transaction versions and enforce the correct epoch validation, ensuring stronger exactly-once guarantees for TV2 transactions while maintaining backward compatibility.

Old Brokers

Old brokers that do not support tagged fields will continue to apply the legacy check (markerEpoch >= currentProducerEpoch) for all requests. This is sufficient for legacy transactions, but does not close the gap for TV2 transactions. The tagged field is simply ignored, ensuring no compatibility issues.

Test Plan

Integration testing will be done to test the various writeTxnMarker request scenarios described above.

Rejected Alternatives

1. Depend on VerificationStateEntry.supportsEpochBump(): This method indicates whether an epoch bump has occurred, and in theory could help leaders distinguish between TV1 and TV2. However, its state is cleared after the first control record is written. This leaves a protection gap for late or duplicate markers, which may still arrive after the state has been reset. Because it is not reliable as a persistent signal of the transaction version, it cannot help in enforcing stricter validation.

2. Store transaction version in producer state entry: Enables explicit version tracking but adds storage overhead and requires bumping the record format, increasing complexity for little gain.