You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "rreddy-22 (via GitHub)" <gi...@apache.org> on 2025/11/12 18:17:13 UTC
[PR] KIP-1228: Add Transaction Version to request builder [kafka]
rreddy-22 opened a new pull request, #20868:
URL: https://github.com/apache/kafka/pull/20868
In this patch, we add a new WriteTxnMarkersRequest builder constructor that accepts
transaction versions for each marker. The transaction version is already stored in
the metadata on the transaction coordinator. We now extract this version
information and pass it to the builder, which propagates it to partition leaders.
Since the TransactionVersion field is marked as ignorable, we always include it when
building the request. The field will be automatically omitted without any errors, during serialization
if the request version is < 2, ensuring backward compatibility with brokers that
only support version 1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520452354
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -369,10 +382,23 @@ class TransactionMarkerChannelManagerTest {
assertEquals(1, channelManager.queueForUnknownBroker.totalNumMarkers(txnTopicPartition2))
val expectedBroker1Request = new WriteTxnMarkersRequest.Builder(
- util.List.of(new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
- new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)))).build()
+ util.List.of(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1))
+ ),
+ util.List.of(
+ TransactionVersion.TV_2.featureLevel().toByte,
Review Comment:
Should we have any tests with two different versions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19446: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2525258135
##########
clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java:
##########
@@ -79,4 +110,55 @@ public void testGetErrorResponse() {
assertEquals(0, errorResponse.throttleTimeMs());
}
}
+
+ @Test
+ public void testTransactionVersion() {
+ // Test that TransactionVersion is set correctly and serialization handles it properly.
+ List<WriteTxnMarkersRequest.TxnMarkerEntry> markersWithVersion = Collections.singletonList(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(
+ PRODUCER_ID, PRODUCER_EPOCH, COORDINATOR_EPOCH,
+ RESULT, Collections.singletonList(TOPIC_PARTITION), (short) 2)
+ );
+ WriteTxnMarkersRequest.Builder builder = new WriteTxnMarkersRequest.Builder(markersWithVersion);
+
+ // Test version 2 - TransactionVersion should be included.
+ WriteTxnMarkersRequest requestV2 = builder.build((short) 2);
+ assertNotNull(requestV2);
+ assertEquals(1, requestV2.markers().size());
+ // Verify TransactionVersion is set to 2 in the request data.
+ assertEquals((byte) 2, requestV2.data().markers().get(0).transactionVersion());
+ // Verify the request can be serialized for version 2 (TransactionVersion field included).
+ // This should not throw an exception.
+ requestV2.serialize();
+ int sizeV2 = requestV2.sizeInBytes();
+ // Verify TransactionVersion is set to 2 in the marker entry (markers() reads the field for version >= 2).
+ // This is what the broker sees when receiving the request - markers() is called on the partition leader side.
+ assertEquals((short) 2, requestV2.markers().get(0).transactionVersion());
+
+ // Test version 1 - TransactionVersion should be omitted (ignorable field).
+ WriteTxnMarkersRequest requestV1 = builder.build((short) 1);
+ assertNotNull(requestV1);
+ assertEquals(1, requestV1.markers().size());
+ // Verify TransactionVersion is still set to 2 in the request data (even for version 1).
+ // This is what the coordinator has when building the request - data() is used before serialization.
+ // The field value is preserved in the data, but will be omitted during serialization.
+ assertEquals((byte) 2, requestV1.data().markers().get(0).transactionVersion());
+ // Verify the request can be serialized for version 1 (TransactionVersion field omitted).
+ // This should not throw an exception even though TransactionVersion is set to 2
+ // because the field is marked as ignorable.
+ requestV1.serialize();
+ int sizeV1 = requestV1.sizeInBytes();
+ // After serialization, verify TransactionVersion is 0 in the marker entry for version 1 (field not read for version < 2).
+ // This is what the broker sees when receiving the request - markers() is called on the partition leader side.
+ assertEquals((short) 0, requestV1.markers().get(0).transactionVersion());
+
+ // Verify that version 2 is larger than version 1 because it includes TransactionVersion field.
+ // TransactionVersion is int8 (1 byte), so version 2 should be at least 1 byte larger.
+ // This check ensures that the serialization logic correctly includes/excludes the field.
+ assertTrue(sizeV2 > sizeV1,
Review Comment:
done!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520497403
##########
clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java:
##########
@@ -79,4 +102,47 @@ public void testGetErrorResponse() {
assertEquals(0, errorResponse.throttleTimeMs());
}
}
+
+ @Test
+ public void testTransactionVersion() {
Review Comment:
As discussed offline, I will add a test to check the deserialization and response handling in a follow up PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520441399
##########
clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java:
##########
@@ -79,4 +102,47 @@ public void testGetErrorResponse() {
assertEquals(0, errorResponse.throttleTimeMs());
}
}
+
+ @Test
+ public void testTransactionVersion() {
Review Comment:
Should we also have a test where we get the value from an older version?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2522415016
##########
clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java:
##########
@@ -79,4 +102,47 @@ public void testGetErrorResponse() {
assertEquals(0, errorResponse.throttleTimeMs());
}
}
+
+ @Test
+ public void testTransactionVersion() {
Review Comment:
Added a test to verify if the right value us extracted
##########
clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java:
##########
@@ -79,4 +102,47 @@ public void testGetErrorResponse() {
assertEquals(0, errorResponse.throttleTimeMs());
}
}
+
+ @Test
+ public void testTransactionVersion() {
Review Comment:
Added a test to verify if the right value is extracted
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520500063
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -260,7 +260,19 @@ class TransactionMarkerChannelManager(
}.filter { case (_, entries) => !entries.isEmpty }.map { case (node, entries) =>
val markersToSend = entries.asScala.map(_.txnMarkerEntry).asJava
val requestCompletionHandler = new TransactionMarkerRequestCompletionHandler(node.id, txnStateManager, this, entries)
- val request = new WriteTxnMarkersRequest.Builder(markersToSend)
+
+ // Always extract transaction versions from entries (one per marker, preserving order)
+ val transactionVersions = new util.ArrayList[java.lang.Byte]()
+ entries.asScala.foreach { entry =>
+ val transactionVersion = entry.pendingCompleteTxn.txnMetadata.clientTransactionVersion()
+ if (transactionVersion != null) {
Review Comment:
Not really, but I just wanted to be safe jic
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520452354
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -369,10 +382,23 @@ class TransactionMarkerChannelManagerTest {
assertEquals(1, channelManager.queueForUnknownBroker.totalNumMarkers(txnTopicPartition2))
val expectedBroker1Request = new WriteTxnMarkersRequest.Builder(
- util.List.of(new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
- new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)))).build()
+ util.List.of(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1))
+ ),
+ util.List.of(
+ TransactionVersion.TV_2.featureLevel().toByte,
Review Comment:
Should we have any tests with two different versions? Maybe not in this file necessarily, but made me think of it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19446: Add Transaction Version to request builder [2/3] [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2524949408
##########
clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java:
##########
@@ -79,4 +110,55 @@ public void testGetErrorResponse() {
assertEquals(0, errorResponse.throttleTimeMs());
}
}
+
+ @Test
+ public void testTransactionVersion() {
+ // Test that TransactionVersion is set correctly and serialization handles it properly.
+ List<WriteTxnMarkersRequest.TxnMarkerEntry> markersWithVersion = Collections.singletonList(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(
+ PRODUCER_ID, PRODUCER_EPOCH, COORDINATOR_EPOCH,
+ RESULT, Collections.singletonList(TOPIC_PARTITION), (short) 2)
+ );
+ WriteTxnMarkersRequest.Builder builder = new WriteTxnMarkersRequest.Builder(markersWithVersion);
+
+ // Test version 2 - TransactionVersion should be included.
+ WriteTxnMarkersRequest requestV2 = builder.build((short) 2);
+ assertNotNull(requestV2);
+ assertEquals(1, requestV2.markers().size());
+ // Verify TransactionVersion is set to 2 in the request data.
+ assertEquals((byte) 2, requestV2.data().markers().get(0).transactionVersion());
+ // Verify the request can be serialized for version 2 (TransactionVersion field included).
+ // This should not throw an exception.
+ requestV2.serialize();
+ int sizeV2 = requestV2.sizeInBytes();
+ // Verify TransactionVersion is set to 2 in the marker entry (markers() reads the field for version >= 2).
+ // This is what the broker sees when receiving the request - markers() is called on the partition leader side.
+ assertEquals((short) 2, requestV2.markers().get(0).transactionVersion());
+
+ // Test version 1 - TransactionVersion should be omitted (ignorable field).
+ WriteTxnMarkersRequest requestV1 = builder.build((short) 1);
+ assertNotNull(requestV1);
+ assertEquals(1, requestV1.markers().size());
+ // Verify TransactionVersion is still set to 2 in the request data (even for version 1).
+ // This is what the coordinator has when building the request - data() is used before serialization.
+ // The field value is preserved in the data, but will be omitted during serialization.
+ assertEquals((byte) 2, requestV1.data().markers().get(0).transactionVersion());
+ // Verify the request can be serialized for version 1 (TransactionVersion field omitted).
+ // This should not throw an exception even though TransactionVersion is set to 2
+ // because the field is marked as ignorable.
+ requestV1.serialize();
+ int sizeV1 = requestV1.sizeInBytes();
+ // After serialization, verify TransactionVersion is 0 in the marker entry for version 1 (field not read for version < 2).
+ // This is what the broker sees when receiving the request - markers() is called on the partition leader side.
+ assertEquals((short) 0, requestV1.markers().get(0).transactionVersion());
+
+ // Verify that version 2 is larger than version 1 because it includes TransactionVersion field.
+ // TransactionVersion is int8 (1 byte), so version 2 should be at least 1 byte larger.
+ // This check ensures that the serialization logic correctly includes/excludes the field.
+ assertTrue(sizeV2 > sizeV1,
Review Comment:
This test makes sense to me but we may want to have something that deserializes too? Just to confirm both work as expected?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520546731
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -369,10 +382,23 @@ class TransactionMarkerChannelManagerTest {
assertEquals(1, channelManager.queueForUnknownBroker.totalNumMarkers(txnTopicPartition2))
val expectedBroker1Request = new WriteTxnMarkersRequest.Builder(
- util.List.of(new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
- new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)))).build()
+ util.List.of(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1))
+ ),
+ util.List.of(
+ TransactionVersion.TV_2.featureLevel().toByte,
Review Comment:
yeah haha, right now there's no test where I found that the outcome could be different for tv1 vs tv2, probably because we're doing all the request tests. On the coordinator side there's no other change in behavior, other than the way we build the requests. I will parameterize tests in the response handling!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520546731
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -369,10 +382,23 @@ class TransactionMarkerChannelManagerTest {
assertEquals(1, channelManager.queueForUnknownBroker.totalNumMarkers(txnTopicPartition2))
val expectedBroker1Request = new WriteTxnMarkersRequest.Builder(
- util.List.of(new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
- new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)))).build()
+ util.List.of(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1))
+ ),
+ util.List.of(
+ TransactionVersion.TV_2.featureLevel().toByte,
Review Comment:
yeah haha, right now there's no test where I found that the outcome could be different for tv1 vs tv2, probably because we're doing all the request tests on the coordinator side and there's no other change in behavior other than the way we build the requests. I will parameterize tests on the response handling!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19446: Add Transaction Version to request builder [2/3] [kafka]
Posted by "firecow (via GitHub)" <gi...@apache.org>.
firecow commented on PR #20868:
URL: https://github.com/apache/kafka/pull/20868#issuecomment-3990186427
We're hitting an issue with this change during a rolling upgrade from 4.1.1 to 4.2.0 in a KRaft cluster (combined mode).
When a 4.2.0 broker's transaction coordinator sends WriteTxnMarkers to a broker still on 4.1.1, it uses v2 of the API. The 4.1.1 broker only supports v1 (validVersions: "1") and rejects it:
org.apache.kafka.common.errors.UnsupportedVersionException: Received request for api with key 27 (WriteTxnMarkers) and unsupported version 2
This fires continuously (~500/sec) for the entire mixed-version window, closing sockets and preventing transaction markers from being written. It stops immediately once the receiving broker is also upgraded to 4.2.0.
The PR states the TransactionVersion field is ignorable and will be omitted when serialized at v1, which is correct — but the version negotiation never falls back to v1 in the first place. The 4.2.0 broker sends v2 despite the target
advertising only v1 in its ApiVersions response.
Cluster state:
- 3-node KRaft, process.roles=[broker, controller]
- metadata.version: 4.0-IV3 (feature level 25)
- transaction.version: 2
- Broker receiving the requests: metadata.version maxSupportedVersion=27 (confirming 4.1.1)
- Broker sending the requests: just restarted on 4.2.0
Rolling upgrades have always worked one broker at a time for us across prior versions. This is a regression in 4.2.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520448976
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala:
##########
@@ -260,7 +260,19 @@ class TransactionMarkerChannelManager(
}.filter { case (_, entries) => !entries.isEmpty }.map { case (node, entries) =>
val markersToSend = entries.asScala.map(_.txnMarkerEntry).asJava
val requestCompletionHandler = new TransactionMarkerRequestCompletionHandler(node.id, txnStateManager, this, entries)
- val request = new WriteTxnMarkersRequest.Builder(markersToSend)
+
+ // Always extract transaction versions from entries (one per marker, preserving order)
+ val transactionVersions = new util.ArrayList[java.lang.Byte]()
+ entries.asScala.foreach { entry =>
+ val transactionVersion = entry.pendingCompleteTxn.txnMetadata.clientTransactionVersion()
+ if (transactionVersion != null) {
Review Comment:
Do we expect this to ever be null?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520546731
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -369,10 +382,23 @@ class TransactionMarkerChannelManagerTest {
assertEquals(1, channelManager.queueForUnknownBroker.totalNumMarkers(txnTopicPartition2))
val expectedBroker1Request = new WriteTxnMarkersRequest.Builder(
- util.List.of(new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
- new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)))).build()
+ util.List.of(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1))
+ ),
+ util.List.of(
+ TransactionVersion.TV_2.featureLevel().toByte,
Review Comment:
yeah haha, right now there's no test where I found that the outcome could be different for tv1 vs tv2, probably because we're doing all the request tests on the coordinator side and there's no other change in behavior here other than the way we build the requests. I will parameterize tests on the response handling!
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -369,10 +382,23 @@ class TransactionMarkerChannelManagerTest {
assertEquals(1, channelManager.queueForUnknownBroker.totalNumMarkers(txnTopicPartition2))
val expectedBroker1Request = new WriteTxnMarkersRequest.Builder(
- util.List.of(new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
- new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)))).build()
+ util.List.of(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1))
+ ),
+ util.List.of(
+ TransactionVersion.TV_2.featureLevel().toByte,
Review Comment:
yeah haha, right now there's no test where I found that the outcome could be different for tv1 vs tv2, probably because we're doing all the request tests on the coordinator side and there's no other change in behavior here other than the way we build the requests. I will parameterize tests in the response handling!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520497403
##########
clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java:
##########
@@ -79,4 +102,47 @@ public void testGetErrorResponse() {
assertEquals(0, errorResponse.throttleTimeMs());
}
}
+
+ @Test
+ public void testTransactionVersion() {
Review Comment:
As discussed offline, I will add a test to check the deserialization and response handling in a follow up PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19446: Add Transaction Version to request builder [2/3] [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2525208994
##########
clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java:
##########
@@ -79,4 +110,55 @@ public void testGetErrorResponse() {
assertEquals(0, errorResponse.throttleTimeMs());
}
}
+
+ @Test
+ public void testTransactionVersion() {
+ // Test that TransactionVersion is set correctly and serialization handles it properly.
+ List<WriteTxnMarkersRequest.TxnMarkerEntry> markersWithVersion = Collections.singletonList(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(
+ PRODUCER_ID, PRODUCER_EPOCH, COORDINATOR_EPOCH,
+ RESULT, Collections.singletonList(TOPIC_PARTITION), (short) 2)
+ );
+ WriteTxnMarkersRequest.Builder builder = new WriteTxnMarkersRequest.Builder(markersWithVersion);
+
+ // Test version 2 - TransactionVersion should be included.
+ WriteTxnMarkersRequest requestV2 = builder.build((short) 2);
+ assertNotNull(requestV2);
+ assertEquals(1, requestV2.markers().size());
+ // Verify TransactionVersion is set to 2 in the request data.
+ assertEquals((byte) 2, requestV2.data().markers().get(0).transactionVersion());
+ // Verify the request can be serialized for version 2 (TransactionVersion field included).
+ // This should not throw an exception.
+ requestV2.serialize();
+ int sizeV2 = requestV2.sizeInBytes();
+ // Verify TransactionVersion is set to 2 in the marker entry (markers() reads the field for version >= 2).
+ // This is what the broker sees when receiving the request - markers() is called on the partition leader side.
+ assertEquals((short) 2, requestV2.markers().get(0).transactionVersion());
+
+ // Test version 1 - TransactionVersion should be omitted (ignorable field).
+ WriteTxnMarkersRequest requestV1 = builder.build((short) 1);
+ assertNotNull(requestV1);
+ assertEquals(1, requestV1.markers().size());
+ // Verify TransactionVersion is still set to 2 in the request data (even for version 1).
+ // This is what the coordinator has when building the request - data() is used before serialization.
+ // The field value is preserved in the data, but will be omitted during serialization.
+ assertEquals((byte) 2, requestV1.data().markers().get(0).transactionVersion());
+ // Verify the request can be serialized for version 1 (TransactionVersion field omitted).
+ // This should not throw an exception even though TransactionVersion is set to 2
+ // because the field is marked as ignorable.
+ requestV1.serialize();
+ int sizeV1 = requestV1.sizeInBytes();
+ // After serialization, verify TransactionVersion is 0 in the marker entry for version 1 (field not read for version < 2).
+ // This is what the broker sees when receiving the request - markers() is called on the partition leader side.
+ assertEquals((short) 0, requestV1.markers().get(0).transactionVersion());
+
+ // Verify that version 2 is larger than version 1 because it includes TransactionVersion field.
+ // TransactionVersion is int8 (1 byte), so version 2 should be at least 1 byte larger.
+ // This check ensures that the serialization logic correctly includes/excludes the field.
+ assertTrue(sizeV2 > sizeV1,
Review Comment:
It's not a round trip though. I was thinking
WriteTxnMarkersRequest.parse(requestV1.serialize(), <version>, context)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520428567
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java:
##########
@@ -2827,7 +2827,7 @@ private EndTxnResponse createEndTxnResponse() {
private WriteTxnMarkersRequest createWriteTxnMarkersRequest(short version) {
List<TopicPartition> partitions = singletonList(new TopicPartition("topic", 73));
WriteTxnMarkersRequest.TxnMarkerEntry txnMarkerEntry = new WriteTxnMarkersRequest.TxnMarkerEntry(21L, (short) 42, 73, TransactionResult.ABORT, partitions);
- return new WriteTxnMarkersRequest.Builder(singletonList(txnMarkerEntry)).build(version);
+ return new WriteTxnMarkersRequest.Builder(singletonList(txnMarkerEntry), null).build(version);
Review Comment:
Do we want to add any tests here with the transactionVersions param? If the tests don't use it, we don't need to, but just want to confirm there isn't any other coverage we want here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520494830
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java:
##########
@@ -2827,7 +2827,7 @@ private EndTxnResponse createEndTxnResponse() {
private WriteTxnMarkersRequest createWriteTxnMarkersRequest(short version) {
List<TopicPartition> partitions = singletonList(new TopicPartition("topic", 73));
WriteTxnMarkersRequest.TxnMarkerEntry txnMarkerEntry = new WriteTxnMarkersRequest.TxnMarkerEntry(21L, (short) 42, 73, TransactionResult.ABORT, partitions);
- return new WriteTxnMarkersRequest.Builder(singletonList(txnMarkerEntry)).build(version);
+ return new WriteTxnMarkersRequest.Builder(singletonList(txnMarkerEntry), null).build(version);
Review Comment:
Yes that's a valid point! I checked the usage of this method and it's only used in the serialization test. When we use null we basically fill up TV with default values so thought it should be okay here. I could add a singleton list with tv0 too if that's better. What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KIP-1228: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2520546731
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala:
##########
@@ -369,10 +382,23 @@ class TransactionMarkerChannelManagerTest {
assertEquals(1, channelManager.queueForUnknownBroker.totalNumMarkers(txnTopicPartition2))
val expectedBroker1Request = new WriteTxnMarkersRequest.Builder(
- util.List.of(new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
- new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)))).build()
+ util.List.of(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId1, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1)),
+ new WriteTxnMarkersRequest.TxnMarkerEntry(producerId2, producerEpoch, coordinatorEpoch, txnResult, util.List.of(partition1))
+ ),
+ util.List.of(
+ TransactionVersion.TV_2.featureLevel().toByte,
Review Comment:
yeah haha, right now there's no test where I found that the outcome could be different for tv1 vs tv2, probably because we're doing all the request tests. On the coordinator side there's no other change in behavior, other than the way we build the requests. I will parameterize tests in the response handling where we expect a diff!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19446: Add Transaction Version to request builder [2/3] [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #20868:
URL: https://github.com/apache/kafka/pull/20868
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19446: Add Transaction Version to request builder [2/3] [kafka]
Posted by "rreddy-22 (via GitHub)" <gi...@apache.org>.
rreddy-22 commented on code in PR #20868:
URL: https://github.com/apache/kafka/pull/20868#discussion_r2525181590
##########
clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java:
##########
@@ -79,4 +110,55 @@ public void testGetErrorResponse() {
assertEquals(0, errorResponse.throttleTimeMs());
}
}
+
+ @Test
+ public void testTransactionVersion() {
+ // Test that TransactionVersion is set correctly and serialization handles it properly.
+ List<WriteTxnMarkersRequest.TxnMarkerEntry> markersWithVersion = Collections.singletonList(
+ new WriteTxnMarkersRequest.TxnMarkerEntry(
+ PRODUCER_ID, PRODUCER_EPOCH, COORDINATOR_EPOCH,
+ RESULT, Collections.singletonList(TOPIC_PARTITION), (short) 2)
+ );
+ WriteTxnMarkersRequest.Builder builder = new WriteTxnMarkersRequest.Builder(markersWithVersion);
+
+ // Test version 2 - TransactionVersion should be included.
+ WriteTxnMarkersRequest requestV2 = builder.build((short) 2);
+ assertNotNull(requestV2);
+ assertEquals(1, requestV2.markers().size());
+ // Verify TransactionVersion is set to 2 in the request data.
+ assertEquals((byte) 2, requestV2.data().markers().get(0).transactionVersion());
+ // Verify the request can be serialized for version 2 (TransactionVersion field included).
+ // This should not throw an exception.
+ requestV2.serialize();
+ int sizeV2 = requestV2.sizeInBytes();
+ // Verify TransactionVersion is set to 2 in the marker entry (markers() reads the field for version >= 2).
+ // This is what the broker sees when receiving the request - markers() is called on the partition leader side.
+ assertEquals((short) 2, requestV2.markers().get(0).transactionVersion());
+
+ // Test version 1 - TransactionVersion should be omitted (ignorable field).
+ WriteTxnMarkersRequest requestV1 = builder.build((short) 1);
+ assertNotNull(requestV1);
+ assertEquals(1, requestV1.markers().size());
+ // Verify TransactionVersion is still set to 2 in the request data (even for version 1).
+ // This is what the coordinator has when building the request - data() is used before serialization.
+ // The field value is preserved in the data, but will be omitted during serialization.
+ assertEquals((byte) 2, requestV1.data().markers().get(0).transactionVersion());
+ // Verify the request can be serialized for version 1 (TransactionVersion field omitted).
+ // This should not throw an exception even though TransactionVersion is set to 2
+ // because the field is marked as ignorable.
+ requestV1.serialize();
+ int sizeV1 = requestV1.sizeInBytes();
+ // After serialization, verify TransactionVersion is 0 in the marker entry for version 1 (field not read for version < 2).
+ // This is what the broker sees when receiving the request - markers() is called on the partition leader side.
+ assertEquals((short) 0, requestV1.markers().get(0).transactionVersion());
+
+ // Verify that version 2 is larger than version 1 because it includes TransactionVersion field.
+ // TransactionVersion is int8 (1 byte), so version 2 should be at least 1 byte larger.
+ // This check ensures that the serialization logic correctly includes/excludes the field.
+ assertTrue(sizeV2 > sizeV1,
Review Comment:
Line 152-153 is basically the deserialization part!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org