From 435853b79f58bb6eccc1ff97ef89c62d629ca0ec Mon Sep 17 00:00:00 2001 From: Jeff Kim Date: Fri, 7 Apr 2023 12:41:13 -0400 Subject: [PATCH 1/9] bump Value records to flexible versions --- .../common/message/TransactionLogValue.json | 23 ++++++------- .../group/GroupMetadataManager.scala | 8 +++-- .../transaction/TransactionLog.scala | 4 ++- .../group/GroupMetadataManagerTest.scala | 32 ++++++++++++++++++- .../transaction/TransactionLogTest.scala | 9 ++++++ .../apache/kafka/message/SchemaGenerator.java | 8 +++++ .../common/message/GroupMetadataValue.json | 5 +-- .../common/message/OffsetCommitValue.json | 5 +-- 8 files changed, 75 insertions(+), 19 deletions(-) diff --git a/core/src/main/resources/common/message/TransactionLogValue.json b/core/src/main/resources/common/message/TransactionLogValue.json index 7915c3d7cb715..5edfd9d9077f4 100644 --- a/core/src/main/resources/common/message/TransactionLogValue.json +++ b/core/src/main/resources/common/message/TransactionLogValue.json @@ -13,27 +13,28 @@ // See the License for the specific language governing permissions and // limitations under the License. +// KIP-915: bumping the version will no longer make this record backward compatible. { "type": "data", "name": "TransactionLogValue", - "validVersions": "0", - "flexibleVersions": "none", + "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"} ] } diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala index 2ea0d79662e73..c60eff5b09151 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala @@ -1087,7 +1087,9 @@ 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 - else 3.toShort + // Serialize with the highest supported non-flexible version + // until a tagged field is introduced or the version is bumped. + else OffsetCommitValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION MessageUtil.toVersionPrefixedBytes(version, new OffsetCommitValue() .setOffset(offsetAndMetadata.offset) .setMetadata(offsetAndMetadata.metadata) @@ -1115,7 +1117,9 @@ 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 - else 3.toShort + // Serialize with the highest supported non-flexible version + // until a tagged field is introduced or the version is bumped. + else GroupMetadataValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION MessageUtil.toVersionPrefixedBytes(version, new GroupMetadataValue() .setProtocolType(groupMetadata.protocolType.getOrElse("")) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala index cb501f774fd9d..0b427705c5e47 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala +++ b/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(TransactionLogValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION, new TransactionLogValue() .setProducerId(txnMetadata.producerId) .setProducerEpoch(txnMetadata.producerEpoch) diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index 333aeac5af711..554cfc21d79bf 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala @@ -40,6 +40,7 @@ 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 +2467,34 @@ 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(GroupMetadataValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION, 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(OffsetCommitValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION, record.value().getShort) + } + } + @Test def testLoadOffsetsWithEmptyControlBatch(): Unit = { val groupMetadataTopicPartition = groupTopicPartition @@ -2616,7 +2645,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) diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala index 32e17d88a7b1e..ff1cee568cd72 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala @@ -17,12 +17,14 @@ package kafka.coordinator.transaction +import kafka.internals.generated.TransactionLogValue import kafka.utils.TestUtils import org.apache.kafka.common.TopicPartition 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 +137,11 @@ class TransactionLogTest { assertEquals(Some(""), 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(TransactionLogValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION, txnLogValueBuffer.getShort) + } + } diff --git a/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java b/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java index c20ae4c001e3d..e0d64c7b9be82 100644 --- a/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java +++ b/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java @@ -369,6 +369,14 @@ void writeSchema(String className, CodeBuffer buffer) throws Exception { buffer.printf("public static final short LOWEST_SUPPORTED_VERSION = %d;%n", versions.lowest()); buffer.printf("public static final short HIGHEST_SUPPORTED_VERSION = %d;%n", versions.highest()); + + if (!messageFlexibleVersions.empty()) { + Versions nonFlexibleVersions = versions.subtract(messageFlexibleVersions); + if(nonFlexibleVersions != null) { + buffer.printf("public static final short HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION = %d;%n", + nonFlexibleVersions.highest()); + } + } buffer.printf("%n"); } } diff --git a/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json b/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json index 826a7c8d326d4..116658f986bba 100644 --- a/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json +++ b/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json @@ -13,11 +13,12 @@ // See the License for the specific language governing permissions and // limitations under the License. +// KIP-915: bumping the version will no longer make this record backward compatible. { "type": "data", "name": "GroupMetadataValue", - "validVersions": "0-3", - "flexibleVersions": "none", + "validVersions": "0-4", + "flexibleVersions": "4+", "fields": [ { "name": "protocolType", "versions": "0+", "type": "string"}, { "name": "generation", "versions": "0+", "type": "int32" }, diff --git a/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json b/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json index db8a6281d43b6..febf98354fe74 100644 --- a/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json +++ b/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json @@ -13,11 +13,12 @@ // See the License for the specific language governing permissions and // limitations under the License. +// KIP-915: bumping the version will no longer make this record backward compatible. { "type": "data", "name": "OffsetCommitValue", - "validVersions": "0-3", - "flexibleVersions": "none", + "validVersions": "0-4", + "flexibleVersions": "4+", "fields": [ { "name": "offset", "type": "int64", "versions": "0+" }, { "name": "leaderEpoch", "type": "int32", "versions": "3+", "default": -1, "ignorable": true}, From 28710d4def1f06d0f259a2efb062dd7cfa74b010 Mon Sep 17 00:00:00 2001 From: Jeff Kim Date: Mon, 10 Apr 2023 17:46:06 -0400 Subject: [PATCH 2/9] first flexible version comment --- core/src/main/resources/common/message/TransactionLogValue.json | 1 + .../src/main/resources/common/message/GroupMetadataValue.json | 1 + .../src/main/resources/common/message/OffsetCommitValue.json | 1 + 3 files changed, 3 insertions(+) diff --git a/core/src/main/resources/common/message/TransactionLogValue.json b/core/src/main/resources/common/message/TransactionLogValue.json index 5edfd9d9077f4..0f03a374c15cb 100644 --- a/core/src/main/resources/common/message/TransactionLogValue.json +++ b/core/src/main/resources/common/message/TransactionLogValue.json @@ -13,6 +13,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +// Version 1 is the first flexible version. // KIP-915: bumping the version will no longer make this record backward compatible. { "type": "data", diff --git a/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json b/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json index 116658f986bba..ab8c8bfc183e4 100644 --- a/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json +++ b/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json @@ -13,6 +13,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +// Version 4 is the first flexible version. // KIP-915: bumping the version will no longer make this record backward compatible. { "type": "data", diff --git a/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json b/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json index febf98354fe74..531d87ba60fc6 100644 --- a/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json +++ b/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json @@ -13,6 +13,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +// Version 4 is the first flexible version. // KIP-915: bumping the version will no longer make this record backward compatible. { "type": "data", From d74107a09232c5ed47512b9004ce910104f5ff1f Mon Sep 17 00:00:00 2001 From: Jeff Kim Date: Thu, 13 Apr 2023 13:59:00 -0400 Subject: [PATCH 3/9] address comments --- build.gradle | 5 +- .../common/message/TransactionLogValue.json | 4 +- .../group/GroupMetadataManager.scala | 4 +- .../transaction/TransactionLog.scala | 2 +- ...pMetadataValueWithUnknownTaggedFields.json | 53 +++++++++++++++++++ ...setCommitValueWithUnknownTaggedFields.json | 36 +++++++++++++ ...actionLogValueWithUnknownTaggedFields.json | 46 ++++++++++++++++ .../group/GroupMetadataManagerTest.scala | 38 +++++++++++-- .../transaction/TransactionLogTest.scala | 22 +++++++- .../kafka/message/MessageGenerator.java | 40 +++++++------- .../apache/kafka/message/SchemaGenerator.java | 8 --- .../common/message/GroupMetadataValue.json | 3 +- .../common/message/OffsetCommitValue.json | 3 +- 13 files changed, 224 insertions(+), 40 deletions(-) create mode 100644 core/src/test/resources/message/GroupMetadataValueWithUnknownTaggedFields.json create mode 100644 core/src/test/resources/message/OffsetCommitValueWithUnknownTaggedFields.json create mode 100644 core/src/test/resources/message/TransactionLogValueWithUnknownTaggedFields.json diff --git a/build.gradle b/build.gradle index 32c95786d9d30..0f4ee2a9bba63 100644 --- a/build.gradle +++ b/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") } diff --git a/core/src/main/resources/common/message/TransactionLogValue.json b/core/src/main/resources/common/message/TransactionLogValue.json index 0f03a374c15cb..b37237766893f 100644 --- a/core/src/main/resources/common/message/TransactionLogValue.json +++ b/core/src/main/resources/common/message/TransactionLogValue.json @@ -14,10 +14,12 @@ // limitations under the License. // Version 1 is the first flexible version. -// KIP-915: bumping the version will no longer make this record backward compatible. +// ONLY USED FOR TESTING { "type": "data", "name": "TransactionLogValue", + // 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": [ diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala index c60eff5b09151..3d03914bf1172 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala @@ -1089,7 +1089,7 @@ object GroupMetadataManager { 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 OffsetCommitValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION + else 3.toShort MessageUtil.toVersionPrefixedBytes(version, new OffsetCommitValue() .setOffset(offsetAndMetadata.offset) .setMetadata(offsetAndMetadata.metadata) @@ -1119,7 +1119,7 @@ object GroupMetadataManager { 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 GroupMetadataValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION + else 3.toShort MessageUtil.toVersionPrefixedBytes(version, new GroupMetadataValue() .setProtocolType(groupMetadata.protocolType.getOrElse("")) diff --git a/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala b/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala index 0b427705c5e47..0fa0280312164 100644 --- a/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala +++ b/core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala @@ -84,7 +84,7 @@ object TransactionLog { // Serialize with the highest supported non-flexible version // until a tagged field is introduced or the version is bumped. - MessageUtil.toVersionPrefixedBytes(TransactionLogValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION, + MessageUtil.toVersionPrefixedBytes(0, new TransactionLogValue() .setProducerId(txnMetadata.producerId) .setProducerEpoch(txnMetadata.producerEpoch) diff --git a/core/src/test/resources/message/GroupMetadataValueWithUnknownTaggedFields.json b/core/src/test/resources/message/GroupMetadataValueWithUnknownTaggedFields.json new file mode 100644 index 0000000000000..c082a10a315d3 --- /dev/null +++ b/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" } + ] + } + ] +} diff --git a/core/src/test/resources/message/OffsetCommitValueWithUnknownTaggedFields.json b/core/src/test/resources/message/OffsetCommitValueWithUnknownTaggedFields.json new file mode 100644 index 0000000000000..78f079525d0fd --- /dev/null +++ b/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+"} + ] +} diff --git a/core/src/test/resources/message/TransactionLogValueWithUnknownTaggedFields.json b/core/src/test/resources/message/TransactionLogValueWithUnknownTaggedFields.json new file mode 100644 index 0000000000000..9a6c219766dd5 --- /dev/null +++ b/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+"} + ] +} diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index 554cfc21d79bf..34feede9f8d2d 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/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,12 +36,11 @@ 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 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 @@ -2479,7 +2479,7 @@ class GroupMetadataManagerTest { val record = TestUtils.records(Seq( buildStableGroupRecordWithMember(generation, protocolType, protocol, memberId, assignmentBytes) )).records.asScala.head - assertEquals(GroupMetadataValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION, record.value().getShort) + assertEquals(3, record.value().getShort) } @Test @@ -2491,10 +2491,40 @@ class GroupMetadataManagerTest { val offsetCommitRecords = createCommittedOffsetRecords(committedOffsets) offsetCommitRecords.foreach { record => - assertEquals(OffsetCommitValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION, record.value().getShort) + 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 diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala index ff1cee568cd72..2728113b72bd1 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala @@ -17,9 +17,10 @@ package kafka.coordinator.transaction -import kafka.internals.generated.TransactionLogValue +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 @@ -141,7 +142,24 @@ class TransactionLogTest { def testSerializeTransactionLogValueToHighestNonFlexibleVersion(): Unit = { val txnTransitMetadata = TxnTransitMetadata(1, 1, 1, 1, 1000, CompleteCommit, Set.empty, 500, 500) val txnLogValueBuffer = ByteBuffer.wrap(TransactionLog.valueToBytes(txnTransitMetadata)) - assertEquals(TransactionLogValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION, txnLogValueBuffer.getShort) + 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) } } diff --git a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java index 56f3f6ab0b2ce..237591b7ac344 100644 --- a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java +++ b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java @@ -217,7 +217,7 @@ private static List createMessageClassGenerators(String p public static void processDirectories(String packageName, String outputDir, - String inputDir, + String inputDirs, List typeClassGeneratorTypes, List messageClassGeneratorTypes) throws Exception { Files.createDirectories(Paths.get(outputDir)); @@ -226,26 +226,28 @@ public static void processDirectories(String packageName, List typeClassGenerators = createTypeClassGenerators(packageName, typeClassGeneratorTypes); HashSet outputFileNames = new HashSet<>(); - try (DirectoryStream directoryStream = Files - .newDirectoryStream(Paths.get(inputDir), JSON_GLOB)) { - for (Path inputPath : directoryStream) { - try { - MessageSpec spec = JSON_SERDE. - readValue(inputPath.toFile(), MessageSpec.class); - List generators = - createMessageClassGenerators(packageName, messageClassGeneratorTypes); - for (MessageClassGenerator generator : generators) { - String name = generator.outputName(spec) + JAVA_SUFFIX; - outputFileNames.add(name); - Path outputPath = Paths.get(outputDir, name); - try (BufferedWriter writer = Files.newBufferedWriter(outputPath)) { - generator.generateAndWrite(spec, writer); + for (String inputDir : inputDirs.split(",")) { + try (DirectoryStream directoryStream = Files + .newDirectoryStream(Paths.get(inputDir), JSON_GLOB)) { + for (Path inputPath : directoryStream) { + try { + MessageSpec spec = JSON_SERDE. + readValue(inputPath.toFile(), MessageSpec.class); + List generators = + createMessageClassGenerators(packageName, messageClassGeneratorTypes); + for (MessageClassGenerator generator : generators) { + String name = generator.outputName(spec) + JAVA_SUFFIX; + outputFileNames.add(name); + Path outputPath = Paths.get(outputDir, name); + try (BufferedWriter writer = Files.newBufferedWriter(outputPath)) { + generator.generateAndWrite(spec, writer); + } } + numProcessed++; + typeClassGenerators.forEach(generator -> generator.registerMessageType(spec)); + } catch (Exception e) { + throw new RuntimeException("Exception while processing " + inputPath.toString(), e); } - numProcessed++; - typeClassGenerators.forEach(generator -> generator.registerMessageType(spec)); - } catch (Exception e) { - throw new RuntimeException("Exception while processing " + inputPath.toString(), e); } } } diff --git a/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java b/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java index e0d64c7b9be82..c20ae4c001e3d 100644 --- a/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java +++ b/generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java @@ -369,14 +369,6 @@ void writeSchema(String className, CodeBuffer buffer) throws Exception { buffer.printf("public static final short LOWEST_SUPPORTED_VERSION = %d;%n", versions.lowest()); buffer.printf("public static final short HIGHEST_SUPPORTED_VERSION = %d;%n", versions.highest()); - - if (!messageFlexibleVersions.empty()) { - Versions nonFlexibleVersions = versions.subtract(messageFlexibleVersions); - if(nonFlexibleVersions != null) { - buffer.printf("public static final short HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION = %d;%n", - nonFlexibleVersions.highest()); - } - } buffer.printf("%n"); } } diff --git a/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json b/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json index ab8c8bfc183e4..f8b0150495df2 100644 --- a/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json +++ b/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json @@ -14,10 +14,11 @@ // limitations under the License. // Version 4 is the first flexible version. -// KIP-915: bumping the version will no longer make this record backward compatible. { "type": "data", "name": "GroupMetadataValue", + // 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": [ diff --git a/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json b/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json index 531d87ba60fc6..53b03686d19ce 100644 --- a/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json +++ b/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json @@ -14,10 +14,11 @@ // limitations under the License. // Version 4 is the first flexible version. -// KIP-915: bumping the version will no longer make this record backward compatible. { "type": "data", "name": "OffsetCommitValue", + // 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": [ From 173802a8127f8f9e296ca3eb4f19be0a4d61d8f9 Mon Sep 17 00:00:00 2001 From: Jeff Kim Date: Fri, 14 Apr 2023 14:19:39 -0400 Subject: [PATCH 4/9] remove test schemas --- build.gradle | 5 +- .../common/message/TransactionLogValue.json | 3 +- ...pMetadataValueWithUnknownTaggedFields.json | 53 ------- ...setCommitValueWithUnknownTaggedFields.json | 36 ----- ...actionLogValueWithUnknownTaggedFields.json | 46 ------ .../group/GroupMetadataManagerTest.scala | 150 +++++++++++++++--- .../transaction/TransactionLogTest.scala | 92 +++++++++-- .../kafka/message/MessageGenerator.java | 40 +++-- .../common/message/GroupMetadataValue.json | 4 +- .../common/message/OffsetCommitValue.json | 2 +- 10 files changed, 229 insertions(+), 202 deletions(-) delete mode 100644 core/src/test/resources/message/GroupMetadataValueWithUnknownTaggedFields.json delete mode 100644 core/src/test/resources/message/OffsetCommitValueWithUnknownTaggedFields.json delete mode 100644 core/src/test/resources/message/TransactionLogValueWithUnknownTaggedFields.json diff --git a/build.gradle b/build.gradle index 0f4ee2a9bba63..32c95786d9d30 100644 --- a/build.gradle +++ b/build.gradle @@ -958,15 +958,12 @@ project(':core') { classpath = configurations.generator args = [ "-p", "kafka.internals.generated", "-o", "src/generated/java/kafka/internals/generated", - "-i", "src/main/resources/common/message,src/test/resources/message", + "-i", "src/main/resources/common/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") } diff --git a/core/src/main/resources/common/message/TransactionLogValue.json b/core/src/main/resources/common/message/TransactionLogValue.json index b37237766893f..c6efc772d58db 100644 --- a/core/src/main/resources/common/message/TransactionLogValue.json +++ b/core/src/main/resources/common/message/TransactionLogValue.json @@ -13,11 +13,10 @@ // 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", + // 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", diff --git a/core/src/test/resources/message/GroupMetadataValueWithUnknownTaggedFields.json b/core/src/test/resources/message/GroupMetadataValueWithUnknownTaggedFields.json deleted file mode 100644 index c082a10a315d3..0000000000000 --- a/core/src/test/resources/message/GroupMetadataValueWithUnknownTaggedFields.json +++ /dev/null @@ -1,53 +0,0 @@ -// 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" } - ] - } - ] -} diff --git a/core/src/test/resources/message/OffsetCommitValueWithUnknownTaggedFields.json b/core/src/test/resources/message/OffsetCommitValueWithUnknownTaggedFields.json deleted file mode 100644 index 78f079525d0fd..0000000000000 --- a/core/src/test/resources/message/OffsetCommitValueWithUnknownTaggedFields.json +++ /dev/null @@ -1,36 +0,0 @@ -// 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+"} - ] -} diff --git a/core/src/test/resources/message/TransactionLogValueWithUnknownTaggedFields.json b/core/src/test/resources/message/TransactionLogValueWithUnknownTaggedFields.json deleted file mode 100644 index 9a6c219766dd5..0000000000000 --- a/core/src/test/resources/message/TransactionLogValueWithUnknownTaggedFields.json +++ /dev/null @@ -1,46 +0,0 @@ -// 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+"} - ] -} diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index 34feede9f8d2d..81151b02d8fac 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala @@ -26,7 +26,6 @@ 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} @@ -36,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, MessageUtil} +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 @@ -2479,7 +2481,7 @@ class GroupMetadataManagerTest { val record = TestUtils.records(Seq( buildStableGroupRecordWithMember(generation, protocolType, protocol, memberId, assignmentBytes) )).records.asScala.head - assertEquals(3, record.value().getShort) + assertEquals(3, record.value.getShort) } @Test @@ -2491,38 +2493,142 @@ class GroupMetadataManagerTest { val offsetCommitRecords = createCommittedOffsetRecords(committedOffsets) offsetCommitRecords.foreach { record => - assertEquals(3, record.value().getShort) + 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) + // 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 = { - 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) + // 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")) + assertTrue(groupMetadata.allMembers.contains("member_id")) } @Test diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala index 2728113b72bd1..8504b276260c5 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala @@ -17,10 +17,12 @@ package kafka.coordinator.transaction -import kafka.internals.generated.TransactionLogValueWithUnknownTaggedFields +import kafka.internals.generated.TransactionLogValue import kafka.utils.TestUtils import org.apache.kafka.common.TopicPartition -import org.apache.kafka.common.protocol.MessageUtil +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 @@ -147,19 +149,79 @@ class TransactionLogTest { @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) + // 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) } } diff --git a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java index 237591b7ac344..4c3cd3e6efd75 100644 --- a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java +++ b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java @@ -217,7 +217,7 @@ private static List createMessageClassGenerators(String p public static void processDirectories(String packageName, String outputDir, - String inputDirs, + String inputDir, List typeClassGeneratorTypes, List messageClassGeneratorTypes) throws Exception { Files.createDirectories(Paths.get(outputDir)); @@ -226,28 +226,26 @@ public static void processDirectories(String packageName, List typeClassGenerators = createTypeClassGenerators(packageName, typeClassGeneratorTypes); HashSet outputFileNames = new HashSet<>(); - for (String inputDir : inputDirs.split(",")) { - try (DirectoryStream directoryStream = Files - .newDirectoryStream(Paths.get(inputDir), JSON_GLOB)) { - for (Path inputPath : directoryStream) { - try { - MessageSpec spec = JSON_SERDE. - readValue(inputPath.toFile(), MessageSpec.class); - List generators = - createMessageClassGenerators(packageName, messageClassGeneratorTypes); - for (MessageClassGenerator generator : generators) { - String name = generator.outputName(spec) + JAVA_SUFFIX; - outputFileNames.add(name); - Path outputPath = Paths.get(outputDir, name); - try (BufferedWriter writer = Files.newBufferedWriter(outputPath)) { - generator.generateAndWrite(spec, writer); - } + try (DirectoryStream directoryStream = Files + .newDirectoryStream(Paths.get(inputDir), JSON_GLOB)) { + for (Path inputPath : directoryStream) { + try { + MessageSpec spec = JSON_SERDE. + readValue(inputPath.toFile(), MessageSpec.class); + List generators = + createMessageClassGenerators(packageName, messageClassGeneratorTypes); + for (MessageClassGenerator generator : generators) { + String name = generator.outputName(spec) + JAVA_SUFFIX; + outputFileNames.add(name); + Path outputPath = Paths.get(outputDir, name); + try (BufferedWriter writer = Files.newBufferedWriter(outputPath)) { + generator.generateAndWrite(spec, writer); } - numProcessed++; - typeClassGenerators.forEach(generator -> generator.registerMessageType(spec)); - } catch (Exception e) { - throw new RuntimeException("Exception while processing " + inputPath.toString(), e); } + numProcessed++; + typeClassGenerators.forEach(generator -> generator.registerMessageType(spec)); + } catch (Exception e) { + throw new RuntimeException("Exception while processing " + inputPath.toString(), e); } } } diff --git a/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json b/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json index f8b0150495df2..8405fbb991853 100644 --- a/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json +++ b/group-coordinator/src/main/resources/common/message/GroupMetadataValue.json @@ -13,10 +13,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Version 4 is the first flexible version. { "type": "data", "name": "GroupMetadataValue", + // Version 4 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-4", @@ -32,7 +32,7 @@ "commonStructs": [ { "name": "MemberMetadata", - "versions": "0-3", + "versions": "0+", "fields": [ { "name": "memberId", "versions": "0+", "type": "string" }, { "name": "groupInstanceId", "versions": "3+", "type": "string", "default": "null", "nullableVersions": "3+", "ignorable": true}, diff --git a/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json b/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json index 53b03686d19ce..2973c5ee12ab2 100644 --- a/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json +++ b/group-coordinator/src/main/resources/common/message/OffsetCommitValue.json @@ -13,10 +13,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Version 4 is the first flexible version. { "type": "data", "name": "OffsetCommitValue", + // Version 4 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-4", From a091b961c5039051f882ab4a6cdbd5356102177c Mon Sep 17 00:00:00 2001 From: Jeff Kim Date: Mon, 17 Apr 2023 11:10:37 -0400 Subject: [PATCH 5/9] address comments --- .../group/GroupMetadataManagerTest.scala | 58 +++++++++++++++++-- .../transaction/TransactionLogTest.scala | 4 +- .../kafka/message/MessageGenerator.java | 4 +- 3 files changed, 58 insertions(+), 8 deletions(-) diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index 81151b02d8fac..a8da3928dd26c 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala @@ -37,7 +37,7 @@ import org.apache.kafka.common.internals.Topic import org.apache.kafka.common.metrics.{JmxReporter, KafkaMetricsContext, Metrics => kMetrics} 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.protocol.{ByteBufferAccessor, Errors, MessageUtil} import org.apache.kafka.common.record._ import org.apache.kafka.common.requests.OffsetFetchResponse import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse @@ -2498,7 +2498,56 @@ class GroupMetadataManagerTest { } @Test - def testDeserializeOffsetCommitValueWithUnknownTaggedFields(): Unit = { + def testDeserializeHighestSupportedGroupMetadataValueVersion(): Unit = { + val member = new GroupMetadataValue.MemberMetadata() + .setMemberId("member") + .setClientId("client") + .setClientHost("host") + + val generation = 935 + val protocolType = "consumer" + val protocol = "range" + val leader = "leader" + val groupMetadataValue = new GroupMetadataValue() + .setProtocolType(protocolType) + .setGeneration(generation) + .setProtocol(protocol) + .setLeader(leader) + .setMembers(java.util.Collections.singletonList(member)) + + val deserialized = GroupMetadataManager.readGroupMessageValue("groupId", + MessageUtil.toVersionPrefixedByteBuffer(4, groupMetadataValue), time) + + assertEquals(generation, deserialized.generationId) + assertEquals(protocolType, deserialized.protocolType.get) + assertEquals(protocol, deserialized.protocolName.get) + assertEquals(leader, deserialized.leaderOrNull) + + val actualMember = deserialized.allMemberMetadata.head + assertEquals(member.memberId, actualMember.memberId) + assertEquals(member.clientId, actualMember.clientId) + assertEquals(member.clientHost, actualMember.clientHost) + } + + @Test + def testDeserializeHighestSupportedOffsetCommitValueVersion(): Unit = { + val offsetCommitValue = new OffsetCommitValue() + .setOffset(1000L) + .setMetadata("metadata") + .setCommitTimestamp(1500L) + .setLeaderEpoch(1) + + val serialized = MessageUtil.toVersionPrefixedByteBuffer(4, offsetCommitValue) + val deserialized = GroupMetadataManager.readOffsetMessageValue(serialized) + + assertEquals(1000L, deserialized.offset) + assertEquals("metadata", deserialized.metadata) + assertEquals(1500L, deserialized.commitTimestamp) + assertEquals(1, deserialized.leaderEpoch.get) + } + + @Test + def testDeserializeFutureOffsetCommitValue(): Unit = { // Copy of OffsetCommitValue.SCHEMA_4 with a few // additional tagged fields. val futureOffsetCommitSchema = new Schema( @@ -2546,7 +2595,7 @@ class GroupMetadataManagerTest { } @Test - def testDeserializeGroupMetadataValueWithUnknownTaggedFields(): Unit = { + def testDeserializeFutureGroupMetadataValue(): Unit = { // Copy of GroupMetadataValue.MemberMetadata.SCHEMA_4 with a few // additional tagged fields. val futureMemberSchema = new Schema( @@ -2781,8 +2830,7 @@ class GroupMetadataManagerTest { memberId: String, assignmentBytes: Array[Byte] = Array.emptyByteArray, metadataVersion: MetadataVersion = MetadataVersion.latest): SimpleRecord = { - val subscription = new Subscription(List("topic").asJava) - val memberProtocols = List((protocol, ConsumerProtocol.serializeSubscription(subscription).array())) + val memberProtocols = List((protocol, Array.emptyByteArray)) 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) diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala index 8504b276260c5..313f272fa53a1 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala @@ -28,6 +28,7 @@ import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows} import org.junit.jupiter.api.Test import java.nio.ByteBuffer +import scala.collection.Seq import scala.jdk.CollectionConverters._ class TransactionLogTest { @@ -148,7 +149,7 @@ class TransactionLogTest { } @Test - def testDeserializeTransactionLogValueWithUnknownTaggedFields(): Unit = { + def testDeserializeFutureTransactionLogValue(): Unit = { // Copy of TransactionLogValue.PartitionsSchema.SCHEMA_1 with a few // additional tagged fields. val futurePartitionsSchema = new Schema( @@ -211,6 +212,7 @@ class TransactionLogTest { buffer.getShort() // Skip version. val value = new TransactionLogValue(new ByteBufferAccessor(buffer), 1.toShort) assertEquals(Seq(0, 1), value.unknownTaggedFields().asScala.map(_.tag)) + assertEquals(Seq(0, 1), value.transactionPartitions().get(0).unknownTaggedFields().asScala.map(_.tag)) // Read the buffer with readTxnRecordValue. buffer.rewind() diff --git a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java index 4c3cd3e6efd75..56f3f6ab0b2ce 100644 --- a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java +++ b/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 generators = - createMessageClassGenerators(packageName, messageClassGeneratorTypes); + createMessageClassGenerators(packageName, messageClassGeneratorTypes); for (MessageClassGenerator generator : generators) { String name = generator.outputName(spec) + JAVA_SUFFIX; outputFileNames.add(name); From 2e82907be7042737fd3988daa98daadb6b6a932a Mon Sep 17 00:00:00 2001 From: Jeff Kim Date: Mon, 17 Apr 2023 11:17:36 -0400 Subject: [PATCH 6/9] add deserialize highest supported txn log value test --- .../transaction/TransactionLogTest.scala | 34 +++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala index 313f272fa53a1..03fca4421359c 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala @@ -20,11 +20,11 @@ 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.{ByteBufferAccessor, MessageUtil} 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.Assertions.{assertEquals, assertThrows, assertTrue} import org.junit.jupiter.api.Test import java.nio.ByteBuffer @@ -148,6 +148,36 @@ class TransactionLogTest { assertEquals(0, txnLogValueBuffer.getShort) } + @Test + def testDeserializeHighestSupportedTransactionLogValue(): Unit = { + val txnPartitions = new TransactionLogValue.PartitionsSchema() + .setTopic("topic") + .setPartitionIds(java.util.Collections.singletonList(0)) + + val txnLogValue = new TransactionLogValue() + .setProducerId(100) + .setProducerEpoch(50.toShort) + .setTransactionStatus(CompleteCommit.id) + .setTransactionStartTimestampMs(750L) + .setTransactionLastUpdateTimestampMs(1000L) + .setTransactionTimeoutMs(500) + .setTransactionPartitions(java.util.Collections.singletonList(txnPartitions)) + + val serialized = MessageUtil.toVersionPrefixedByteBuffer(1, txnLogValue) + val deserialized = TransactionLog.readTxnRecordValue("transactionId", serialized).get + + assertEquals(100, deserialized.producerId) + assertEquals(50, deserialized.producerEpoch) + assertEquals(CompleteCommit, deserialized.state) + assertEquals(750L, deserialized.txnStartTimestamp) + assertEquals(1000L, deserialized.txnLastUpdateTimestamp) + assertEquals(500, deserialized.txnTimeoutMs) + + val actualTxnPartitions = deserialized.topicPartitions + assertEquals(1, actualTxnPartitions.size) + assertTrue(actualTxnPartitions.contains(new TopicPartition("topic", 0))) + } + @Test def testDeserializeFutureTransactionLogValue(): Unit = { // Copy of TransactionLogValue.PartitionsSchema.SCHEMA_1 with a few From 1a78f4ecda87ea2444046ff54d31d19422c587a1 Mon Sep 17 00:00:00 2001 From: Jeff Kim Date: Mon, 17 Apr 2023 12:39:45 -0400 Subject: [PATCH 7/9] nit --- .../kafka/coordinator/transaction/TransactionLogTest.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala index 03fca4421359c..80af1f6ba98a9 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala @@ -191,7 +191,7 @@ class TransactionLogTest { ) ) - // create TransactionLogValue.PartitionsSchema with tagged fields + // Create TransactionLogValue.PartitionsSchema with tagged fields val txnPartitions = new Struct(futurePartitionsSchema) txnPartitions.set("topic", "topic") txnPartitions.set("partition_ids", Array(Integer.valueOf(1))) @@ -216,7 +216,7 @@ class TransactionLogTest { ) ) - // create TransactionLogValue with tagged fields + // Create TransactionLogValue with tagged fields val transactionLogValue = new Struct(futureTransactionLogValueSchema) transactionLogValue.set("producer_id", 1000L) transactionLogValue.set("producer_epoch", 100.toShort) From 533363db2469bb8fc2d95ba53881b5c3764f735a Mon Sep 17 00:00:00 2001 From: David Jacot Date: Tue, 18 Apr 2023 09:53:16 +0200 Subject: [PATCH 8/9] fix typo --- .../unit/kafka/coordinator/group/GroupMetadataManagerTest.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index a8da3928dd26c..607dce890fd35 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala @@ -2561,7 +2561,7 @@ class GroupMetadataManagerTest { ) ) - // create OffsetCommitValue with tagged fields + // Create OffsetCommitValue with tagged fields val offsetCommit = new Struct(futureOffsetCommitSchema) offsetCommit.set("offset", 1000L) offsetCommit.set("leader_epoch", 100) From 52a14ebbb10e725fdb27d319272b2052b32fd0c9 Mon Sep 17 00:00:00 2001 From: David Jacot Date: Tue, 18 Apr 2023 10:19:04 +0200 Subject: [PATCH 9/9] scala 2.12 fixes --- .../coordinator/group/GroupMetadataManagerTest.scala | 12 ++++++------ .../coordinator/transaction/TransactionLogTest.scala | 8 ++++---- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala index 607dce890fd35..eed33c4de752a 100644 --- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala @@ -2556,8 +2556,8 @@ class GroupMetadataManagerTest { 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, "") + Int.box(0), new Field("offset_foo", Type.STRING, ""), + Int.box(1), new Field("offset_bar", Type.INT32, "") ) ) @@ -2608,8 +2608,8 @@ class GroupMetadataManagerTest { 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, "") + Int.box(0), new Field("member_foo", Type.STRING, ""), + Int.box(1), new Field("member_foo", Type.INT32, "") ) ) @@ -2623,8 +2623,8 @@ class GroupMetadataManagerTest { 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, "") + Int.box(0), new Field("group_foo", Type.STRING, ""), + Int.box(1), new Field("group_bar", Type.INT32, "") ) ) diff --git a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala index 80af1f6ba98a9..f0c883ff21b13 100644 --- a/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala +++ b/core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala @@ -186,8 +186,8 @@ class TransactionLogTest { 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, "") + Int.box(0), new Field("partition_foo", Type.STRING, ""), + Int.box(1), new Field("partition_foo", Type.INT32, "") ) ) @@ -211,8 +211,8 @@ class TransactionLogTest { 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, "") + Int.box(0), new Field("txn_foo", Type.STRING, ""), + Int.box(1), new Field("txn_bar", Type.INT32, "") ) )