KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators

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

@@ -55,7 +55,7 @@ object TransactionCoordinator {
       config.transactionRemoveExpiredTransactionalIdCleanupIntervalMs,
       config.requestTimeoutMs)
 
-    val txnStateManager = new TransactionStateManager(config.brokerId, scheduler, replicaManager, txnConfig,
+    val txnStateManager = new TransactionStateManager(config.brokerId, scheduler, replicaManager, metadataCache, txnConfig,
       time, metrics)
 
     val logContext = new LogContext(s"[TransactionCoordinator id=${config.brokerId}] ")

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

@@ -62,7 +62,8 @@ object TransactionLog {
     *
     * @return value payload bytes
     */
-  private[transaction] def valueToBytes(txnMetadata: TxnTransitMetadata): Array[Byte] = {
+  private[transaction] def valueToBytes(txnMetadata: TxnTransitMetadata,
+                                        usesFlexibleRecords: Boolean): Array[Byte] = {
     if (txnMetadata.txnState == Empty && txnMetadata.topicPartitions.nonEmpty)
         throw new IllegalStateException(s"Transaction is not expected to have any partitions since its state is ${txnMetadata.txnState}: $txnMetadata")
 
@@ -75,9 +76,11 @@ object TransactionLog {
             .setPartitionIds(partitions.map(tp => Integer.valueOf(tp.partition)).toList.asJava)
         }.toList.asJava
 
-    // Serialize with the highest supported non-flexible version
-    // until a tagged field is introduced or the version is bumped.
-    MessageUtil.toVersionPrefixedBytes(0,
+    // Serialize with version 0 (highest non-flexible version) until transaction.version 1 is enabled
+    // which enables flexible fields in records.
+    val version: Short =
+      if (usesFlexibleRecords) 1 else 0
+    MessageUtil.toVersionPrefixedBytes(version,
       new TransactionLogValue()
         .setProducerId(txnMetadata.producerId)
         .setProducerEpoch(txnMetadata.producerEpoch)

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

@@ -20,7 +20,7 @@ import java.nio.ByteBuffer
 import java.util.Properties
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.locks.ReentrantReadWriteLock
-import kafka.server.{ReplicaManager, RequestLocal}
+import kafka.server.{MetadataCache, ReplicaManager, RequestLocal}
 import kafka.utils.CoreUtils.{inReadLock, inWriteLock}
 import kafka.utils.{Logging, Pool}
 import kafka.utils.Implicits._
@@ -36,6 +36,7 @@ import org.apache.kafka.common.requests.TransactionResult
 import org.apache.kafka.common.utils.{Time, Utils}
 import org.apache.kafka.common.{KafkaException, TopicPartition}
 import org.apache.kafka.coordinator.transaction.{TransactionLogConfigs, TransactionStateManagerConfigs}
+import org.apache.kafka.server.common.TransactionVersion
 import org.apache.kafka.server.config.ServerConfigs
 import org.apache.kafka.server.record.BrokerCompressionType
 import org.apache.kafka.server.util.Scheduler
@@ -65,6 +66,7 @@ import scala.collection.mutable
 class TransactionStateManager(brokerId: Int,
                               scheduler: Scheduler,
                               replicaManager: ReplicaManager,
+                              metadataCache: MetadataCache,
                               config: TransactionConfig,
                               time: Time,
                               metrics: Metrics) extends Logging {
@@ -99,6 +101,10 @@ class TransactionStateManager(brokerId: Int,
     TransactionStateManagerConfigs.METRICS_GROUP,
     "The avg time it took to load the partitions in the last 30sec"), new Avg())
 
+  private[transaction] def usesFlexibleRecords(): Boolean = {
+    metadataCache.features().finalizedFeatures().getOrDefault(TransactionVersion.FEATURE_NAME, 0.toShort) > 0
+  }
+
   // visible for testing only
   private[transaction] def addLoadingPartition(partitionId: Int, coordinatorEpoch: Int): Unit = {
     val partitionAndLeaderEpoch = TransactionPartitionAndLeaderEpoch(partitionId, coordinatorEpoch)
@@ -618,7 +624,7 @@ class TransactionStateManager(brokerId: Int,
 
     // generate the message for this transaction metadata
     val keyBytes = TransactionLog.keyToBytes(transactionalId)
-    val valueBytes = TransactionLog.valueToBytes(newMetadata)
+    val valueBytes = TransactionLog.valueToBytes(newMetadata, usesFlexibleRecords())
     val timestamp = time.milliseconds()
 
     val records = MemoryRecords.withRecords(TransactionLog.EnforcedCompression, new SimpleRecord(timestamp, keyBytes, valueBytes))

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

@@ -90,13 +90,15 @@ object BrokerFeatures extends Logging {
           } else {
             MetadataVersion.latestProduction.featureLevel
           }))
-    PRODUCTION_FEATURES.forEach { feature => features.put(feature.featureName,
-          new SupportedVersionRange(0,
-            if (unstableFeatureVersionsEnabled) {
-              feature.latestTesting
-            } else {
-              feature.latestProduction
-            }))
+    PRODUCTION_FEATURES.forEach {
+      feature =>
+        val maxVersion = if (unstableFeatureVersionsEnabled)
+          feature.latestTesting
+        else
+          feature.latestProduction
+        if (maxVersion > 0) {
+          features.put(feature.featureName, new SupportedVersionRange(feature.minimumProduction(), maxVersion))
+        }
     }
     Features.supportedFeatures(features)
   }

core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala

@@ -37,7 +37,7 @@ import org.apache.kafka.metadata.properties.MetaPropertiesEnsemble.VerificationF
 import org.apache.kafka.metadata.properties.{MetaProperties, MetaPropertiesEnsemble, MetaPropertiesVersion}
 import org.apache.kafka.network.SocketServerConfigs
 import org.apache.kafka.raft.QuorumConfig
-import org.apache.kafka.server.common.{ApiMessageAndVersion, MetadataVersion}
+import org.apache.kafka.server.common.{ApiMessageAndVersion, Features, MetadataVersion}
 import org.apache.kafka.server.config.{KRaftConfigs, ServerConfigs, ServerLogConfigs}
 import org.apache.kafka.server.fault.{FaultHandler, MockFaultHandler}
 import org.apache.zookeeper.client.ZKClientConfig
@@ -346,6 +346,13 @@ abstract class QuorumTestHarness extends Logging {
                         setName(MetadataVersion.FEATURE_NAME).
                         setFeatureLevel(metadataVersion.featureLevel()), 0.toShort))
 
+    metadataRecords.add(new ApiMessageAndVersion(
+      new FeatureLevelRecord()
+        .setName(Features.TRANSACTION_VERSION.featureName)
+        .setFeatureLevel(Features.TRANSACTION_VERSION.latestTesting),
+      0.toShort
+    ))
+
     optionalMetadataRecords.foreach { metadataArguments =>
       for (record <- metadataArguments) metadataRecords.add(record)
     }

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

@@ -35,6 +35,7 @@ import org.apache.kafka.common.record.{FileRecords, MemoryRecords, RecordBatch,
 import org.apache.kafka.common.requests._
 import org.apache.kafka.common.utils.{LogContext, MockTime, ProducerIdAndEpoch}
 import org.apache.kafka.common.{Node, TopicPartition}
+import org.apache.kafka.server.common.{FinalizedFeatures, MetadataVersion, TransactionVersion}
 import org.apache.kafka.storage.internals.log.{FetchDataInfo, FetchIsolation, LogConfig, LogOffsetMetadata}
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
@@ -75,7 +76,23 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
     when(zkClient.getTopicPartitionCount(TRANSACTION_STATE_TOPIC_NAME))
       .thenReturn(Some(numPartitions))
 
-    txnStateManager = new TransactionStateManager(0, scheduler, replicaManager, txnConfig, time,
+    val brokerNode = new Node(0, "host", 10)
+    val metadataCache: MetadataCache = mock(classOf[MetadataCache])
+    when(metadataCache.getPartitionLeaderEndpoint(
+      anyString,
+      anyInt,
+      any[ListenerName])
+    ).thenReturn(Some(brokerNode))
+    when(metadataCache.features()).thenReturn {
+      new FinalizedFeatures(
+        MetadataVersion.latestTesting(),
+        Collections.singletonMap(TransactionVersion.FEATURE_NAME, TransactionVersion.TV_2.featureLevel()),
+        0,
+        true
+      )
+    }
+
+    txnStateManager = new TransactionStateManager(0, scheduler, replicaManager, metadataCache, txnConfig, time,
       new Metrics())
     txnStateManager.startup(() => zkClient.getTopicPartitionCount(TRANSACTION_STATE_TOPIC_NAME).get,
       enableTransactionalIdExpiration = true)
@@ -89,13 +106,6 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
       } else {
         Success(producerId)
       })
-    val brokerNode = new Node(0, "host", 10)
-    val metadataCache: MetadataCache = mock(classOf[MetadataCache])
-    when(metadataCache.getPartitionLeaderEndpoint(
-      anyString,
-      anyInt,
-      any[ListenerName])
-    ).thenReturn(Some(brokerNode))
     val networkClient: NetworkClient = mock(classOf[NetworkClient])
     txnMarkerChannelManager = new TransactionMarkerChannelManager(
       KafkaConfig.fromProps(serverProps),
@@ -451,10 +461,10 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
     addPartitionsOp.awaitAndVerify(txn)
 
     val txnMetadata = transactionMetadata(txn).getOrElse(throw new IllegalStateException(s"Transaction not found $txn"))
-    txnRecords += new SimpleRecord(txn.txnMessageKeyBytes, TransactionLog.valueToBytes(txnMetadata.prepareNoTransit()))
+    txnRecords += new SimpleRecord(txn.txnMessageKeyBytes, TransactionLog.valueToBytes(txnMetadata.prepareNoTransit(), true))
 
     txnMetadata.state = PrepareCommit
-    txnRecords += new SimpleRecord(txn.txnMessageKeyBytes, TransactionLog.valueToBytes(txnMetadata.prepareNoTransit()))
+    txnRecords += new SimpleRecord(txn.txnMessageKeyBytes, TransactionLog.valueToBytes(txnMetadata.prepareNoTransit(), true))
 
     prepareTxnLog(partitionId)
   }

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

@@ -51,7 +51,7 @@ class TransactionLogTest {
     val txnMetadata = TransactionMetadata(transactionalId, producerId, producerEpoch, transactionTimeoutMs, 0)
     txnMetadata.addPartitions(topicPartitions)
 
-    assertThrows(classOf[IllegalStateException], () => TransactionLog.valueToBytes(txnMetadata.prepareNoTransit()))
+    assertThrows(classOf[IllegalStateException], () => TransactionLog.valueToBytes(txnMetadata.prepareNoTransit(), true))
   }
 
   @Test
@@ -79,7 +79,7 @@ class TransactionLogTest {
         txnMetadata.addPartitions(topicPartitions)
 
       val keyBytes = TransactionLog.keyToBytes(transactionalId)
-      val valueBytes = TransactionLog.valueToBytes(txnMetadata.prepareNoTransit())
+      val valueBytes = TransactionLog.valueToBytes(txnMetadata.prepareNoTransit(), true)
 
       new SimpleRecord(keyBytes, valueBytes)
     }.toSeq
@@ -119,7 +119,7 @@ class TransactionLogTest {
     txnMetadata.addPartitions(Set(topicPartition))
 
     val keyBytes = TransactionLog.keyToBytes(transactionalId)
-    val valueBytes = TransactionLog.valueToBytes(txnMetadata.prepareNoTransit())
+    val valueBytes = TransactionLog.valueToBytes(txnMetadata.prepareNoTransit(), true)
     val transactionMetadataRecord = TestUtils.records(Seq(
       new SimpleRecord(keyBytes, valueBytes)
     )).records.asScala.head
@@ -145,10 +145,17 @@ class TransactionLogTest {
   @Test
   def testSerializeTransactionLogValueToHighestNonFlexibleVersion(): Unit = {
     val txnTransitMetadata = TxnTransitMetadata(1, 1, 1, 1, 1000, CompleteCommit, Set.empty, 500, 500)
-    val txnLogValueBuffer = ByteBuffer.wrap(TransactionLog.valueToBytes(txnTransitMetadata))
+    val txnLogValueBuffer = ByteBuffer.wrap(TransactionLog.valueToBytes(txnTransitMetadata, false))
     assertEquals(0, txnLogValueBuffer.getShort)
   }
 
+  @Test
+  def testSerializeTransactionLogValueToFlexibleVersion(): Unit = {
+    val txnTransitMetadata = TxnTransitMetadata(1, 1, 1, 1, 1000, CompleteCommit, Set.empty, 500, 500)
+    val txnLogValueBuffer = ByteBuffer.wrap(TransactionLog.valueToBytes(txnTransitMetadata, true))
+    assertEquals(TransactionLogValue.HIGHEST_SUPPORTED_VERSION, txnLogValueBuffer.getShort)
+  }
+
   @Test
   def testDeserializeHighestSupportedTransactionLogValue(): Unit = {
     val txnPartitions = new TransactionLogValue.PartitionsSchema()