KAFKA-20322: Add API version discovery true for transaction manager (apache#21782)

[KIP-1228](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1228%3A+Add+Transaction+Version+to+WriteTxnMarkersRequest)
added a WriteTxnMarkersRequest v2 with a TransactionVersion field.
However, TransactionMarkerChannelManager creates its NetworkClient with
discoverBrokerVersions=false, which disables API version negotiation
with peer brokers. Without version discovery, the ApiVersions cache is
never populated — apiVersions.get(nodeId) returns null in
NetworkClient.doSend(), causing it to fall through to
builder.latestAllowedVersion() which blindly uses the highest version
the sending broker knows about rather than negotiating a mutually
supported version.  In this fix we enable discovery and also fix the
system test that could've caught this issue earlier. The system test was
run locally and I verified that it failed without the fix and passed
with the fix.

Reviewers: Artem Livshits <alivshits@confluent.io>, David Jacot
 <djacot@confluent.io>

Author: rreddy-22

core/src/main/scala/kafka/coordinator/transaction/TransactionMarkerChannelManager.scala

@@ -95,7 +95,7 @@ object TransactionMarkerChannelManager {
       config.connectionSetupTimeoutMs,
       config.connectionSetupTimeoutMaxMs,
       time,
-      false,
+      true,
       new ApiVersions,
       logContext,
       MetadataRecoveryStrategy.NONE

core/src/test/scala/unit/kafka/coordinator/transaction/TransactionMarkerChannelManagerTest.scala

@@ -25,13 +25,14 @@ import org.apache.kafka.clients.{ClientResponse, NetworkClient}
 import org.apache.kafka.common.protocol.{ApiKeys, Errors}
 import org.apache.kafka.common.record.internal.RecordBatch
 import org.apache.kafka.common.requests.{RequestHeader, TransactionResult, WriteTxnMarkersRequest, WriteTxnMarkersResponse}
-import org.apache.kafka.common.utils.MockTime
+import org.apache.kafka.common.utils.{LogContext, MockTime}
+import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.{Node, TopicPartition}
 import org.apache.kafka.coordinator.transaction.{TransactionMetadata, TransactionState}
 import org.apache.kafka.metadata.MetadataCache
 import org.apache.kafka.server.common.{MetadataVersion, TransactionVersion}
 import org.apache.kafka.server.metrics.{KafkaMetricsGroup, KafkaYammerMetrics}
-import org.apache.kafka.server.util.RequestAndCompletionHandler
+import org.apache.kafka.server.util.{InterBrokerSendThread, RequestAndCompletionHandler}
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.params.ParameterizedTest
@@ -627,6 +628,35 @@ class TransactionMarkerChannelManagerTest {
     })
   }
 
