KAFKA-14869: Bump coordinator Value records to flexible versions (KIP-915, Part-2)

core/src/main/resources/common/message/TransactionLogValue.json

@@ -16,24 +16,27 @@
 {
   "type": "data",
   "name": "TransactionLogValue",
-  "validVersions": "0",
-  "flexibleVersions": "none",
+  // Version 1 is the first flexible version.
+  // KIP-915: bumping the version will no longer make this record backward compatible.
+  // We suggest to add/remove only tagged fields to maintain backward compatibility.
+  "validVersions": "0-1",
+  "flexibleVersions": "1+",
   "fields": [
-    { "name": "ProducerId", "type": "int64", "versions": "0",
+    { "name": "ProducerId", "type": "int64", "versions": "0+",
       "about": "Producer id in use by the transactional id"},
-    { "name": "ProducerEpoch", "type": "int16", "versions": "0",
+    { "name": "ProducerEpoch", "type": "int16", "versions": "0+",
       "about": "Epoch associated with the producer id"},
-    { "name": "TransactionTimeoutMs", "type": "int32", "versions": "0",
+    { "name": "TransactionTimeoutMs", "type": "int32", "versions": "0+",
       "about": "Transaction timeout in milliseconds"},
-    { "name": "TransactionStatus", "type": "int8", "versions": "0",
+    { "name": "TransactionStatus", "type": "int8", "versions": "0+",
       "about": "TransactionState the transaction is in"},
-    { "name": "TransactionPartitions", "type": "[]PartitionsSchema", "versions": "0", "nullableVersions": "0",
+    { "name": "TransactionPartitions", "type": "[]PartitionsSchema", "versions": "0+", "nullableVersions": "0+",
       "about": "Set of partitions involved in the transaction", "fields": [
-      { "name": "Topic", "type": "string", "versions": "0"},
-      { "name": "PartitionIds", "type": "[]int32", "versions": "0"}]},
-    { "name": "TransactionLastUpdateTimestampMs", "type": "int64", "versions": "0",
+      { "name": "Topic", "type": "string", "versions": "0+"},
+      { "name": "PartitionIds", "type": "[]int32", "versions": "0+"}]},
+    { "name": "TransactionLastUpdateTimestampMs", "type": "int64", "versions": "0+",
       "about": "Time the transaction was last updated"},
-    { "name": "TransactionStartTimestampMs", "type": "int64", "versions": "0",
+    { "name": "TransactionStartTimestampMs", "type": "int64", "versions": "0+",
       "about": "Time the transaction was started"}
   ]
 }

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

@@ -1087,6 +1087,8 @@ object GroupMetadataManager {
     val version =
       if (metadataVersion.isLessThan(IBP_2_1_IV0) || offsetAndMetadata.expireTimestamp.nonEmpty) 1.toShort
       else if (metadataVersion.isLessThan(IBP_2_1_IV1)) 2.toShort
+      // Serialize with the highest supported non-flexible version
+      // until a tagged field is introduced or the version is bumped.
       else 3.toShort
     MessageUtil.toVersionPrefixedBytes(version, new OffsetCommitValue()
       .setOffset(offsetAndMetadata.offset)
@@ -1115,6 +1117,8 @@ object GroupMetadataManager {
       if (metadataVersion.isLessThan(IBP_0_10_1_IV0)) 0.toShort
       else if (metadataVersion.isLessThan(IBP_2_1_IV0)) 1.toShort
       else if (metadataVersion.isLessThan(IBP_2_3_IV0)) 2.toShort
+      // Serialize with the highest supported non-flexible version
+      // until a tagged field is introduced or the version is bumped.
       else 3.toShort
 
     MessageUtil.toVersionPrefixedBytes(version, new GroupMetadataValue()

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

@@ -82,7 +82,9 @@ object TransactionLog {
             .setPartitionIds(partitions.map(tp => Integer.valueOf(tp.partition)).toList.asJava)
         }.toList.asJava
 
-    MessageUtil.toVersionPrefixedBytes(TransactionLogValue.HIGHEST_SUPPORTED_VERSION,
+    // Serialize with the highest supported non-flexible version
+    // until a tagged field is introduced or the version is bumped.
+    MessageUtil.toVersionPrefixedBytes(0,
       new TransactionLogValue()
         .setProducerId(txnMetadata.producerId)
         .setProducerEpoch(txnMetadata.producerEpoch)

core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala

@@ -35,11 +35,14 @@ import org.apache.kafka.clients.consumer.internals.ConsumerProtocol
 import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
 import org.apache.kafka.common.internals.Topic
 import org.apache.kafka.common.metrics.{JmxReporter, KafkaMetricsContext, Metrics => kMetrics}
-import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.protocol.types.Field.TaggedFieldsSection
+import org.apache.kafka.common.protocol.types.{CompactArrayOf, Field, Schema, Struct, Type}
+import org.apache.kafka.common.protocol.{ByteBufferAccessor, Errors}
 import org.apache.kafka.common.record._
 import org.apache.kafka.common.requests.OffsetFetchResponse
 import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
 import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.coordinator.group.generated.{GroupMetadataValue, OffsetCommitValue}
 import org.apache.kafka.server.common.MetadataVersion
 import org.apache.kafka.server.common.MetadataVersion._
 import org.apache.kafka.server.metrics.KafkaYammerMetrics
@@ -2466,6 +2469,168 @@ class GroupMetadataManagerTest {
       verifySerde(version)
   }
 
+  @Test
+  def testSerializeGroupMetadataValueToHighestNonFlexibleVersion(): Unit = {
+    val generation = 935
+    val protocolType = "consumer"
+    val protocol = "range"
+    val memberId = "98098230493"
+    val assignmentBytes = Utils.toArray(ConsumerProtocol.serializeAssignment(
+      new ConsumerPartitionAssignor.Assignment(List(new TopicPartition("topic", 0)).asJava, null)
+    ))
+    val record = TestUtils.records(Seq(
+      buildStableGroupRecordWithMember(generation, protocolType, protocol, memberId, assignmentBytes)
+    )).records.asScala.head
+    assertEquals(3, record.value.getShort)
+  }
+
+  @Test
+  def testSerializeOffsetCommitValueToHighestNonFlexibleVersion(): Unit = {
+    val committedOffsets = Map(
+      new TopicPartition("foo", 0) -> 23L,
+      new TopicPartition("foo", 1) -> 455L,
+    )
+
+    val offsetCommitRecords = createCommittedOffsetRecords(committedOffsets)
+    offsetCommitRecords.foreach { record =>
+      assertEquals(3, record.value.getShort)
+    }
+  }
+
+  @Test
+  def testDeserializeOffsetCommitValueWithUnknownTaggedFields(): Unit = {
+    // Copy of OffsetCommitValue.SCHEMA_4 with a few
+    // additional tagged fields.
+    val futureOffsetCommitSchema = new Schema(
+      new Field("offset", Type.INT64, ""),
+      new Field("leader_epoch", Type.INT32, ""),
+      new Field("metadata", Type.COMPACT_STRING, ""),
+      new Field("commit_timestamp", Type.INT64, ""),
+      TaggedFieldsSection.of(
+        0, new Field("offset_foo", Type.STRING, ""),
+        1, new Field("offset_bar", Type.INT32, "")
+      )
+    )
+
+    // create OffsetCommitValue with tagged fields
+    val offsetCommit = new Struct(futureOffsetCommitSchema)
+    offsetCommit.set("offset", 1000L)
+    offsetCommit.set("leader_epoch", 100)
+    offsetCommit.set("metadata", "metadata")
+    offsetCommit.set("commit_timestamp", 2000L)
+    val offsetCommitTaggedFields = new java.util.TreeMap[Integer, Any]()
+    offsetCommitTaggedFields.put(0, "foo")
+    offsetCommitTaggedFields.put(1, 4000)
+    offsetCommit.set("_tagged_fields", offsetCommitTaggedFields)
+
+    // Prepare the buffer.
+    val buffer = ByteBuffer.allocate(offsetCommit.sizeOf() + 2)
+    buffer.put(0.toByte)
+    buffer.put(4.toByte) // Add 4 as version.
+    offsetCommit.writeTo(buffer)
+    buffer.flip()
+
+    // Read the buffer with the real schema and verify that tagged
+    // fields were read but ignored.
+    buffer.getShort() // Skip version.
+    val value = new OffsetCommitValue(new ByteBufferAccessor(buffer), 4.toShort)
+    assertEquals(Seq(0, 1), value.unknownTaggedFields().asScala.map(_.tag))
+
+    // Read the buffer with readOffsetMessageValue.
+    buffer.rewind()
+    val offsetAndMetadata = GroupMetadataManager.readOffsetMessageValue(buffer)
+    assertEquals(1000L, offsetAndMetadata.offset)
+    assertEquals(100, offsetAndMetadata.leaderEpoch.get)
+    assertEquals("metadata", offsetAndMetadata.metadata)
+    assertEquals(2000L, offsetAndMetadata.commitTimestamp)
+  }
+
+  @Test
+  def testDeserializeGroupMetadataValueWithUnknownTaggedFields(): Unit = {
+    // Copy of GroupMetadataValue.MemberMetadata.SCHEMA_4 with a few
+    // additional tagged fields.
+    val futureMemberSchema = new Schema(
+      new Field("member_id", Type.COMPACT_STRING, ""),
+      new Field("group_instance_id", Type.COMPACT_NULLABLE_STRING, ""),
+      new Field("client_id", Type.COMPACT_STRING, ""),
+      new Field("client_host", Type.COMPACT_STRING, ""),
+      new Field("rebalance_timeout", Type.INT32, ""),
+      new Field("session_timeout", Type.INT32, ""),
+      new Field("subscription", Type.COMPACT_BYTES, ""),
+      new Field("assignment", Type.COMPACT_BYTES, ""),
+      TaggedFieldsSection.of(
+        0, new Field("member_foo", Type.STRING, ""),
+        1, new Field("member_foo", Type.INT32, "")
+      )
+    )
+
+    // Copy of GroupMetadataValue.SCHEMA_4 with a few
+    // additional tagged fields.
+    val futureGroupSchema = new Schema(
+      new Field("protocol_type", Type.COMPACT_STRING, ""),
+      new Field("generation", Type.INT32, ""),
+      new Field("protocol", Type.COMPACT_NULLABLE_STRING, ""),
+      new Field("leader", Type.COMPACT_NULLABLE_STRING, ""),
+      new Field("current_state_timestamp", Type.INT64, ""),
+      new Field("members", new CompactArrayOf(futureMemberSchema), ""),
+      TaggedFieldsSection.of(
+        0, new Field("group_foo", Type.STRING, ""),
+        1, new Field("group_bar", Type.INT32, "")
+      )
+    )
+
+    // Create a member with tagged fields.
+    val member = new Struct(futureMemberSchema)
+    member.set("member_id", "member_id")
+    member.set("group_instance_id", "group_instance_id")
+    member.set("client_id", "client_id")
+    member.set("client_host", "client_host")
+    member.set("rebalance_timeout", 1)
+    member.set("session_timeout", 2)
+    member.set("subscription", ByteBuffer.allocate(0))
+    member.set("assignment", ByteBuffer.allocate(0))
+
+    val memberTaggedFields = new java.util.TreeMap[Integer, Any]()
+    memberTaggedFields.put(0, "foo")
+    memberTaggedFields.put(1, 4000)
+    member.set("_tagged_fields", memberTaggedFields)
+
+    // Create a group with tagged fields.
+    val group = new Struct(futureGroupSchema)
+    group.set("protocol_type", "consumer")
+    group.set("generation", 10)
+    group.set("protocol", "range")
+    group.set("leader", "leader")
+    group.set("current_state_timestamp", 1000L)
+    group.set("members", Array(member))
+
+    val groupTaggedFields = new java.util.TreeMap[Integer, Any]()
+    groupTaggedFields.put(0, "foo")
+    groupTaggedFields.put(1, 4000)
+    group.set("_tagged_fields", groupTaggedFields)
+
+    // Prepare the buffer.
+    val buffer = ByteBuffer.allocate(group.sizeOf() + 2)
+    buffer.put(0.toByte)
+    buffer.put(4.toByte) // Add 4 as version.
+    group.writeTo(buffer)
+    buffer.flip()
+
+    // Read the buffer with the real schema and verify that tagged
+    // fields were read but ignored.
+    buffer.getShort() // Skip version.
+    val value = new GroupMetadataValue(new ByteBufferAccessor(buffer), 4.toShort)
+    assertEquals(Seq(0, 1), value.unknownTaggedFields().asScala.map(_.tag))
+    assertEquals(Seq(0, 1), value.members().get(0).unknownTaggedFields().asScala.map(_.tag))
+
+    // Read the buffer with readGroupMessageValue.
+    buffer.rewind()
+    val groupMetadata = GroupMetadataManager.readGroupMessageValue("group", buffer, time)
+    assertEquals("consumer", groupMetadata.protocolType.get)
+    assertEquals("leader", groupMetadata.leaderOrNull)
+    assertTrue(groupMetadata.allMembers.contains("member_id"))
+  }
+
   @Test
   def testLoadOffsetsWithEmptyControlBatch(): Unit = {
     val groupMetadataTopicPartition = groupTopicPartition
@@ -2616,7 +2781,8 @@ class GroupMetadataManagerTest {
                                                memberId: String,
                                                assignmentBytes: Array[Byte] = Array.emptyByteArray,
                                                metadataVersion: MetadataVersion = MetadataVersion.latest): SimpleRecord = {
-    val memberProtocols = List((protocol, Array.emptyByteArray))
+    val subscription = new Subscription(List("topic").asJava)
+    val memberProtocols = List((protocol, ConsumerProtocol.serializeSubscription(subscription).array()))
     val member = new MemberMetadata(memberId, Some(groupInstanceId), "clientId", "clientHost", 30000, 10000, protocolType, memberProtocols)
     val group = GroupMetadata.loadGroup(groupId, Stable, generation, protocolType, protocol, memberId,
       if (metadataVersion.isAtLeast(IBP_2_1_IV0)) Some(time.milliseconds()) else None, Seq(member), time)

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

@@ -17,12 +17,17 @@
 package kafka.coordinator.transaction
 
 
+import kafka.internals.generated.TransactionLogValue
 import kafka.utils.TestUtils
 import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.protocol.ByteBufferAccessor
+import org.apache.kafka.common.protocol.types.Field.TaggedFieldsSection
+import org.apache.kafka.common.protocol.types.{CompactArrayOf, Field, Schema, Struct, Type}
 import org.apache.kafka.common.record.{CompressionType, MemoryRecords, SimpleRecord}
 import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
 import org.junit.jupiter.api.Test
 
+import java.nio.ByteBuffer
 import scala.jdk.CollectionConverters._
 
 class TransactionLogTest {
@@ -135,4 +140,88 @@ class TransactionLogTest {
     assertEquals(Some("<DELETE>"), valueStringOpt)
   }
 
+  @Test
+  def testSerializeTransactionLogValueToHighestNonFlexibleVersion(): Unit = {
+    val txnTransitMetadata = TxnTransitMetadata(1, 1, 1, 1, 1000, CompleteCommit, Set.empty, 500, 500)
+    val txnLogValueBuffer = ByteBuffer.wrap(TransactionLog.valueToBytes(txnTransitMetadata))
+    assertEquals(0, txnLogValueBuffer.getShort)
+  }
+
+  @Test
+  def testDeserializeTransactionLogValueWithUnknownTaggedFields(): Unit = {
+    // Copy of TransactionLogValue.PartitionsSchema.SCHEMA_1 with a few
+    // additional tagged fields.
+    val futurePartitionsSchema = new Schema(
+      new Field("topic", Type.COMPACT_STRING, ""),
+      new Field("partition_ids", new CompactArrayOf(Type.INT32), ""),
+      TaggedFieldsSection.of(
+        0, new Field("partition_foo", Type.STRING, ""),
+        1, new Field("partition_foo", Type.INT32, "")
+      )
+    )
+
+    // create TransactionLogValue.PartitionsSchema with tagged fields
+    val txnPartitions = new Struct(futurePartitionsSchema)
+    txnPartitions.set("topic", "topic")
+    txnPartitions.set("partition_ids", Array(Integer.valueOf(1)))
+    val txnPartitionsTaggedFields = new java.util.TreeMap[Integer, Any]()
+    txnPartitionsTaggedFields.put(0, "foo")
+    txnPartitionsTaggedFields.put(1, 4000)
+    txnPartitions.set("_tagged_fields", txnPartitionsTaggedFields)
+
+    // Copy of TransactionLogValue.SCHEMA_1 with a few
+    // additional tagged fields.
+    val futureTransactionLogValueSchema = new Schema(
+      new Field("producer_id", Type.INT64, ""),
+      new Field("producer_epoch", Type.INT16, ""),
+      new Field("transaction_timeout_ms", Type.INT32, ""),
+      new Field("transaction_status", Type.INT8, ""),
+      new Field("transaction_partitions", CompactArrayOf.nullable(futurePartitionsSchema), ""),
+      new Field("transaction_last_update_timestamp_ms", Type.INT64, ""),
+      new Field("transaction_start_timestamp_ms", Type.INT64, ""),
+      TaggedFieldsSection.of(
+        0, new Field("txn_foo", Type.STRING, ""),
+        1, new Field("txn_bar", Type.INT32, "")
+      )
+    )
+
+    // create TransactionLogValue with tagged fields
+    val transactionLogValue = new Struct(futureTransactionLogValueSchema)
+    transactionLogValue.set("producer_id", 1000L)
+    transactionLogValue.set("producer_epoch", 100.toShort)
+    transactionLogValue.set("transaction_timeout_ms", 1000)
+    transactionLogValue.set("transaction_status", CompleteCommit.id)
+    transactionLogValue.set("transaction_partitions", Array(txnPartitions))
+    transactionLogValue.set("transaction_last_update_timestamp_ms", 2000L)
+    transactionLogValue.set("transaction_start_timestamp_ms", 3000L)
+    val txnLogValueTaggedFields = new java.util.TreeMap[Integer, Any]()
+    txnLogValueTaggedFields.put(0, "foo")
+    txnLogValueTaggedFields.put(1, 4000)
+    transactionLogValue.set("_tagged_fields", txnLogValueTaggedFields)
+
+    // Prepare the buffer.
+    val buffer = ByteBuffer.allocate(transactionLogValue.sizeOf() + 2)
+    buffer.put(0.toByte)
+    buffer.put(1.toByte) // Add 1 as version.
+    transactionLogValue.writeTo(buffer)
+    buffer.flip()
+
+    // Read the buffer with the real schema and verify that tagged
+    // fields were read but ignored.
+    buffer.getShort() // Skip version.
+    val value = new TransactionLogValue(new ByteBufferAccessor(buffer), 1.toShort)
+    assertEquals(Seq(0, 1), value.unknownTaggedFields().asScala.map(_.tag))
+
+    // Read the buffer with readTxnRecordValue.
+    buffer.rewind()
+    val txnMetadata = TransactionLog.readTxnRecordValue("transaction-id", buffer).get
+    assertEquals(1000L, txnMetadata.producerId)
+    assertEquals(100, txnMetadata.producerEpoch)
+    assertEquals(1000L, txnMetadata.txnTimeoutMs)
+    assertEquals(CompleteCommit, txnMetadata.state)
+    assertEquals(Set(new TopicPartition("topic", 1)), txnMetadata.topicPartitions)
+    assertEquals(2000L, txnMetadata.txnLastUpdateTimestamp)
+    assertEquals(3000L, txnMetadata.txnStartTimestamp)
+  }
+
 }

generator/src/main/java/org/apache/kafka/message/MessageGenerator.java

@@ -231,9 +231,9 @@ public static void processDirectories(String packageName,
             for (Path inputPath : directoryStream) {
                 try {
                     MessageSpec spec = JSON_SERDE.
-                        readValue(inputPath.toFile(), MessageSpec.class);
+                            readValue(inputPath.toFile(), MessageSpec.class);
                     List<MessageClassGenerator> generators =
-                        createMessageClassGenerators(packageName, messageClassGeneratorTypes);
+                            createMessageClassGenerators(packageName, messageClassGeneratorTypes);
                     for (MessageClassGenerator generator : generators) {
                         String name = generator.outputName(spec) + JAVA_SUFFIX;
                         outputFileNames.add(name);