Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dajac merged 10 commits into from
Apr 18, 2023
Merged

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

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
435853b
bump Value records to flexible versions
jeffkbkim Apr 7, 2023
28710d4
first flexible version comment
jeffkbkim Apr 10, 2023
d74107a
address comments
jeffkbkim Apr 13, 2023
173802a
remove test schemas
jeffkbkim Apr 14, 2023
a091b96
address comments
jeffkbkim Apr 17, 2023
2e82907
add deserialize highest supported txn log value test
jeffkbkim Apr 17, 2023
1a78f4e
nit
jeffkbkim Apr 17, 2023
533363d
fix typo
dajac Apr 18, 2023
52a14eb
scala 2.12 fixes
dajac Apr 18, 2023
91fbf91
Merge remote-tracking branch 'upstream/trunk' into jeffkbkim-kafka-14…
dajac Apr 18, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 13 additions & 11 deletions core/src/main/resources/common/message/TransactionLogValue.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,27 +13,29 @@
// See the License for the specific language governing permissions and
// limitations under the License.

// Version 1 is the first flexible version.
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
// KIP-915: bumping the version will no longer make this record backward compatible.
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
{
"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"}
]
}
8 changes: 6 additions & 2 deletions core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
MessageUtil.toVersionPrefixedBytes(version, new OffsetCommitValue()
.setOffset(offsetAndMetadata.offset)
.setMetadata(offsetAndMetadata.metadata)
Expand Down Expand Up @@ -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
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated

MessageUtil.toVersionPrefixedBytes(version, new GroupMetadataValue()
.setProtocolType(groupMetadata.protocolType.getOrElse(""))
Expand Down
4 changes: 3 additions & 1 deletion core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
new TransactionLogValue()
.setProducerId(txnMetadata.producerId)
.setProducerEpoch(txnMetadata.producerEpoch)
Expand Down
32 changes: 31 additions & 1 deletion core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -2466,6 +2467,34 @@ class GroupMetadataManagerTest {
verifySerde(version)
}

@Test
Comment thread
jeffkbkim marked this conversation as resolved.
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)
}
}

Comment thread
dajac marked this conversation as resolved.
@Test
def testLoadOffsetsWithEmptyControlBatch(): Unit = {
val groupMetadataTopicPartition = groupTopicPartition
Expand Down Expand Up @@ -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()))
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
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)
Expand Down
9 changes: 9 additions & 0 deletions core/src/test/scala/unit/kafka/coordinator/transaction/TransactionLogTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -135,4 +137,11 @@ 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(TransactionLogValue.HIGHEST_SUPPORTED_NON_FLEXIBLE_VERSION, txnLogValueBuffer.getShort)
}

}
8 changes: 8 additions & 0 deletions generator/src/main/java/org/apache/kafka/message/SchemaGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
}
}
6 changes: 4 additions & 2 deletions group-coordinator/src/main/resources/common/message/GroupMetadataValue.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.

// Version 4 is the first flexible version.
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
// KIP-915: bumping the version will no longer make this record backward compatible.
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
{
"type": "data",
"name": "GroupMetadataValue",
"validVersions": "0-3",
"flexibleVersions": "none",
"validVersions": "0-4",
Comment thread
dajac marked this conversation as resolved.
Comment thread
dajac marked this conversation as resolved.
"flexibleVersions": "4+",
"fields": [
{ "name": "protocolType", "versions": "0+", "type": "string"},
{ "name": "generation", "versions": "0+", "type": "int32" },
Expand Down
6 changes: 4 additions & 2 deletions group-coordinator/src/main/resources/common/message/OffsetCommitValue.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.

// Version 4 is the first flexible version.
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
// KIP-915: bumping the version will no longer make this record backward compatible.
Comment thread
jeffkbkim marked this conversation as resolved.
Outdated
{
"type": "data",
"name": "OffsetCommitValue",
"validVersions": "0-3",
"flexibleVersions": "none",
"validVersions": "0-4",
Comment thread
dajac marked this conversation as resolved.
"flexibleVersions": "4+",
"fields": [
{ "name": "offset", "type": "int64", "versions": "0+" },
{ "name": "leaderEpoch", "type": "int32", "versions": "3+", "default": -1, "ignorable": true},
Expand Down