+  @Test
+  def shouldEnableApiVersionDiscoveryInFactoryMethod(): Unit = {
+    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1))
+    val metrics = new Metrics()
+    val logContext = new LogContext()
+    try {
+      val channelManager = TransactionMarkerChannelManager(
+        config,
+        metrics,
+        metadataCache,
+        txnStateManager,
+        time,
+        logContext
+      )
+      try {
+        val field = classOf[InterBrokerSendThread].getDeclaredField("networkClient")
+        field.setAccessible(true)
+        val client = field.get(channelManager).asInstanceOf[NetworkClient]
+        assertTrue(client.discoverBrokerVersions(),
+          "TransactionMarkerChannelManager should enable API version discovery to " +
+            "ensure compatibility during rolling upgrades")
+      } finally {
+        channelManager.shutdown()
+      }
+    } finally {
+      metrics.close()
+    }
+  }
+
   /**
    * Adjusts the transaction metadata based on the transaction version.
    * When transaction V2 is enabled, the producer epoch is incremented

tests/kafkatest/tests/core/transactions_mixed_versions_test.py

@@ -22,7 +22,7 @@
 from kafkatest.utils.transactions_utils import create_and_start_copiers
 from kafkatest.version import LATEST_3_3, LATEST_3_4, LATEST_3_5, \
     LATEST_3_6, LATEST_3_7, LATEST_3_8, LATEST_3_9, LATEST_4_0, \
-    LATEST_4_1, DEV_BRANCH, KafkaVersion, LATEST_STABLE_METADATA_VERSION
+    LATEST_4_1, LATEST_4_2, DEV_BRANCH, KafkaVersion, LATEST_STABLE_METADATA_VERSION
 
 from ducktape.tests.test import Test
 from ducktape.mark import matrix
@@ -147,30 +147,41 @@ def copy_messages_transactionally(self, input_topic, output_topic,
 
         return self.drain_consumer(concurrent_consumer, num_messages_to_copy)
 
-    def setup_topics(self):
-        assignment = ":".join(map(str, [self.kafka.idx(node) for node in self.kafka.nodes]))
-        transaction_assignment = ",".join(map(str, [assignment[::-1]] * 50))
+    def setup_topics(self, txn_state_leader_node, data_leader_node):
+        all_node_ids = [self.kafka.idx(node) for node in self.kafka.nodes]
+
+        # Data topics: data_leader_node first
+        data_leader_id = self.kafka.idx(data_leader_node)
+        data_replicas = [data_leader_id] + [nid for nid in all_node_ids if nid != data_leader_id]
+        data_assignment = ":".join(map(str, data_replicas))
+
+        # Internal topics: txn_state_leader_node first
+        txn_leader_id = self.kafka.idx(txn_state_leader_node)
+        txn_replicas = [txn_leader_id] + [nid for nid in all_node_ids if nid != txn_leader_id]
+        txn_assignment = ":".join(map(str, txn_replicas))
+        internal_topic_assignment = ",".join([txn_assignment] * 50)
+
         self.kafka.topics = {
             self.input_topic: {
                 "partitions": self.num_input_partitions,
                 "replication-factor": self.replication_factor,
-                "replica-assignment": assignment,
+                "replica-assignment": data_assignment,
                 "configs": {
                     "min.insync.replicas": 2
                 }
             },
             self.output_topic: {
                 "partitions": self.num_output_partitions,
                 "replication-factor": self.replication_factor,
-                "replica-assignment": assignment,
+                "replica-assignment": data_assignment,
                 "configs": {
                     "min.insync.replicas": 2
                 }
             },
             "__transaction_state": {
                 "partitions": 50,
                 "replication-factor": self.replication_factor,
-                "replica-assignment": transaction_assignment,
+                "replica-assignment": internal_topic_assignment,
                 "configs": {
                     "min.insync.replicas": 2
                 }
@@ -179,11 +190,12 @@ def setup_topics(self):
 
     @cluster(num_nodes=8)
     @matrix(
-        old_kafka_version=[str(LATEST_4_1), str(LATEST_4_0), str(LATEST_3_9), str(LATEST_3_8), str(LATEST_3_7), str(LATEST_3_6), str(LATEST_3_5), str(LATEST_3_4), str(LATEST_3_3)],
+        old_kafka_version=[str(LATEST_4_2), str(LATEST_4_1), str(LATEST_4_0), str(LATEST_3_9), str(LATEST_3_8), str(LATEST_3_7), str(LATEST_3_6), str(LATEST_3_5), str(LATEST_3_4), str(LATEST_3_3)],
         metadata_quorum=[isolated_kraft],
-        group_protocol=[None]
+        group_protocol=[None],
+        coordinator_on_new_broker=[True, False]
     )
-    def test_transactions_mixed_versions(self, old_kafka_version, metadata_quorum=quorum.isolated_kraft, group_protocol=None):
+    def test_transactions_mixed_versions(self, old_kafka_version, metadata_quorum=quorum.isolated_kraft, group_protocol=None, coordinator_on_new_broker=True):
         oldKafkaVersion = KafkaVersion(old_kafka_version)
         self.kafka = KafkaService(self.test_context,
                                   num_nodes=self.num_brokers,
@@ -193,6 +205,9 @@ def test_transactions_mixed_versions(self, old_kafka_version, metadata_quorum=qu
 
         self.kafka.nodes[0].version = DEV_BRANCH
 
+        new_node = self.kafka.nodes[0]  # DEV_BRANCH
+        old_node = self.kafka.nodes[1]  # old version
+
         security_protocol = 'PLAINTEXT'
         self.kafka.security_protocol = security_protocol
         self.kafka.interbroker_security_protocol = security_protocol
@@ -202,7 +217,10 @@ def test_transactions_mixed_versions(self, old_kafka_version, metadata_quorum=qu
 
         self.kafka.log_level = "DEBUG"
 
-        self.setup_topics()
+        if coordinator_on_new_broker:
+            self.setup_topics(txn_state_leader_node=new_node, data_leader_node=old_node)
+        else:
+            self.setup_topics(txn_state_leader_node=old_node, data_leader_node=new_node)
         self.kafka.start()
 
         input_messages = self.seed_messages(self.input_topic, self.num_seed_messages)