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

build.gradle

@@ -958,12 +958,15 @@ project(':core') {
     classpath = configurations.generator
     args = [ "-p", "kafka.internals.generated",
              "-o", "src/generated/java/kafka/internals/generated",
-             "-i", "src/main/resources/common/message",
+             "-i", "src/main/resources/common/message,src/test/resources/message",
              "-m", "MessageDataGenerator"
     ]
     inputs.dir("src/main/resources/common/message")
         .withPropertyName("messages")
         .withPathSensitivity(PathSensitivity.RELATIVE)
+    inputs.dir("src/test/resources/message")
+            .withPropertyName("testMessages")
+            .withPathSensitivity(PathSensitivity.RELATIVE)
     outputs.cacheIf { true }
     outputs.dir("src/generated/java/kafka/internals/generated")
   }

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

@@ -13,27 +13,31 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+// Version 1 is the first flexible version.
+// ONLY USED FOR TESTING
 {
   "type": "data",
   "name": "TransactionLogValue",
-  "validVersions": "0",
-  "flexibleVersions": "none",
+  // 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/resources/message/GroupMetadataValueWithUnknownTaggedFields.json

@@ -0,0 +1,53 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Version 4 is the first flexible version.
+// ONLY USED FOR TESTING
+{
+  "type": "data",
+  "name": "GroupMetadataValueWithUnknownTaggedFields",
+  // 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-4",
+  "flexibleVersions": "4+",
+  "fields": [
+    { "name": "protocolType", "versions": "0+", "type": "string"},
+    { "name": "generation", "versions": "0+", "type": "int32" },
+    { "name": "protocol", "versions": "0+", "type": "string", "nullableVersions": "0+" },
+    { "name": "leader", "versions": "0+", "type": "string", "nullableVersions": "0+" },
+    { "name": "currentStateTimestamp", "versions": "2+", "type": "int64", "default": -1, "ignorable": true},
+    { "name": "members", "versions": "0+", "type": "[]MemberMetadata" },
+
+    // Introduce unknown tagged fields. The tag ids should not overlap with tag ids from GroupMetadataValue.
+    { "name":  "unknownTaggedField1", "type":  "string", "versions":  "4+", "tag": 0, "taggedVersions": "4+"},
+    { "name":  "unknownTaggedField2", "type":  "int64", "versions":  "4+", "tag": 1, "taggedVersions": "4+"}
+  ],
+  "commonStructs": [
+    {
+      "name": "MemberMetadata",
+      "versions": "0-3",
+      "fields": [
+        { "name": "memberId", "versions": "0+", "type": "string" },
+        { "name": "groupInstanceId", "versions": "3+", "type": "string", "default": "null", "nullableVersions": "3+", "ignorable": true},
+        { "name": "clientId", "versions": "0+", "type": "string" },
+        { "name": "clientHost", "versions": "0+", "type": "string" },
+        { "name": "rebalanceTimeout", "versions": "1+", "type": "int32", "ignorable": true},
+        { "name": "sessionTimeout", "versions": "0+", "type": "int32" },
+        { "name": "subscription", "versions": "0+", "type": "bytes" },
+        { "name": "assignment", "versions": "0+", "type": "bytes" }
+      ]
+    }
+  ]
+}

core/src/test/resources/message/OffsetCommitValueWithUnknownTaggedFields.json

@@ -0,0 +1,36 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Version 4 is the first flexible version.
+// ONLY USED FOR TESTING
+{
+  "type": "data",
+  "name": "OffsetCommitValueWithUnknownTaggedFields",
+  // 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-4",
+  "flexibleVersions": "4+",
+  "fields": [
+    { "name": "offset", "type": "int64", "versions": "0+" },
+    { "name": "leaderEpoch", "type": "int32", "versions": "3+", "default": -1, "ignorable": true},
+    { "name": "metadata", "type": "string", "versions": "0+" },
+    { "name": "commitTimestamp", "type": "int64", "versions": "0+" },
+    { "name": "expireTimestamp", "type": "int64", "versions": "1", "default": -1, "ignorable": true},
+
+    // Introduce unknown tagged fields. The tag ids should not overlap with tag ids from OffsetCommitValue.
+    { "name":  "unknownTaggedField1", "type":  "string", "versions":  "4+", "tag": 0, "taggedVersions": "4+"},
+    { "name":  "unknownTaggedField2", "type":  "int64", "versions":  "4+", "tag": 1, "taggedVersions": "4+"}
+  ]
+}

core/src/test/resources/message/TransactionLogValueWithUnknownTaggedFields.json

@@ -0,0 +1,46 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Version 1 is the first flexible version.
+{
+  "type": "data",
+  "name": "TransactionLogValueWithUnknownTaggedFields",
+  // 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+",
+      "about": "Producer id in use by the transactional id"},
+    { "name": "ProducerEpoch", "type": "int16", "versions": "0+",
+      "about": "Epoch associated with the producer id"},
+    { "name": "TransactionTimeoutMs", "type": "int32", "versions": "0+",
+      "about": "Transaction timeout in milliseconds"},
+    { "name": "TransactionStatus", "type": "int8", "versions": "0+",
+      "about": "TransactionState the transaction is in"},
+    { "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+",
+      "about": "Time the transaction was last updated"},
+    { "name": "TransactionStartTimestampMs", "type": "int64", "versions": "0+",
+      "about": "Time the transaction was started"},
+
+    // Introduce unknown tagged fields. The tag ids should not overlap with tag ids from TransactionLogValue.
+    { "name":  "unknownTaggedField1", "type":  "string", "versions":  "1+", "tag": 0, "taggedVersions": "1+"},
+    { "name":  "unknownTaggedField2", "type":  "int64", "versions":  "1+", "tag": 1, "taggedVersions": "1+"}
+  ]
+}

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

