...

Kafka’s transactional model enables exactly-once semantics (EOS) by having the coordinator manage commits and aborts, while partition leaders apply them through control records. However, late or duplicate EndTransaction markers have historically posed a correctness risk, as leaders could not reliably distinguish them from markers belonging to the active transaction. With Transaction Version 2 (TV2) and its epoch-bump contract, we can finally close this long-standing gap and reinforce EOS exactly-once guarantees.

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 the partition leaders involved in the transaction. Upon receiving these requests, each leader performs a 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:

...

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

...

bump the WriteTxnMarkersRequest API to version 2 and introduce a new

...

field, TransactionVersion,

...

which will enable the transaction coordinator

...

to pass version information to the partition leaders

...

.

...

WriteTxnMarkersRequest (apiKey: 27)

{
  "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 TAGGED FIELD (TV) ADDED BELOW --------
      { "name": "TransactionVersion", "type": "int8", "versions": "1+", "taggedVersions": "1+", "tag": 02+",
        "about": "Transaction version: 0/1 = legacy (TV0/TV1), 2 = TV2.", "default": "0" }
    ]}
  ]
}

...

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

short current = updatedEntry.producerEpoch();      // Producer state epoch at leader
short marker = markerProducerEpoch;          
// Check if TransactionVersion field is //available From request
byte(version 2+)
int txnVersion = (request.hasTransactionVersion(version >= 2) ? 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.");
    }
}

...

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

...

WriteTxnMarkersRequest v2, which includes the

...

TransactionVersion field

...

.

When processing

...

requests, brokers will apply the appropriate validation rule based on the field value:

• TV0 / TV1 (value = 0

...

• or 1) → markerEpoch >= currentProducerEpoch (legacy validation)

• TV2 (value = 2)

...

•                    → markerEpoch > currentProducerEpoch (

...

• strict validation)

• Missing tagged field → Default to legacy validation

...

Version negotiation between brokers happens through the standard ApiVersionsRequest handshake. The transaction coordinator sends a WriteTxnMarkersRequest v2 only when both the coordinator and the target partition leader support version 2. This maintains compatibility in mixed-version clusters and ensures that stricter validation is applied only when both sides support the new protocol.

Old Brokers

Old brokers that

...

only support WriteTxnMarkersRequest v1 will continue to

...

use the legacy

...

validation rule (markerEpoch >= currentProducerEpoch) for all requests. This

...

behavior remains correct for legacy transactions

...

but does not close the gap for TV2

...

.

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.

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.

...

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 record batch 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.

3. Use a tagged field in the WriteTxnMarkersRequest :