KAFKA-14563: RemoveClient-Side AddPartitionsToTxn Requests #16840

CalvinConfluent · 2024-08-08T22:19:17Z

https://issues.apache.org/jira/browse/KAFKA-14563

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitRequest.java

clients/src/main/resources/common/message/ProduceRequest.json

core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala

gradle/spotbugs-exclude.xml

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java

jolshan · 2024-08-19T18:08:49Z

Also cc: @artemlivshits to take a look :)

clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

artemlivshits · 2024-08-21T00:13:58Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

+        if (nodeApiVersions.finalizedFeatures() != null) {
+            /*
+                To enable the transaction V2, it requires:
+                1. transaction.version finalized version >= 2


We should check every node API versions (not just coordinator's) and keep track of the latest transaction version (where "latest" is defined as the latest epoch). So that logic would be something like this:

When we get API versions from any node, check if the transaction manager has the epoch that is less than the new epoch.

If the epoch stored in the transaction manager is less than the new epoch, store the new value and the new epoch in the transaction manager.

Have a method boolean coordinatorSupportsTransactionV2() { return this.transactionVersion >= 2; }

This way we learn the cluster-wide consensus on the transaction version, vs. individual brokers that may have different value so our view doesn't jump back and forth (if coordinator moves, for example).

This is a good point. I agree 👍

When we get API versions from any node

The optimal way of handling this is to have an ApiVersionsRequest callback and check the transaction version in it. However, it is a bit tricky here. The ApiVersionsRequest is handled in the KafkaClient, a layer below the Sender class. It does not expose any interaction method other than the apiVersions object. If we want to register a callback, we should have an interface change in the KafkaClient which may need a KIP?

A suboptimal way is to check the apiVersions for all the node features when calling the coordinatorSupportsTransactionV2. I guess it is acceptable as the checking does not cost much. If so, I guess we don't need to store the epoch and transactionVersion anymore.

Maybe I'm missing something, but can we store this within ApiVersions and update the value every time a new apiVersions request is handled? We already call apiVersions.update(node, nodeVersionInfo); and that does a calculation for max produce magic, so could we do another method in update to check the epoch and if it is greater than the existing epoch, we update the internal value in ApiVersions? (We may need to add epoch to NodeApiVersions -- and this is not a public api)

I did something a bit different here. Every time the ApiVersions are updated, the maxFinalizedFeaturesEpoch among the NodeApiVersions is stored. Then the transaction manager can just check this maxFinalizedFeaturesEpoch to determine if it needs to check the feature updates.

I think it gets the same thing accomplished. 👍

clients/src/main/java/org/apache/kafka/clients/ApiVersions.java

jolshan · 2024-09-06T18:35:31Z

Hmm have we checked the tests @CalvinConfluent? There seem to be some issues. (Not just the threads, but some related to transactions)

artemlivshits · 2024-09-24T00:44:05Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

@@ -971,7 +987,7 @@ synchronized boolean isInitializing() {
        return isTransactional() && currentState == State.INITIALIZING;
    }

-    void handleCoordinatorReady() {
+    synchronized void handleCoordinatorReady() {


What's the reason to make this synchronized?

artemlivshits · 2024-09-24T20:06:57Z

clients/src/main/java/org/apache/kafka/clients/ApiVersions.java


    public synchronized void update(String nodeId, NodeApiVersions nodeApiVersions) {
+        if (maxFinalizedFeaturesEpoch > nodeApiVersions.finalizedFeaturesEpoch()) {


I would expect that we still want to put the specific node's api versions into the per-node map. Then we can check the epoch and put cluster-wide finalized features if it has newer epoch. Something like this:

this.nodeApiVersions.put(nodeId, nodeApiVersions); this.maxUsableProduceMagic = computeMaxUsableProduceMagic(); if (maxFinalizedFeaturesEpoch > nodeApiVersions.finalizedFeaturesEpoch()) { this.maxFinalizedFeaturesEpoch = nodeApiVersions.finalizedFeaturesEpoch(); this.finalizedFeatures = nodeApiVersions.finalizedFeatures(); }

artemlivshits · 2024-09-24T20:13:30Z

clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java

+        if (finalizedFeaturesEpoch == -1) {
+            this.finalizedFeatures = Collections.emptyMap();
+            return;
+        }


nit: do we need this special case or the code below would just work? I think it's good to avoid extra code.

artemlivshits · 2024-09-24T20:15:13Z

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -1570,6 +1576,7 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
                metadata = new RecordMetadata(topicPartition(), -1, -1, RecordBatch.NO_TIMESTAMP, -1, -1);
            }
            this.interceptors.onAcknowledgement(metadata, exception);
+


nit: extra empty line?

artemlivshits · 2024-09-24T20:31:42Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

@@ -397,6 +399,7 @@ private long sendProducerData(long now) {
                // Update both readyTimeMs and drainTimeMs, this would "reset" the node
                // latency.
                this.accumulator.updateNodeLatencyStats(node.id(), now, true);
+                if (transactionManager != null) this.transactionManager.handleCoordinatorReady();


Do we still need this line or it's a left-over from a preview version of this PR?

artemlivshits · 2024-09-24T21:08:31Z

clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java

@@ -216,6 +221,10 @@ public String transactionalId() {
        return transactionalId;
    }

+    public boolean isTransactionV2Requested() {


Is this used?

artemlivshits · 2024-09-24T21:08:54Z

clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java

@@ -216,6 +221,10 @@ public String transactionalId() {
        return transactionalId;
    }

+    public boolean isTransactionV2Requested() {
+        return version() > LAST_BEFORE_TRANSACTION_V2_VERSION;


Should we call the static function?

artemlivshits · 2024-09-24T21:13:11Z

clients/src/main/resources/common/message/ProduceResponse.json

@@ -36,7 +36,12 @@
  // Version 10 adds 'CurrentLeader' and 'NodeEndpoints' as tagged fields (KIP-951)
  //
  // Version 11 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
-  "validVersions": "0-11",
+  //
+  // Version 12 is the same as version 11 (KIP-890). Note when produce requests are used in transaction, if


Should the extended comment be part of the request instead of response? So that it's consistent with TxnOffsetCommitRequest.

artemlivshits · 2024-09-24T22:07:13Z

core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala

 }

 /**
 * This is an enum which handles the Partition Response based on the Request Version and the exact operation
 *    defaultError:       This is the default workflow which maps to cases when the Produce Request Version or the Txn_offset_commit request was lower than the first version supporting the new Error Class
 *    genericError:       This maps to the case when the clients are updated to handle the TransactionAbortableException
- *    addPartition:       This is a WIP. To be updated as a part of KIP-890 Part 2
+ *    addPartition:       This allows the partition to be added to the transactions inflight with the Produce and TxnOffsetCommit requests.


Is this the right code model to represent client capabilities? The way it's coded it looks like there are 3 mutually exclusive options, but the options are not mutually exclusive: clients that support txnv2 can also support rollback error.

artemlivshits · 2024-09-24T22:12:28Z

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -848,6 +848,8 @@ class ReplicaManager(val config: KafkaConfig,
                   Errors.COORDINATOR_NOT_AVAILABLE |
                   Errors.NOT_COORDINATOR => Some(new NotEnoughReplicasException(
                s"Unable to verify the partition has been added to the transaction. Underlying error: ${error.toString}"))
+              case Errors.UNKNOWN_PRODUCER_ID => Some(new OutOfOrderSequenceException(


Can we add a comment why we translate this error this way? Also, it looks like we don't return this error any more (at least not in the product code).

CalvinConfluent changed the title ~~[kafka-14563] RemoveClient-Side AddPartitionsToTxn Requests~~ KAFKA-14563: RemoveClient-Side AddPartitionsToTxn Requests Aug 8, 2024