CalvinConfluent commented on code in PR #16840:
URL: https://github.com/apache/kafka/pull/16840#discussion_r1725948535
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -980,6 +1002,28 @@ void handleCoordinatorReady() {
null;
this.coordinatorSupportsBumpingEpoch = initProducerIdVersion != null &&
initProducerIdVersion.maxVersion() >= 3;
+
+ if (nodeApiVersions == null) return;
+
+ if (nodeApiVersions.finalizedFeatures() != null) {
+ /*
+ To enable the transaction V2, it requires:
+ 1. transaction.version finalized version >= 2
Review Comment:
> 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.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
