You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jolshan (via GitHub)" <gi...@apache.org> on 2024/06/03 23:11:37 UTC
[PR] KAFKA-16192: Introduce usage of flexible records to coordinators [kafka]
jolshan opened a new pull request, #16183:
URL: https://github.com/apache/kafka/pull/16183
This change includes adding transaction.version (part of KIP-1022)
New transaction version 1 is introduced to support writing flexible fields in transaction state log messages.
Transaction version 2 is created in anticipation for further KIP-890 changes.
Neither are made production ready. Tests for the new transaction version and new MV are created.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1683614145
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,11 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ IBP_3_9_IV2(23, "3.9", "IV2", false),
Review Comment:
I've updated the PR to include the change for broker registration
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684690411
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
The main thing is that we need to exclude range 0-0 for supported features in ApiVersionResponse too. Colin's code in ApiVersionsResponse doesn't seem to do what you are doing here for BrokerRegistrationRequest by excluding range 0-0 for supported features. It only changes v0 to v1 for older requests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1674294082
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,11 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ IBP_3_9_IV2(23, "3.9", "IV2", false),
Review Comment:
> I planned to make the update where we don't send features as part of the registration/api versions if the version range is 0-0.
Is that for the old versions of the requests or both the old and the latest version?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #16183:
URL: https://github.com/apache/kafka/pull/16183#issuecomment-2221692444
I think I also need to fix the case where max version is 0 as well. 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1628482592
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionLog.scala:
##########
@@ -75,9 +76,9 @@ object TransactionLog {
.setPartitionIds(partitions.map(tp => Integer.valueOf(tp.partition)).toList.asJava)
}.toList.asJava
- // Serialize with the highest supported non-flexible version
- // until a tagged field is introduced or the version is bumped.
- MessageUtil.toVersionPrefixedBytes(0,
+ val version: Short =
Review Comment:
Could we retain and extend the comment to explain the new logic?
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -205,7 +205,10 @@ public enum MetadataVersion {
IBP_3_8_IV0(20, "3.8", "IV0", true),
// Introduce version 1 of the GroupVersion feature (KIP-848).
- IBP_4_0_IVO(21, "4.0", "IV0", false);
+ IBP_4_0_IVO(21, "4.0", "IV0", false),
Review Comment:
Not introduced in this change, but looks like IBP_4_0_IVO uses letter "O" at the end which is different from other versions that use number "0".
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -205,7 +205,10 @@ public enum MetadataVersion {
IBP_3_8_IV0(20, "3.8", "IV0", true),
// Introduce version 1 of the GroupVersion feature (KIP-848).
- IBP_4_0_IVO(21, "4.0", "IV0", false);
+ IBP_4_0_IVO(21, "4.0", "IV0", false),
+
+ // Introduce version 1 of the TransactionVersion feature (KIP-890).
+ IBP_4_0_IV1(22, "4.0", "IV1", false);
Review Comment:
Do we need a separate MV version? We could bind TV to IBP_4_0_IVO. It's still in the future, so this should be ok as far as the bootstrap is concerned.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1683517785
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,11 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ IBP_3_9_IV2(23, "3.9", "IV2", false),
Review Comment:
Should we add a new MV just for bootstrapping a new feature or just piggyback on an existing one (e.g., IBP_3_9_IV1)?
> I planned to make the update where we don't send features as part of the registration/api versions if the version range is 0-0.
Colin's PR only skipped a finalized version 0 feature in ApiResponse. Do you plan to do that for supported features for registration/api? Range 0-0 can now happen since some of the v1 features are not stable yet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684962284
##########
core/src/test/scala/unit/kafka/server/BrokerFeaturesTest.scala:
##########
@@ -95,10 +97,20 @@ class BrokerFeaturesTest {
val expectedFeatures = Map[String, Short](
MetadataVersion.FEATURE_NAME -> MetadataVersion.latestTesting().featureLevel(),
+ ServerFeatures.TRANSACTION_VERSION.featureName() -> ServerFeatures.TRANSACTION_VERSION.latestTesting(),
"kraft.version" -> 0,
"test_feature_1" -> 4,
"test_feature_2" -> 3,
"test_feature_3" -> 7)
assertEquals(expectedFeatures, brokerFeatures.defaultFinalizedFeatures)
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def ensureDefaultSupportedFeaturesRangeNotZeroZero(unstableVersionsEnabled: Boolean): Unit = {
+ val brokerFeatures = BrokerFeatures.createDefault(unstableVersionsEnabled)
+ brokerFeatures.supportedFeatures.features().values().forEach { supportedVersionRange =>
+ assertFalse(supportedVersionRange.min() == 0 && supportedVersionRange.max() == 0)
Review Comment:
Sure. I can do that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684642190
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,12 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ // Bootstrapping transaction version 2
Review Comment:
I think I am going to move this to 4.0 anyway, so we can move it to an existing one there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684705617
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
From my point of view:
1. We should never report a `0-0` supported versions range in an RPC. There is no point! This applies to both `ApiVersionsResponse` and `BrokerRegistrationRequest`.
2. Obviously a `1-0` range is even worse, being incorrect as well as useless.
3. Having a HashMap or something that contains a `0-0` range is fine, and may simplify the code in some ways. For example if we have code that changes the maximum supported range based on a config, it would probably be annoying to have to mutate the HashMap or use a different hash map if that config was on or off.
I don't feel really strongly about the third point, but I suspect that the code will be messier if you try to remove features with supported version 0-0 from the internal data structures than if you don't.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684712684
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
My proposal was that in defaultSupportedFeatues, we iterate over the features and add them to a map. Instead of just adding them all, we check if the range is 0-0. If so, we don't add them. Thus, we don't need specific logic in the apis to remove the feature later.
The only concern is if somehow the features map expected a feature to exist. Not sure I follow mutating if the config is on or off. I assume we can't change the config to enable unstable features dynamically. (If we did, we'd have to mutate the map anyway)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1685017655
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
@jolshan : If you think it's easier to just omit them from the map, that's fine.
I still think we should be checking that these things don't exist later on (and for example, throw an exception if they do) to avoid regressions if someone changes something (as has been happening a lot lately. :) )
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1674310171
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,11 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ IBP_3_9_IV2(23, "3.9", "IV2", false),
Review Comment:
I think we can just do it regardless of the version of the request. I was planning to do it in Quorum/BrokerFeatures files, which would just be the controller/broker's own view of supported features. Alternatively I can do it in the request.
cc: @cmccabe
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1638849949
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -205,7 +205,10 @@ public enum MetadataVersion {
IBP_3_8_IV0(20, "3.8", "IV0", true),
// Introduce version 1 of the GroupVersion feature (KIP-848).
- IBP_4_0_IVO(21, "4.0", "IV0", false);
+ IBP_4_0_IVO(21, "4.0", "IV0", false),
+
+ // Introduce version 1 of the TransactionVersion feature (KIP-890).
+ IBP_4_0_IV1(22, "4.0", "IV1", false);
Review Comment:
Sounds good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #16183:
URL: https://github.com/apache/kafka/pull/16183#issuecomment-2239559725
Will look at the apiversions test failures 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on PR #16183:
URL: https://github.com/apache/kafka/pull/16183#issuecomment-2245675058
@jolshan : Thanks for the updated PR. The following test failure seems related to the PR.
```
[2024-07-23T02:46:53.731Z]
[2024-07-23T02:46:53.731Z] Gradle Test Run :metadata:test > Gradle Test Executor 88 > QuorumFeaturesTest > testDefaultFeatureMap() STARTED
[2024-07-23T02:46:53.731Z] org.apache.kafka.controller.QuorumFeaturesTest.testDefaultFeatureMap() failed, log available in /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-16183/metadata/build/reports/testOutput/org.apache.kafka.controller.QuorumFeaturesTest.testDefaultFeatureMap().test.stdout
[2024-07-23T02:46:53.731Z]
[2024-07-23T02:46:53.731Z] Gradle Test Run :metadata:test > Gradle Test Executor 88 > QuorumFeaturesTest > testDefaultFeatureMap() FAILED
[2024-07-23T02:46:53.731Z] org.opentest4j.AssertionFailedError: expected: <{kraft.version=0, transaction.version=0, metadata.version=1-20}> but was: <{metadata.version=1-20}>
[2024-07-23T02:46:53.731Z] at org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
[2024-07-23T02:46:53.731Z] at org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
[2024-07-23T02:46:53.731Z] at org.junit.jupiter.api.AssertEquals.failNotEqual(AssertEquals.java:197)
[2024-07-23T02:46:53.731Z] at org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:182)
[2024-07-23T02:46:53.731Z] at org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:177)
[2024-07-23T02:46:53.731Z] at org.junit.jupiter.api.Assertions.assertEquals(Assertions.java:1145)
[2024-07-23T02:46:53.731Z] at org.apache.kafka.controller.QuorumFeaturesTest.testDefaultFeatureMap(QuorumFeaturesTest.java:69)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684693402
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
Yeah. It is already excluded from Colin's commit. If we want to share the logic we can move it to Broker/QuorumFeatures, but I wasn't sure if there was a reason why Colin did it in the response before.
cc: @cmccabe
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1631735975
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -205,7 +205,10 @@ public enum MetadataVersion {
IBP_3_8_IV0(20, "3.8", "IV0", true),
// Introduce version 1 of the GroupVersion feature (KIP-848).
- IBP_4_0_IVO(21, "4.0", "IV0", false);
+ IBP_4_0_IVO(21, "4.0", "IV0", false),
+
+ // Introduce version 1 of the TransactionVersion feature (KIP-890).
+ IBP_4_0_IV1(22, "4.0", "IV1", false);
Review Comment:
Sorry I misunderstood your question. I thought you were referring to the TVs being the same. I think we should keep separate for now since I assume these will be released at different times. But don't feel super strongly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1638510363
##########
server-common/src/main/java/org/apache/kafka/server/common/TransactionVersion.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.server.common;
+
+import java.util.Collections;
+import java.util.Map;
+
+public enum TransactionVersion implements FeatureVersion {
+
+ // Version 1 enables flexible transactional state records. (KIP-890)
+ TV_1(1, MetadataVersion.IBP_4_0_IV1, Collections.emptyMap()),
+ // Version 2 enables epoch bump per transaction and optimizations. (KIP-890)
+ TV_2(2, MetadataVersion.IBP_4_0_IV1, Collections.emptyMap());
+
+ public static final String FEATURE_NAME = "transaction.version";
+
+ private final short featureLevel;
+ private final MetadataVersion bootstrapMetadataVersion;
+ private final Map<String, Short> dependencies;
+
+ TransactionVersion(
+ int featureLevel,
+ MetadataVersion bootstrapMetadataVersion,
+ Map<String, Short> dependencies
Review Comment:
I need to figure out how to fix this in my IDE 😵
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1638509930
##########
server-common/src/main/java/org/apache/kafka/server/common/TransactionVersion.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.server.common;
+
+import java.util.Collections;
+import java.util.Map;
+
+public enum TransactionVersion implements FeatureVersion {
+
+ // Version 1 enables flexible transactional state records. (KIP-890)
+ TV_1(1, MetadataVersion.IBP_4_0_IV1, Collections.emptyMap()),
Review Comment:
This was a discussion I had with @cmccabe. I think the idea was that even if we want to disable transaction version, we may still want to keep our flexible records due to the compatibility reasons. I know Jeff worked on that KIP to ignore the fields, but this was an extra safeguard.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1685037958
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
I'm happy to try to make things safer. Can you point me to an example of where you would expect to check if things exist? Do you mean like when building the request? Or something else?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684885144
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
Pushed code with this example.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #16183:
URL: https://github.com/apache/kafka/pull/16183
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1673242703
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,11 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ IBP_3_9_IV2(23, "3.9", "IV2", false),
Review Comment:
Sounds good. I also need to make a few other changes when I rebase, so be on the lookout for more soon :)
I planned to make the update where we don't send features as part of the registration/api versions if the version range is 0-0. I need to do this because Colin's fix only changes the min version. It is safe because a version range 0-0 is essentially the same as not supporting the feature. We will have it here until the MV is marked as production ready.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1638158613
##########
server-common/src/main/java/org/apache/kafka/server/common/TransactionVersion.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.server.common;
+
+import java.util.Collections;
+import java.util.Map;
+
+public enum TransactionVersion implements FeatureVersion {
+
+ // Version 1 enables flexible transactional state records. (KIP-890)
+ TV_1(1, MetadataVersion.IBP_4_0_IV1, Collections.emptyMap()),
Review Comment:
I wonder why we need a version to enable flexible records. Couldn't we use the version supporting it when we enable version 2? I may have missed something.
##########
server-common/src/main/java/org/apache/kafka/server/common/TransactionVersion.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.server.common;
+
+import java.util.Collections;
+import java.util.Map;
+
+public enum TransactionVersion implements FeatureVersion {
+
+ // Version 1 enables flexible transactional state records. (KIP-890)
+ TV_1(1, MetadataVersion.IBP_4_0_IV1, Collections.emptyMap()),
+ // Version 2 enables epoch bump per transaction and optimizations. (KIP-890)
+ TV_2(2, MetadataVersion.IBP_4_0_IV1, Collections.emptyMap());
+
+ public static final String FEATURE_NAME = "transaction.version";
+
+ private final short featureLevel;
+ private final MetadataVersion bootstrapMetadataVersion;
+ private final Map<String, Short> dependencies;
+
+ TransactionVersion(
+ int featureLevel,
+ MetadataVersion bootstrapMetadataVersion,
+ Map<String, Short> dependencies
Review Comment:
nit: Indentation seems to be incorrect here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1639981075
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -618,7 +624,7 @@ class TransactionStateManager(brokerId: Int,
// generate the message for this transaction metadata
val keyBytes = TransactionLog.keyToBytes(transactionalId)
- val valueBytes = TransactionLog.valueToBytes(newMetadata)
+ val valueBytes = TransactionLog.valueToBytes(newMetadata, usesFlexibleRecords())
Review Comment:
Functionally there is no difference now, so a test didn't come to my mind easily. I guess the best thing is to just look at the version of the record generated. I can add that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1629730668
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -205,7 +205,10 @@ public enum MetadataVersion {
IBP_3_8_IV0(20, "3.8", "IV0", true),
// Introduce version 1 of the GroupVersion feature (KIP-848).
- IBP_4_0_IVO(21, "4.0", "IV0", false);
+ IBP_4_0_IVO(21, "4.0", "IV0", false),
+
+ // Introduce version 1 of the TransactionVersion feature (KIP-890).
+ IBP_4_0_IV1(22, "4.0", "IV1", false);
Review Comment:
I don't think we need it if we plan to make them both production ready at the same time
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1631735975
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -205,7 +205,10 @@ public enum MetadataVersion {
IBP_3_8_IV0(20, "3.8", "IV0", true),
// Introduce version 1 of the GroupVersion feature (KIP-848).
- IBP_4_0_IVO(21, "4.0", "IV0", false);
+ IBP_4_0_IVO(21, "4.0", "IV0", false),
+
+ // Introduce version 1 of the TransactionVersion feature (KIP-890).
+ IBP_4_0_IV1(22, "4.0", "IV1", false);
Review Comment:
Sorry I misunderstood your question. I thought you were referring to the TVs being the same. I think we should keep separate from group version for now since I assume these will be released at different times. But don't feel super strongly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1631716436
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -205,7 +205,10 @@ public enum MetadataVersion {
IBP_3_8_IV0(20, "3.8", "IV0", true),
// Introduce version 1 of the GroupVersion feature (KIP-848).
- IBP_4_0_IVO(21, "4.0", "IV0", false);
+ IBP_4_0_IVO(21, "4.0", "IV0", false),
+
+ // Introduce version 1 of the TransactionVersion feature (KIP-890).
+ IBP_4_0_IV1(22, "4.0", "IV1", false);
Review Comment:
I think for now the current plan is to release both in 4.0, we could reflect this in the code (if the plans change we could change the code accordingly).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #16183:
URL: https://github.com/apache/kafka/pull/16183#issuecomment-2249007740
Some of the gradle task executors failed so I will run again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1673162688
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,11 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ IBP_3_9_IV2(23, "3.9", "IV2", false),
Review Comment:
Could we add a comment that this is for bootstrapping TV?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1683242528
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,11 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ IBP_3_9_IV2(23, "3.9", "IV2", false),
Review Comment:
Looks like Colin already pushed this as part of the kraft.version PR. I will just resolve the conflicts and update the comment you mention.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684690411
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
The main thing is that we need to exclude range 0-0 for supported features in ApiVersionResponse too. Colin's code in ApiVersionsResponse doesn't seem to do what you are doing here for BrokerRegistrationRequest by excluding range 0-0 for supported features. It only changes minVersion v0 to v1 for older requests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684712684
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
My proposal was that in defaultSupportedFeatues, we iterate over the features and add them to a map. Instead of just adding them all, we check if the range is 0-0. If so, we don't add them. (No removal needed) Thus, we don't need specific logic in the apis to remove the feature later.
The only concern is if somehow the features map expected a feature to exist. Not sure I follow mutating if the config is on or off. I assume we can't change the config to enable unstable features dynamically. (If we did, we'd have to mutate the map anyway)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684718844
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
Code I refer to is:
https://github.com/apache/kafka/blob/208bb1fbc22d32e2dc70cc9924be085c62997c65/metadata/src/main/java/org/apache/kafka/controller/QuorumFeatures.java#L65
https://github.com/apache/kafka/blob/208bb1fbc22d32e2dc70cc9924be085c62997c65/core/src/main/scala/kafka/server/BrokerFeatures.scala#L93
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684941966
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,9 +217,10 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
- // Introduce version 1 of the GroupVersion feature (KIP-848).
+ // Introduce version 1 of the GroupVersion feature (KIP-848) and transaction versions 1 and 2 (KIP-890)
IBP_4_0_IV0(23, "4.0", "IV0", false);
+
Review Comment:
extra new line
##########
core/src/test/scala/unit/kafka/server/BrokerFeaturesTest.scala:
##########
@@ -95,10 +97,20 @@ class BrokerFeaturesTest {
val expectedFeatures = Map[String, Short](
MetadataVersion.FEATURE_NAME -> MetadataVersion.latestTesting().featureLevel(),
+ ServerFeatures.TRANSACTION_VERSION.featureName() -> ServerFeatures.TRANSACTION_VERSION.latestTesting(),
"kraft.version" -> 0,
"test_feature_1" -> 4,
"test_feature_2" -> 3,
"test_feature_3" -> 7)
assertEquals(expectedFeatures, brokerFeatures.defaultFinalizedFeatures)
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def ensureDefaultSupportedFeaturesRangeNotZeroZero(unstableVersionsEnabled: Boolean): Unit = {
+ val brokerFeatures = BrokerFeatures.createDefault(unstableVersionsEnabled)
+ brokerFeatures.supportedFeatures.features().values().forEach { supportedVersionRange =>
+ assertFalse(supportedVersionRange.min() == 0 && supportedVersionRange.max() == 0)
Review Comment:
Perhaps it's better to only test `supportedVersionRange.max() == 0` since we want to disallow ranges like (1,0) too? If so, we want to adjust the method name accordingly. Ditto for a similar test in [QuorumFeaturesTest.java](https://github.com/apache/kafka/pull/16183/commits/dc34d935673d6c4eb9373dfea092795ce99a1d8e#diff-7160ed0388ecc53d2af1e7b722f59a25f95475cd36baf4a14e0c038a5473ab97).
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,9 +217,10 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
- // Introduce version 1 of the GroupVersion feature (KIP-848).
+ // Introduce version 1 of the GroupVersion feature (KIP-848) and transaction versions 1 and 2 (KIP-890)
Review Comment:
Since group/txn features don't depend on this MV, perhaps changing the comment to sth like the following?
bootstrap metadata version of version 1 of the GroupVersion feature (KIP-848) and transaction versions 1 and 2 (KIP-890)
##########
core/src/main/scala/kafka/server/BrokerFeatures.scala:
##########
@@ -90,13 +90,16 @@ object BrokerFeatures extends Logging {
} else {
MetadataVersion.latestProduction.featureLevel
}))
- PRODUCTION_FEATURES.forEach { feature => features.put(feature.featureName,
- new SupportedVersionRange(0,
- if (unstableFeatureVersionsEnabled) {
- feature.latestTesting
- } else {
- feature.latestProduction
- }))
+ PRODUCTION_FEATURES.forEach {
+ feature => {
Review Comment:
We could get rid of the {} here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1688658774
##########
metadata/src/test/java/org/apache/kafka/controller/QuorumFeaturesTest.java:
##########
@@ -58,10 +61,13 @@ public void testDefaultFeatureMap() {
MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel(),
MetadataVersion.LATEST_PRODUCTION.featureLevel()));
for (Features feature : Features.PRODUCTION_FEATURES) {
- expectedFeatures.put(feature.featureName(), VersionRange.of(
- 0,
- feature.defaultValue(MetadataVersion.LATEST_PRODUCTION)
- ));
+ short maxVersion = feature.defaultValue(MetadataVersion.LATEST_PRODUCTION);
+ if (maxVersion > 0) {
+ expectedFeatures.put(feature.featureName(), VersionRange.of(
+ feature.minimumProduction(),
+ maxVersion
+ ));
Review Comment:
I would say maybe if someone changes in the file, we would catch that this is desired behavior and the test would fail?
I can add a comment if that helps.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1688638549
##########
metadata/src/test/java/org/apache/kafka/controller/QuorumFeaturesTest.java:
##########
@@ -58,10 +61,13 @@ public void testDefaultFeatureMap() {
MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel(),
MetadataVersion.LATEST_PRODUCTION.featureLevel()));
for (Features feature : Features.PRODUCTION_FEATURES) {
- expectedFeatures.put(feature.featureName(), VersionRange.of(
- 0,
- feature.defaultValue(MetadataVersion.LATEST_PRODUCTION)
- ));
+ short maxVersion = feature.defaultValue(MetadataVersion.LATEST_PRODUCTION);
+ if (maxVersion > 0) {
+ expectedFeatures.put(feature.featureName(), VersionRange.of(
+ feature.minimumProduction(),
+ maxVersion
+ ));
Review Comment:
The code in the test is pretty much the same as the actual code. Does this and the next test add much value?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #16183:
URL: https://github.com/apache/kafka/pull/16183#issuecomment-2221688829
Yeehaw time to rebase
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1639511150
##########
core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala:
##########
@@ -618,7 +624,7 @@ class TransactionStateManager(brokerId: Int,
// generate the message for this transaction metadata
val keyBytes = TransactionLog.keyToBytes(transactionalId)
- val valueBytes = TransactionLog.valueToBytes(newMetadata)
+ val valueBytes = TransactionLog.valueToBytes(newMetadata, usesFlexibleRecords())
Review Comment:
Should we add a test to ensure that this new logic works as expected?
##########
server-common/src/main/java/org/apache/kafka/server/common/TransactionVersion.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.server.common;
+
+import java.util.Collections;
+import java.util.Map;
+
+public enum TransactionVersion implements FeatureVersion {
+
+ // Version 1 enables flexible transactional state records. (KIP-890)
+ TV_1(1, MetadataVersion.IBP_4_0_IV1, Collections.emptyMap()),
Review Comment:
Understood. As you said, this is not strictly necessary but it does not hurt.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684690411
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
The main thing is that we need to exclude range 0-0 for supported features in ApiVersionResponse too. It would be better if exclusion logic is shared between ApiVersionResponse and BrokerRegistrationRequest.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1683529869
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,11 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ IBP_3_9_IV2(23, "3.9", "IV2", false),
Review Comment:
Hmmm-- I thought Colin's PR would cover what I need to cover since I didn't think his MV was stable yet either
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684620247
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
We should do the same for supported features in ApiVersionResponse in the following code in BrokerFeatures, right? Also, to avoid duplicating the code, perhaps it's better to do the filtering logic in BrokerFeatures?
```
def defaultSupportedFeatures(unstableFeatureVersionsEnabled: Boolean): Features[SupportedVersionRange] = {
val features = new util.HashMap[String, SupportedVersionRange]()
features.put(MetadataVersion.FEATURE_NAME,
new SupportedVersionRange(
MetadataVersion.MINIMUM_KRAFT_VERSION.featureLevel(),
if (unstableFeatureVersionsEnabled) {
MetadataVersion.latestTesting.featureLevel
} else {
MetadataVersion.latestProduction.featureLevel
}))
PRODUCTION_FEATURES.forEach { feature => features.put(feature.featureName,
new SupportedVersionRange(0,
if (unstableFeatureVersionsEnabled) {
feature.latestTesting
} else {
feature.latestProduction
}))
}
Features.supportedFeatures(features)
}
```
##########
core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala:
##########
@@ -99,6 +101,27 @@ class BrokerLifecycleManagerTest {
}
}
+ @Test
+ def testFeatureRangeMaxZeroNotIncludedInRegistration(): Unit = {
+ val context = new RegistrationTestContext(configProperties)
+ manager = new BrokerLifecycleManager(context.config, context.time, "successful-registration-", isZkBroker = false, Set(Uuid.fromString("gCpDJgRlS2CBCpxoP2VMsQ")))
+ val controllerNode = new Node(3000, "localhost", 8021)
+ context.controllerNodeProvider.node.set(controllerNode)
+
+ val features = new util.HashMap[String, VersionRange]()
+ features.put(Features.TEST_VERSION.featureName(), VersionRange.of(0, 0))
+ features.put(Features.KRAFT_VERSION.featureName(), VersionRange.of(0, 1))
+
+ manager.start(() => context.highestMetadataOffset.get(),
+ context.mockChannelManager, context.clusterId, context.advertisedListeners,
+ features, OptionalLong.of(10L))
+ TestUtils.retry(60000) {
+ assertEquals(1, context.mockChannelManager.unsentQueue.size)
+ // Only the feature with a range that is not 0-0 is included.
+ assertEquals(1, context.mockChannelManager.unsentQueue.getFirst.request.build().asInstanceOf[BrokerRegistrationRequest].data().features().size())
Review Comment:
Could we further verify that the only feature is KRAFT_VERSION?
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -217,8 +217,12 @@ public enum MetadataVersion {
// Add ELR related supports (KIP-966).
IBP_3_9_IV1(22, "3.9", "IV1", true),
+ // Bootstrapping transaction version 2
Review Comment:
Should we add a new MV just for bootstrapping a new feature or just piggyback on an existing one (e.g., IBP_3_9_IV1)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684645307
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
I had considered putting it in BrokerFeatures/QuorumFeatures, but decided against it.
Are you suggesting that I remove how Colin wrote the code (in the ApiVersionsResponse itself).
I was trying to avoid removing his code, but if you think it is better that way I can do it. (I will also need to move the test I wrote elsewhere)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #16183:
URL: https://github.com/apache/kafka/pull/16183#issuecomment-2239635110
Given that Kraft version starts at 0, I guess we should have all versions start a zero for consistency. I also noticed one test asserts all features with version zero bootstrap with minimum kraft version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1684705617
##########
core/src/main/scala/kafka/server/BrokerLifecycleManager.scala:
##########
@@ -381,10 +381,12 @@ class BrokerLifecycleManager(
private def sendBrokerRegistration(): Unit = {
val features = new BrokerRegistrationRequestData.FeatureCollection()
_supportedFeatures.asScala.foreach {
- case (name, range) => features.add(new BrokerRegistrationRequestData.Feature().
+ // Do not include features with the range 0-0.
+ case (name, range) if range.max() > 0 => features.add(new BrokerRegistrationRequestData.Feature().
Review Comment:
Can you be clearer about what you're proposing?
From my point of view:
1. We should never report a `0-0` supported versions range in an RPC. There is no point! This applies to both `ApiVersionsResponse` and `BrokerRegistrationRequest`.
2. Obviously a `1-0` range is even worse, being incorrect as well as useless.
3. Having a HashMap or something that contains a `0-0` range is fine, and may simplify the code in some ways. For example if we have code that changes the maximum supported range based on a config, it would probably be annoying to have to mutate the HashMap or use a different hash map if that config was on or off.
I don't feel really strongly about the third point, but I suspect that the code will be messier if you try to remove features with supported version 0-0 from the internal data structures than if you don't.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16192: Introduce transaction.version and usage of flexible records to coordinators [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #16183:
URL: https://github.com/apache/kafka/pull/16183#discussion_r1629730197
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -205,7 +205,10 @@ public enum MetadataVersion {
IBP_3_8_IV0(20, "3.8", "IV0", true),
// Introduce version 1 of the GroupVersion feature (KIP-848).
- IBP_4_0_IVO(21, "4.0", "IV0", false);
+ IBP_4_0_IVO(21, "4.0", "IV0", false),
Review Comment:
Hehe. This will be fixed when I fix conflicts. But good to keep in mind 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org