@@ -26,6 +26,7 @@ import com.yammer.metrics.core.Gauge
 import javax.management.ObjectName
 import kafka.cluster.Partition
 import kafka.common.OffsetAndMetadata
+import kafka.internals.generated.{GroupMetadataValueWithUnknownTaggedFields, OffsetCommitValueWithUnknownTaggedFields}
 import kafka.log.UnifiedLog
 import kafka.server.{HostedPartition, KafkaConfig, ReplicaManager, RequestLocal}
 import kafka.utils.{MockTime, TestUtils}
@@ -35,7 +36,7 @@ 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.{Errors, MessageUtil}
 import org.apache.kafka.common.record._
 import org.apache.kafka.common.requests.OffsetFetchResponse
 import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
@@ -2466,6 +2467,64 @@ 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 = {
+    val offsetCommitValue = new OffsetCommitValueWithUnknownTaggedFields()
+      .setOffset(1000L)
+      .setMetadata("metadata")
+      .setUnknownTaggedField1("unknown tagged field")
+      .setUnknownTaggedField2(2000L)
+    val serialized = MessageUtil.toVersionPrefixedByteBuffer(4, offsetCommitValue)
+    val offsetAndMetadata = GroupMetadataManager.readOffsetMessageValue(serialized)
+    assertEquals(1000L, offsetAndMetadata.offset)
+    assertEquals("metadata", offsetAndMetadata.metadata)
+  }
+
+  @Test
+  def testDeserializeGroupMetadataValueWithUnknownTaggedFields(): Unit = {
+    val member = new GroupMetadataValueWithUnknownTaggedFields.MemberMetadata()
+      .setMemberId("member")
+    val groupMetadataValue = new GroupMetadataValueWithUnknownTaggedFields()
+      .setProtocolType("consumer")
+      .setLeader("leader")
+      .setMembers(Collections.singletonList(member))
+      .setUnknownTaggedField1("unknown tagged field")
+      .setUnknownTaggedField2(2000L)
+    val serialized = MessageUtil.toVersionPrefixedByteBuffer(4, groupMetadataValue)
+    val groupMetadata = GroupMetadataManager.readGroupMessageValue("group", serialized, time)
+    assertEquals("consumer", groupMetadata.protocolType.get)
+    assertEquals("leader", groupMetadata.leaderOrNull)
+    assertTrue(groupMetadata.allMembers.contains("member"))
+  }
+
   @Test
   def testLoadOffsetsWithEmptyControlBatch(): Unit = {
     val groupMetadataTopicPartition = groupTopicPartition
@@ -2616,7 +2675,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,15 @@
 package kafka.coordinator.transaction
 
 
+import kafka.internals.generated.TransactionLogValueWithUnknownTaggedFields
 import kafka.utils.TestUtils
 import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.protocol.MessageUtil
 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 +138,28 @@ 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 = {
+    val transactionLogValue = new TransactionLogValueWithUnknownTaggedFields()
+      .setProducerId(1000L)
+      .setTransactionStatus(CompleteCommit.id)
+      .setUnknownTaggedField1("unknown tagged field")
+      .setUnknownTaggedField2(2000L)
+
+    val serialized = MessageUtil.toVersionPrefixedBytes(1, transactionLogValue)
+
+    val transactionMetadata = TransactionLog.readTxnRecordValue("transaction",
+      ByteBuffer.wrap(serialized)).get
+
+    assertEquals(1000L, transactionMetadata.producerId)
+    assertEquals(CompleteCommit, transactionMetadata.state)
+  }
+
 }