DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
| Table of Contents |
|---|
Status
Current state: Under DiscussionAccepted
Discussion thread: here
JIRA:
| Jira | ||||||
|---|---|---|---|---|---|---|
|
...
| Code Block |
|---|
{
"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:
| Code Block |
|---|
// 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.");
}
} |
...