You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ijuma (via GitHub)" <gi...@apache.org> on 2025/02/09 17:04:57 UTC
[PR] KAFKA-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
ijuma opened a new pull request, #18845:
URL: https://github.com/apache/kafka/pull/18845
TODO:
1. Review the PR carefully
2. Update upgrade notes
3. Explain the changes in more detail
### 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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950084465
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java:
##########
@@ -128,18 +126,6 @@ public void testReassignment() throws Exception {
executeAndVerifyReassignment();
}
- @ClusterTests({
- @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, metadataVersion = IBP_3_3_IV0)
- })
- public void testReassignmentWithAlterPartitionDisabled() throws Exception {
- // Test reassignment when the IBP is on an older version which does not use
- // the `AlterPartition` API. In this case, the controller will register individual
- // watches for each reassigning partition so that the reassignment can be
- // completed as soon as the ISR is expanded.
Review Comment:
> you're saying we can remove this test and we don't need any other follow-up since the comment was for the zk controller. Right?
we are on the same page!
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1957186628
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -495,7 +495,9 @@ BrokerFeature processRegistrationFeature(
FinalizedControllerFeatures finalizedFeatures,
BrokerRegistrationRequestData.Feature feature
) {
- int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ? 1 : 0; // The default value for MetadataVersion is 1 not 0.
+ // The default value for MetadataVersion changes over time while other features start at `0`
+ int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ?
+ MetadataVersion.MINIMUM_VERSION.featureLevel() : 0;
Review Comment:
I made this change.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950994148
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -368,7 +320,8 @@ public static MetadataVersion fromVersionString(String versionString) {
key = String.join(".", Arrays.copyOfRange(versionSegments, 0, numSegments));
}
return Optional.ofNullable(IBP_VERSIONS.get(key)).orElseThrow(() ->
- new IllegalArgumentException("Version " + versionString + " is not a valid version")
+ new IllegalArgumentException("Version " + versionString + " is not a valid version. The minimum version is " + MINIMUM_VERSION
+ + " and the maximum version is " + latestTesting())
Review Comment:
As you said, that seems to be inconsistent with the docs for `StorageTool` - perhaps we should file an issue to fix that in its own PR (assuming it's not a regression in this one, but I don't think the relevant code was changed).
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953309570
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1650,13 +1642,7 @@ private QuorumController(
*/
private void registerWriteNoOpRecord(long maxIdleIntervalNs) {
periodicControl.registerTask(new PeriodicTask("writeNoOpRecord",
- () -> {
- ArrayList<ApiMessageAndVersion> records = new ArrayList<>(1);
- if (featureControl.metadataVersion().isNoOpRecordSupported()) {
- records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
- }
- return ControllerResult.of(records, false);
- },
+ () -> ControllerResult.of(List.of(new ApiMessageAndVersion(new NoOpRecord(), (short) 0)), false),
Review Comment:
The javadoc above on `if the MetadataVersion supports it.` needs to be changed too.
##########
metadata/src/test/java/org/apache/kafka/image/MetadataVersionChangeTest.java:
##########
@@ -30,31 +32,31 @@
@Timeout(value = 40)
public class MetadataVersionChangeTest {
- private static final MetadataVersionChange CHANGE_3_0_IV1_TO_3_3_IV0 =
- new MetadataVersionChange(IBP_3_0_IV1, IBP_3_3_IV0);
+ private static final MetadataVersionChange CHANGE_MINUMUM_TO_LATEST =
+ new MetadataVersionChange(MetadataVersion.MINIMUM_VERSION, MetadataVersion.latestProduction());
- private static final MetadataVersionChange CHANGE_3_3_IV0_TO_3_0_IV1 =
- new MetadataVersionChange(IBP_3_3_IV0, IBP_3_0_IV1);
+ private static final MetadataVersionChange CHANGE_3_6_IV0_TO_3_3_IV3 =
Review Comment:
Could we change this to use MetadataVersion.MINIMUM_VERSION and MetadataVersion.latestProduction() too?
##########
metadata/src/main/java/org/apache/kafka/image/FeaturesDelta.java:
##########
@@ -58,7 +59,14 @@ public void finishSnapshot() {
public void replay(FeatureLevelRecord record) {
if (record.name().equals(MetadataVersion.FEATURE_NAME)) {
- metadataVersionChange = MetadataVersion.fromFeatureLevel(record.featureLevel());
+ // Support for the `metadata.version` feature flag was added in IBP_3_3_IV0, so it's possible (but unlikely) that we read
+ // records with a feature level that is no longer supported for clusters that used a pre-release version of 3.3.0.
+ // We automatically fallback to IBP_3_3_IV3 in that case. We use explicit versions instead of `MINIMUM_VERSION` because
+ // we want to force an explicit decision if we change `MetadataVersion.MINIMUM_VERSION` in the future.
+ if (record.featureLevel() >= MINIMUM_PERSISTED_FEATURE_LEVEL && record.featureLevel() <= MetadataVersion.IBP_3_3_IV3.featureLevel())
+ metadataVersionChange = MetadataVersion.IBP_3_3_IV3;
Review Comment:
Since we only support upgrading from officially released 3.3 to 4.0, we could probably just fail MV 3_3_IV0 to 3_3_IV2 as other unsupported MVs.
##########
server-common/src/main/java/org/apache/kafka/server/common/UnitTestFeatureVersion.java:
##########
@@ -321,7 +321,7 @@ public Map<String, Short> dependencies() {
* The feature is used to test the default value has MV dependency that is behind the bootstrap MV.
*/
public enum FV7 implements FeatureVersion {
- UT_FV7_0(0, MetadataVersion.MINIMUM_KRAFT_VERSION, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
+ UT_FV7_0(0, MetadataVersion.MINIMUM_VERSION, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
Review Comment:
The test passes since the error message is only used if an expected exception is not thrown.
##########
metadata/src/main/java/org/apache/kafka/image/FeaturesImage.java:
##########
@@ -56,7 +53,7 @@ public FeaturesImage(
public boolean isEmpty() {
return finalizedVersions.isEmpty() &&
- metadataVersion.equals(MetadataVersion.MINIMUM_KRAFT_VERSION);
+ metadataVersion.equals(MetadataVersion.MINIMUM_VERSION);
Review Comment:
`isEmpty()` is only used in `MetadataLoaderTest`. So, we can address this in a followup jira. But I am not sure if `isEmpty()` will ever be false given that metadata version is required in the metadata log and probably the snapshot.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1952741967
##########
metadata/src/test/java/org/apache/kafka/image/FeaturesImageTest.java:
##########
@@ -163,7 +163,7 @@ public void testEmpty() {
assertFalse(new FeaturesImage(Collections.singletonMap("foo", (short) 1),
FeaturesImage.EMPTY.metadataVersion()).isEmpty());
assertFalse(new FeaturesImage(FeaturesImage.EMPTY.finalizedVersions(),
- MetadataVersion.IBP_3_3_IV0).isEmpty());
+ MetadataVersion.IBP_3_4_IV0).isEmpty());
Review Comment:
That causes the test to fail. This is related to your question of whether `FeatuesImage.isEmpty` should be changed not to include `metadataVersion.equals(MetadataVersion.MINIMUM_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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma merged PR #18845:
URL: https://github.com/apache/kafka/pull/18845
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1960678323
##########
metadata/src/test/java/org/apache/kafka/image/writer/ImageWriterOptionsTest.java:
##########
@@ -48,40 +48,33 @@
public class ImageWriterOptionsTest {
@Test
public void testDefaultLossHandler() {
- ImageWriterOptions options = new ImageWriterOptions.Builder().build();
+ ImageWriterOptions options = new ImageWriterOptions.Builder(MetadataVersion.latestProduction()).build();
assertEquals("stuff", assertThrows(UnwritableMetadataException.class,
() -> options.handleLoss("stuff")).loss());
}
@Test
public void testSetMetadataVersion() {
Review Comment:
I agree that this test isn't useful since the special logic for the setter was removed - even without removing the builder.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1951767994
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -86,12 +84,7 @@ BootstrapMetadata readFromConfiguration() {
return BootstrapMetadata.fromVersion(MetadataVersion.latestProduction(), "the default bootstrap");
}
MetadataVersion version = MetadataVersion.fromVersionString(ibp.get());
- if (version.isLessThan(MINIMUM_BOOTSTRAP_VERSION)) {
Review Comment:
yes, we're getting rid of the fallback
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950991354
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -368,7 +320,8 @@ public static MetadataVersion fromVersionString(String versionString) {
key = String.join(".", Arrays.copyOfRange(versionSegments, 0, numSegments));
}
return Optional.ofNullable(IBP_VERSIONS.get(key)).orElseThrow(() ->
- new IllegalArgumentException("Version " + versionString + " is not a valid version")
+ new IllegalArgumentException("Version " + versionString + " is not a valid version. The minimum version is " + MINIMUM_VERSION
+ + " and the maximum version is " + latestTesting())
Review Comment:
If you check `MetadataVersion.testFromVersionString`, this code is accurate - `4.0-IV3` works even though it's a testing version still.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950084549
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -86,12 +84,7 @@ BootstrapMetadata readFromConfiguration() {
return BootstrapMetadata.fromVersion(MetadataVersion.latestProduction(), "the default bootstrap");
}
MetadataVersion version = MetadataVersion.fromVersionString(ibp.get());
- if (version.isLessThan(MINIMUM_BOOTSTRAP_VERSION)) {
Review Comment:
Sure, will push a commit with this change.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950205574
##########
metadata/src/main/java/org/apache/kafka/image/FeaturesImage.java:
##########
@@ -56,7 +53,7 @@ public FeaturesImage(
public boolean isEmpty() {
return finalizedVersions.isEmpty() &&
- metadataVersion.equals(MetadataVersion.MINIMUM_KRAFT_VERSION);
+ metadataVersion.equals(MetadataVersion.MINIMUM_VERSION);
Review Comment:
This is a good question, I assumed both conditions together would still mean empty, but this assumption may be wrong.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18845:
URL: https://github.com/apache/kafka/pull/18845#issuecomment-2649831949
@junrao Thanks for the review. I addressed your comments and/or replied within the relevant thread. I also pushed a couple of commits for changes I had in progress: (1) Use `MINIMUM_VERSION` in more places (to reduce future changes) and (2) Set topic id more consistently in `PartitionTest`. Finally, I also merged from trunk hoping it fixes some of the test failures.
Let's see how the test results look after this.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1952779571
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadataTest.java:
##########
@@ -126,13 +126,13 @@ public void testFeatureLevelForFeature() {
static final List<ApiMessageAndVersion> RECORDS_WITH_OLD_METADATA_VERSION = Collections.singletonList(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(FEATURE_NAME).
- setFeatureLevel(IBP_3_0_IV1.featureLevel()), (short) 0));
+ setFeatureLevel(MetadataVersionTestUtils.IBP_3_0_IV1_FEATURE_LEVEL), (short) 0));
@Test
public void testFromRecordsListWithOldMetadataVersion() {
RuntimeException exception = assertThrows(RuntimeException.class,
() -> BootstrapMetadata.fromRecords(RECORDS_WITH_OLD_METADATA_VERSION, "quux"));
- assertEquals("Bootstrap metadata.version before 3.3-IV0 are not supported. Can't load " +
- "metadata from quux", exception.getMessage());
+ assertEquals("No MetadataVersion with feature level 1. Valid feature levels are from 7 to 25.",
Review Comment:
I changed it to include constants for both cases.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1948439556
##########
metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java:
##########
@@ -77,59 +77,6 @@ static ApiMessageAndVersion metadataVersionRecord(MetadataVersion metadataVersio
setFeatureLevel(metadataVersion.featureLevel()), (short) 0);
}
- /**
- * Test downgrading to a MetadataVersion that doesn't support FeatureLevelRecord.
- */
- @Test
- public void testPremodernVersion() {
Review Comment:
We don't support these, so remove it.
##########
metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java:
##########
@@ -77,59 +77,6 @@ static ApiMessageAndVersion metadataVersionRecord(MetadataVersion metadataVersio
setFeatureLevel(metadataVersion.featureLevel()), (short) 0);
}
- /**
- * Test downgrading to a MetadataVersion that doesn't support FeatureLevelRecord.
- */
- @Test
- public void testPremodernVersion() {
Review Comment:
We don't support these anymore, so remove it.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1948438967
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -86,12 +84,7 @@ BootstrapMetadata readFromConfiguration() {
return BootstrapMetadata.fromVersion(MetadataVersion.latestProduction(), "the default bootstrap");
}
MetadataVersion version = MetadataVersion.fromVersionString(ibp.get());
- if (version.isLessThan(MINIMUM_BOOTSTRAP_VERSION)) {
Review Comment:
@cmccabe Is it important to automatically fallback if older metadata versions are received here or can we throw an error? It's harder to fallback (but not impossible) since I removed older metadata versions.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949569316
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1813,7 +1813,7 @@ class Partition(val topicPartition: TopicPartition,
private def submitAlterPartition(proposedIsrState: PendingPartitionChange): CompletableFuture[LeaderAndIsr] = {
debug(s"Submitting ISR state change $proposedIsrState")
val future = alterIsrManager.submit(
- new TopicIdPartition(topicId.getOrElse(Uuid.ZERO_UUID), topicPartition),
+ new org.apache.kafka.server.common.TopicIdPartition(topicId.getOrElse(throw new IllegalStateException("Topic id not set")), topicPartition.partition),
Review Comment:
I wonder whether we should make the topic id mandatory (not an Option) in the Partition now. It is always set in KRaft 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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1952687484
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -240,38 +241,29 @@ public void testReplayRecords() {
private static final FeatureControlManager.Builder TEST_MANAGER_BUILDER1 =
new FeatureControlManager.Builder().
setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_3_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV2);
+ MetadataVersion.MINIMUM_VERSION.featureLevel(), MetadataVersion.IBP_3_6_IV0.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_4_IV0);
@Test
public void testApplyMetadataVersionChangeRecord() {
FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
manager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel()));
- assertEquals(MetadataVersion.IBP_3_3_IV3, manager.metadataVersion());
- }
-
- @Test
- public void testCannotDowngradeToVersionBeforeMinimumSupportedKraftVersion() {
- FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
- assertEquals(ControllerResult.of(Collections.emptyList(), new ApiError(Errors.INVALID_UPDATE_VERSION,
- "Invalid update version 3 for feature metadata.version. Local controller 0 only " +
- "supports versions 4-7")),
- manager.updateFeatures(
- singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_2_IV0.featureLevel()),
- singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.UNSAFE_DOWNGRADE),
- true));
+ setFeatureLevel(MetadataVersion.IBP_3_4_IV0.featureLevel()));
Review Comment:
This was previously checking that the metadata version would change and using minimum would take it backwards. But I did notice an issue where we are actually using the same version versus a newer one - fixing 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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1957186474
##########
metadata/src/test/java/org/apache/kafka/image/FeaturesImageTest.java:
##########
@@ -163,7 +163,7 @@ public void testEmpty() {
assertFalse(new FeaturesImage(Collections.singletonMap("foo", (short) 1),
FeaturesImage.EMPTY.metadataVersion()).isEmpty());
assertFalse(new FeaturesImage(FeaturesImage.EMPTY.finalizedVersions(),
- MetadataVersion.IBP_3_3_IV0).isEmpty());
+ MetadataVersion.IBP_3_4_IV0).isEmpty());
Review Comment:
I updated `FeaturesImage.isEmpty` and also this test. Resolving.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1957186684
##########
metadata/src/main/java/org/apache/kafka/image/FeaturesImage.java:
##########
@@ -56,7 +53,7 @@ public FeaturesImage(
public boolean isEmpty() {
return finalizedVersions.isEmpty() &&
- metadataVersion.equals(MetadataVersion.MINIMUM_KRAFT_VERSION);
+ metadataVersion.equals(MetadataVersion.MINIMUM_VERSION);
Review Comment:
It turns out that this had to be fixed for all the tests to pass. Addressed in the latest commit.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1957186965
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -437,18 +438,6 @@ class KRaftClusterTest {
}
}
- @Test
- def testCreateClusterInvalidMetadataVersion(): Unit = {
Review Comment:
It's no longer possible to set the bootstrap metadata version to lower than 3.3-IV3.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949874774
##########
metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java:
##########
@@ -328,15 +314,10 @@ private ApiError updateMetadataVersion(
try {
newVersion = MetadataVersion.fromFeatureLevel(newVersionLevel);
} catch (IllegalArgumentException e) {
- return invalidMetadataVersion(newVersionLevel, "Unknown metadata.version.");
+ return invalidMetadataVersion(newVersionLevel, "Valid versions are from "
Review Comment:
Should we enhance the error message for `fromFeatureLevel` as well?
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1652,9 +1652,7 @@ private void registerWriteNoOpRecord(long maxIdleIntervalNs) {
periodicControl.registerTask(new PeriodicTask("writeNoOpRecord",
() -> {
ArrayList<ApiMessageAndVersion> records = new ArrayList<>(1);
- if (featureControl.metadataVersion().isNoOpRecordSupported()) {
- records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
- }
+ records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
Review Comment:
Maybe we can streamline the code?
```
() -> ControllerResult.of(List.of(new ApiMessageAndVersion(new NoOpRecord(), (short) 0)), false),
```
##########
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java:
##########
@@ -1424,15 +1406,10 @@ void handleBrokerUnregistered(int brokerId, long brokerEpoch,
* @param records The record list to append to.
*/
void handleBrokerUnfenced(int brokerId, long brokerEpoch, List<ApiMessageAndVersion> records) {
- if (featureControl.metadataVersion().isBrokerRegistrationChangeRecordSupported()) {
- records.add(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord().
- setBrokerId(brokerId).setBrokerEpoch(brokerEpoch).
- setFenced(BrokerRegistrationFencingChange.UNFENCE.value()),
- (short) 0));
- } else {
- records.add(new ApiMessageAndVersion(new UnfenceBrokerRecord().setId(brokerId).
- setEpoch(brokerEpoch), (short) 0));
- }
+ records.add(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord().
Review Comment:
ditto
##########
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java:
##########
@@ -1379,16 +1367,10 @@ void handleBrokerFenced(int brokerId, List<ApiMessageAndVersion> records) {
}
generateLeaderAndIsrUpdates("handleBrokerFenced", brokerId, NO_LEADER, NO_LEADER, records,
brokersToIsrs.partitionsWithBrokerInIsr(brokerId));
- if (featureControl.metadataVersion().isBrokerRegistrationChangeRecordSupported()) {
- records.add(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord().
- setBrokerId(brokerId).setBrokerEpoch(brokerRegistration.epoch()).
- setFenced(BrokerRegistrationFencingChange.FENCE.value()),
- (short) 0));
- } else {
- records.add(new ApiMessageAndVersion(new FenceBrokerRecord().
- setId(brokerId).setEpoch(brokerRegistration.epoch()),
- (short) 0));
- }
+ records.add(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord().
Review Comment:
Could you please update comment of `handleBrokerFenced` according to this change?
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadataTest.java:
##########
@@ -126,7 +125,7 @@ public void testFeatureLevelForFeature() {
static final List<ApiMessageAndVersion> RECORDS_WITH_OLD_METADATA_VERSION = Collections.singletonList(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(FEATURE_NAME).
- setFeatureLevel(IBP_3_0_IV1.featureLevel()), (short) 0));
+ setFeatureLevel((short) 1), (short) 0));
Review Comment:
This feature version is not supported any more, so we should either remove or modify the test.
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java:
##########
@@ -128,18 +126,6 @@ public void testReassignment() throws Exception {
executeAndVerifyReassignment();
}
- @ClusterTests({
- @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, metadataVersion = IBP_3_3_IV0)
- })
- public void testReassignmentWithAlterPartitionDisabled() throws Exception {
- // Test reassignment when the IBP is on an older version which does not use
- // the `AlterPartition` API. In this case, the controller will register individual
- // watches for each reassigning partition so that the reassignment can be
- // completed as soon as the ISR is expanded.
Review Comment:
It seems to me the behavior is related to zk only, and in kraft mode `AlterPartition` should be supported totally. Hence, it is fine to remove the test.
BTW, the test case for kraft is unnecessary - I overlook the scenario in merging #15675
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -86,12 +84,7 @@ BootstrapMetadata readFromConfiguration() {
return BootstrapMetadata.fromVersion(MetadataVersion.latestProduction(), "the default bootstrap");
}
MetadataVersion version = MetadataVersion.fromVersionString(ibp.get());
- if (version.isLessThan(MINIMUM_BOOTSTRAP_VERSION)) {
Review Comment:
the `ibp` is always empty after the config is removed by #18566
I had opened https://issues.apache.org/jira/browse/KAFKA-18740 to cleanup it. Maybe this PR can include the cleanup as well?
##########
metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java:
##########
@@ -293,11 +293,7 @@ public ApiMessageAndVersion toRecord(ImageWriterOptions options) {
setFenced(fenced);
if (inControlledShutdown) {
- if (options.metadataVersion().isInControlledShutdownStateSupported()) {
- registrationRecord.setInControlledShutdown(true);
- } else {
- options.handleLoss("the inControlledShutdown state of one or more brokers");
- }
+ registrationRecord.setInControlledShutdown(true);
Review Comment:
Maybe we can set the flag in line#294.
```
RegisterBrokerRecord registrationRecord = new RegisterBrokerRecord().
setBrokerId(id).
setRack(rack.orElse(null)).
setBrokerEpoch(epoch).
setIncarnationId(incarnationId).
setFenced(fenced).
setInControlledShutdown(inControlledShutdown);
```
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1952564479
##########
metadata/src/main/java/org/apache/kafka/controller/ActivationRecordsGenerator.java:
##########
@@ -153,7 +153,7 @@ static ControllerResult<Void> recordsForNonEmptyLog(
* </p>
* If the log is empty, write the bootstrap records. If the log is not empty, do some validation and
* possibly write some records to put the log into a valid state. For bootstrap records, if KIP-868
- * metadata transactions are supported, ues them. Otherwise, write the bootstrap records as an
+ * metadata transactions are supported, use them. Otherwise, write the bootstrap records as an
* atomic batch. The single atomic batch can be problematic if the bootstrap records are too large
* (e.g., lots of SCRAM credentials). If ZK migrations are enabled, the activation records will
* include a ZkMigrationState record regardless of whether the log was empty or not.
Review Comment:
Please also cleanup this outdate ZK migrations comment
##########
core/src/test/scala/unit/kafka/server/ApiVersionsResponseIntegrationTest.scala:
##########
@@ -38,28 +37,18 @@ class ApiVersionsResponseIntegrationTest extends BaseRequestTest {
connectAndReceive[ApiVersionsResponse](request)
}
- @ParameterizedTest
- @ValueSource(strings = Array("kraft"))
- def testSendV3ApiVersionsRequest(quorum: String): Unit = {
+ @Test
+ def testSendV3ApiVersionsRequest(): Unit = {
val response = sendApiVersionsRequest(3)
- if (quorum.equals("kraft")) {
- assertFeatureHasMinVersion("metadata.version", response.data().supportedFeatures(), 1)
- assertFeatureMissing("kraft.version", response.data().supportedFeatures())
- } else {
- assertEquals(0, response.data().supportedFeatures().size())
- }
+ assertFeatureHasMinVersion("metadata.version", response.data().supportedFeatures(), 7)
+ assertFeatureMissing("kraft.version", response.data().supportedFeatures())
}
- @ParameterizedTest
- @ValueSource(strings = Array("kraft"))
- def testSendV4ApiVersionsRequest(quorum: String): Unit = {
+ @Test
+ def testSendV4ApiVersionsRequest(): Unit = {
val response = sendApiVersionsRequest(4)
- if (quorum.equals("kraft")) {
- assertFeatureHasMinVersion("metadata.version", response.data().supportedFeatures(), 1)
- assertFeatureHasMinVersion("kraft.version", response.data().supportedFeatures(), 0)
- } else {
- assertEquals(0, response.data().supportedFeatures().size())
- }
+ assertFeatureHasMinVersion("metadata.version", response.data().supportedFeatures(), 7)
Review Comment:
ditto `MINIMUM_VERSION.featureLevel()` instead of 7 ?
##########
core/src/main/scala/kafka/server/AlterPartitionManager.scala:
##########
@@ -224,53 +215,35 @@ class DefaultAlterPartitionManager(
private def buildRequest(
inflightAlterPartitionItems: Seq[AlterPartitionItem],
brokerEpoch: Long
Review Comment:
Please also update the return value document.
##########
core/src/test/scala/unit/kafka/server/AlterPartitionManagerTest.scala:
##########
@@ -552,24 +535,10 @@ class AlterPartitionManagerTest {
}
object AlterPartitionManagerTest {
- def provideMetadataVersions(): JStream[MetadataVersion] = {
+ def provideLeaderRecoveryState(): JStream[Arguments] = {
Review Comment:
we can use `@EnumSource(classOf[LeaderRecoveryState])`, thus this method can remove
##########
core/src/test/scala/unit/kafka/server/ApiVersionsResponseIntegrationTest.scala:
##########
@@ -38,28 +37,18 @@ class ApiVersionsResponseIntegrationTest extends BaseRequestTest {
connectAndReceive[ApiVersionsResponse](request)
}
- @ParameterizedTest
- @ValueSource(strings = Array("kraft"))
- def testSendV3ApiVersionsRequest(quorum: String): Unit = {
+ @Test
+ def testSendV3ApiVersionsRequest(): Unit = {
val response = sendApiVersionsRequest(3)
- if (quorum.equals("kraft")) {
- assertFeatureHasMinVersion("metadata.version", response.data().supportedFeatures(), 1)
- assertFeatureMissing("kraft.version", response.data().supportedFeatures())
- } else {
- assertEquals(0, response.data().supportedFeatures().size())
- }
+ assertFeatureHasMinVersion("metadata.version", response.data().supportedFeatures(), 7)
Review Comment:
ditto `MINIMUM_VERSION.featureLevel()` instead of 7 ?
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -99,10 +100,10 @@ public void testUpdateFeatures() {
FeatureControlManager manager = new FeatureControlManager.Builder().
setQuorumFeatures(features(TestFeatureVersion.FEATURE_NAME, 0, 2)).
setSnapshotRegistry(snapshotRegistry).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
+ setMetadataVersion(MetadataVersion.MINIMUM_VERSION).
build();
snapshotRegistry.idempotentCreateSnapshot(-1);
- assertEquals(new FinalizedControllerFeatures(Collections.singletonMap("metadata.version", (short) 4), -1),
+ assertEquals(new FinalizedControllerFeatures(Collections.singletonMap("metadata.version", (short) 7), -1),
Review Comment:
ditto `MINIMUM_VERSION.featureLevel()` instead of 7 ?
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -142,11 +143,11 @@ public void testReplay() {
setLogContext(logContext).
setQuorumFeatures(features("foo", 1, 2)).
setSnapshotRegistry(snapshotRegistry).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
+ setMetadataVersion(MetadataVersion.MINIMUM_VERSION).
build();
manager.replay(record);
snapshotRegistry.idempotentCreateSnapshot(123);
- assertEquals(new FinalizedControllerFeatures(versionMap("metadata.version", 4, "foo", 2), 123),
+ assertEquals(new FinalizedControllerFeatures(versionMap("metadata.version", 7, "foo", 2), 123),
Review Comment:
ditto `MINIMUM_VERSION.featureLevel()` instead of 7 ?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953113246
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -495,7 +495,9 @@ BrokerFeature processRegistrationFeature(
FinalizedControllerFeatures finalizedFeatures,
BrokerRegistrationRequestData.Feature feature
) {
- int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ? 1 : 0; // The default value for MetadataVersion is 1 not 0.
+ // The default value for MetadataVersion changes over time while other features start at `0`
+ int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ?
+ MetadataVersion.MINIMUM_VERSION.featureLevel() : 0;
Review Comment:
If we're dropping support for everything before 3.3 we can always count on having a FeatureRecord for `metadata.version`. However doing this may require a lot of code changes since many junit tests may initialize with an empty log.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953228668
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -495,7 +495,9 @@ BrokerFeature processRegistrationFeature(
FinalizedControllerFeatures finalizedFeatures,
BrokerRegistrationRequestData.Feature feature
) {
- int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ? 1 : 0; // The default value for MetadataVersion is 1 not 0.
+ // The default value for MetadataVersion changes over time while other features start at `0`
+ int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ?
+ MetadataVersion.MINIMUM_VERSION.featureLevel() : 0;
Review Comment:
I can try. We have one test failing only in CI due to the minimum being used, perhaps this will help.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1951875202
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -368,7 +320,8 @@ public static MetadataVersion fromVersionString(String versionString) {
key = String.join(".", Arrays.copyOfRange(versionSegments, 0, numSegments));
}
return Optional.ofNullable(IBP_VERSIONS.get(key)).orElseThrow(() ->
- new IllegalArgumentException("Version " + versionString + " is not a valid version")
+ new IllegalArgumentException("Version " + versionString + " is not a valid version. The minimum version is " + MINIMUM_VERSION
+ + " and the maximum version is " + latestTesting())
Review Comment:
> As you said, that seems to be inconsistent with the docs for StorageTool - perhaps we should file an issue to fix that in its own PR (assuming it's not a regression in this one, but I don't think the relevant code was changed).
Make sense to me, I will file an issue to track it
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1952781872
##########
metadata/src/main/java/org/apache/kafka/controller/ActivationRecordsGenerator.java:
##########
@@ -137,10 +137,10 @@ static ControllerResult<Void> recordsForNonEmptyLog(
}
}
- if (curMetadataVersion.equals(MetadataVersion.MINIMUM_KRAFT_VERSION)) {
+ if (curMetadataVersion.equals(MetadataVersion.MINIMUM_VERSION)) {
Review Comment:
I went ahead and removed the log. @cmccabe Let me know if you think we should do something different.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950707798
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -378,7 +331,8 @@ public static MetadataVersion fromFeatureLevel(short version) {
return metadataVersion;
}
}
- throw new IllegalArgumentException("No MetadataVersion with feature level " + version);
+ throw new IllegalArgumentException("No MetadataVersion with feature level " + version + ". Valid feature levels are from "
+ + MINIMUM_VERSION.featureLevel + " to " + latestTesting().featureLevel + ".");
Review Comment:
ditto, we should use `latestProduction()`.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1960679302
##########
metadata/src/main/java/org/apache/kafka/image/writer/ImageWriterOptions.java:
##########
@@ -29,30 +29,22 @@
public final class ImageWriterOptions {
public static class Builder {
private MetadataVersion metadataVersion;
- private MetadataVersion requestedMetadataVersion;
- private boolean isEligibleLeaderReplicasEnabled = false;
private Consumer<UnwritableMetadataException> lossHandler = e -> {
throw e;
};
+ private boolean isEligibleLeaderReplicasEnabled = false;
- public Builder() {
- this.metadataVersion = MetadataVersion.latestProduction();
+ public Builder(MetadataVersion metadataVersion) {
+ this.metadataVersion = metadataVersion;
}
public Builder(MetadataImage image) {
- this.metadataVersion = image.features().metadataVersion();
+ this.metadataVersion = image.features().metadataVersionOrThrow();
this.isEligibleLeaderReplicasEnabled = image.features().isElrEnabled();
}
public Builder setMetadataVersion(MetadataVersion metadataVersion) {
Review Comment:
This is a general pattern for the metadata module, I would rather not change it as part of this PR. We can have that discussion via a separate JIRA if useful.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949971600
##########
metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java:
##########
@@ -328,15 +314,10 @@ private ApiError updateMetadataVersion(
try {
newVersion = MetadataVersion.fromFeatureLevel(newVersionLevel);
} catch (IllegalArgumentException e) {
- return invalidMetadataVersion(newVersionLevel, "Unknown metadata.version.");
+ return invalidMetadataVersion(newVersionLevel, "Valid versions are from "
Review Comment:
I think I did this in a subsequent commit, perhaps after you did your review.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1952730662
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -349,28 +344,27 @@ public void testCanUseUnsafeDowngradeIfMetadataChanged() {
public void testCanUseSafeDowngradeIfMetadataDidNotChange() {
FeatureControlManager manager = new FeatureControlManager.Builder().
setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV1.featureLevel())).
- setMetadataVersion(MetadataVersion.IBP_3_1_IV0).
- setMinimumBootstrapVersion(MetadataVersion.IBP_3_0_IV1).
+ MetadataVersion.MINIMUM_VERSION.featureLevel(), MetadataVersion.IBP_3_6_IV0.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_5_IV0).
build();
assertEquals(ControllerResult.of(Collections.emptyList(), ApiError.NONE),
manager.updateFeatures(
- singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_0_IV1.featureLevel()),
+ singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_4_IV0.featureLevel()),
singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.SAFE_DOWNGRADE),
true));
}
@Test
- public void testCannotDowngradeBefore3_3_IV0() {
+ public void testCannotDowngradeBeforeMinimumKraftVersion() {
FeatureControlManager manager = new FeatureControlManager.Builder().
setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
+ MetadataVersionTestUtils.IBP_3_0_IV1_FEATURE_LEVEL, MetadataVersion.MINIMUM_VERSION.featureLevel())).
Review Comment:
The current code tests a slightly different path than the one you're suggesting. The error is thrown in a different place. I looked a bit more deeply though and perhaps your suggestion is closer to what will happen in production:
```java
public static Map<String, VersionRange> defaultSupportedFeatureMap(boolean enableUnstable) {
Map<String, VersionRange> features = new HashMap<>(1);
features.put(MetadataVersion.FEATURE_NAME, VersionRange.of(
MetadataVersion.MINIMUM_VERSION.featureLevel(),
enableUnstable ?
MetadataVersion.latestTesting().featureLevel() :
MetadataVersion.latestProduction().featureLevel()));
for (Feature feature : Feature.PRODUCTION_FEATURES) {
short maxVersion = enableUnstable ? feature.latestTesting() : feature.latestProduction();
if (maxVersion > 0) {
features.put(feature.featureName(), VersionRange.of(feature.minimumProduction(), maxVersion));
}
}
return features;
}
```
Changing it to your suggestion.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1952539978
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -368,7 +320,8 @@ public static MetadataVersion fromVersionString(String versionString) {
key = String.join(".", Arrays.copyOfRange(versionSegments, 0, numSegments));
}
return Optional.ofNullable(IBP_VERSIONS.get(key)).orElseThrow(() ->
- new IllegalArgumentException("Version " + versionString + " is not a valid version")
+ new IllegalArgumentException("Version " + versionString + " is not a valid version. The minimum version is " + MINIMUM_VERSION
+ + " and the maximum version is " + latestTesting())
Review Comment:
The `FeatureCommand.java` also have same issue, I addressed in same Jira https://issues.apache.org/jira/browse/KAFKA-18778
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18845:
URL: https://github.com/apache/kafka/pull/18845#issuecomment-2651247113
Besides the open discussion threads, there are two additional issues to consider/resolve:
1. Some tests in `PlaintextAdminIntegrationTest`are failing in the CI runs, but not locally[1]
2. `FeaturesDelta.replay` reads feature levels from the log and converts them to `MetadataVersion`. It's possible (but unlikely) that someone ended up with 3.3-IV0, 3.3-IV1, 3.3-IV2 in the log if using an unreleased version of Apache Kafka 3.3. We should consider automatically converting to `3.3-IV3` in such cases.
The failures have the following error:
> org.apache.kafka.server.fault.FaultHandlerException: quorumTestHarnessFaultHandler: Broker configuration does not support the cluster MetadataVersion: requirement failed: Multiple log directories (aka JBOD) are not supported in the current MetadataVersion 3.3-IV3. Need 3.7-IV2 or higher
Even though there is a log message just before saying:
> Formatting metadata directory /tmp/kafka-2205696658360646817 with metadata.version 4.0-IV3.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1960653141
##########
metadata/src/test/java/org/apache/kafka/image/writer/ImageWriterOptionsTest.java:
##########
@@ -48,40 +48,33 @@
public class ImageWriterOptionsTest {
@Test
public void testDefaultLossHandler() {
- ImageWriterOptions options = new ImageWriterOptions.Builder().build();
+ ImageWriterOptions options = new ImageWriterOptions.Builder(MetadataVersion.latestProduction()).build();
assertEquals("stuff", assertThrows(UnwritableMetadataException.class,
() -> options.handleLoss("stuff")).loss());
}
@Test
public void testSetMetadataVersion() {
Review Comment:
If we rewrite `ImageWriterOptions` by record class, this `testSetMetadataVersion` is unnecessary as all are pure setters.
##########
metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java:
##########
@@ -718,30 +710,14 @@ public void testRegistrationWithUnsupportedMetadataVersion() {
setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
Collections.singleton(new BrokerRegistrationRequestData.Feature().
setName(MetadataVersion.FEATURE_NAME).
- setMinSupportedVersion(MetadataVersion.IBP_3_3_IV3.featureLevel()).
- setMaxSupportedVersion(MetadataVersion.IBP_3_3_IV3.featureLevel())).iterator())).
+ setMinSupportedVersion(MetadataVersion.IBP_3_4_IV0.featureLevel()).
+ setMaxSupportedVersion(MetadataVersion.IBP_3_4_IV0.featureLevel())).iterator())).
setIncarnationId(Uuid.fromString("0H4fUu1xQEKXFYwB1aBjhg")),
123L,
featureControl.finalizedFeatures(Long.MAX_VALUE),
false)).getMessage());
}
- @Test
- public void testRegisterControlWithOlderMetadataVersion() {
- FeatureControlManager featureControl = new FeatureControlManager.Builder().
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
Review Comment:
This test appears necessary as it verifies MV which is older than `IBP_3_7_IV0`. Perhaps we could simply change `IBP_3_3_IV0` to `IBP_3_3_IV3` instead.
##########
metadata/src/main/java/org/apache/kafka/image/writer/ImageWriterOptions.java:
##########
@@ -29,30 +29,22 @@
public final class ImageWriterOptions {
public static class Builder {
private MetadataVersion metadataVersion;
- private MetadataVersion requestedMetadataVersion;
- private boolean isEligibleLeaderReplicasEnabled = false;
private Consumer<UnwritableMetadataException> lossHandler = e -> {
throw e;
};
+ private boolean isEligibleLeaderReplicasEnabled = false;
- public Builder() {
- this.metadataVersion = MetadataVersion.latestProduction();
+ public Builder(MetadataVersion metadataVersion) {
+ this.metadataVersion = metadataVersion;
}
public Builder(MetadataImage image) {
- this.metadataVersion = image.features().metadataVersion();
+ this.metadataVersion = image.features().metadataVersionOrThrow();
this.isEligibleLeaderReplicasEnabled = image.features().isElrEnabled();
}
public Builder setMetadataVersion(MetadataVersion metadataVersion) {
Review Comment:
After removing those if-else, it seems we don't need the builder anymore. Maybe we can rewrite `ImageWriterOptions` by record class
##########
metadata/src/test/java/org/apache/kafka/image/MetadataVersionChangeTest.java:
##########
@@ -30,31 +30,31 @@
@Timeout(value = 40)
public class MetadataVersionChangeTest {
- private static final MetadataVersionChange CHANGE_3_0_IV1_TO_3_3_IV0 =
- new MetadataVersionChange(IBP_3_0_IV1, IBP_3_3_IV0);
+ private static final MetadataVersionChange CHANGE_MINUMUM_TO_LATEST =
Review Comment:
typo: `MINUMUM` -> `MINIMUM`
##########
core/src/main/scala/kafka/server/AlterPartitionManager.scala:
##########
@@ -92,8 +87,7 @@ object AlterPartitionManager {
scheduler = scheduler,
time = time,
brokerId = config.brokerId,
- brokerEpochSupplier = brokerEpochSupplier,
- metadataVersionSupplier = () => metadataCache.metadataVersion()
Review Comment:
we can remove `metadataCache` from the `apply` method
##########
core/src/main/scala/kafka/server/AlterPartitionManager.scala:
##########
@@ -114,7 +107,7 @@ class DefaultAlterPartitionManager(
// and re-created, we cannot have two entries in this Map especially if we cannot
// use an AlterPartition request version which supports topic ids in the end because
// the two updates with the same name would be merged together.
- private[server] val unsentIsrUpdates: util.Map[TopicPartition, AlterPartitionItem] = new ConcurrentHashMap[TopicPartition, AlterPartitionItem]()
+ private[server] val unsentIsrUpdates = new ConcurrentHashMap[TopicIdPartition, AlterPartitionItem]()
Review Comment:
Could you please update the docs according to the change?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1960528979
##########
metadata/src/main/java/org/apache/kafka/image/FeaturesDelta.java:
##########
@@ -58,7 +59,14 @@ public void finishSnapshot() {
public void replay(FeatureLevelRecord record) {
if (record.name().equals(MetadataVersion.FEATURE_NAME)) {
- metadataVersionChange = MetadataVersion.fromFeatureLevel(record.featureLevel());
+ // Support for the `metadata.version` feature flag was added in IBP_3_3_IV0, so it's possible (but unlikely) that we read
+ // records with a feature level that is no longer supported for clusters that used a pre-release version of 3.3.0.
+ // We automatically fallback to IBP_3_3_IV3 in that case. We use explicit versions instead of `MINIMUM_VERSION` because
+ // we want to force an explicit decision if we change `MetadataVersion.MINIMUM_VERSION` in the future.
+ if (record.featureLevel() >= MINIMUM_PERSISTED_FEATURE_LEVEL && record.featureLevel() <= MetadataVersion.IBP_3_3_IV3.featureLevel())
+ metadataVersionChange = MetadataVersion.IBP_3_3_IV3;
Review Comment:
I discussed this with a few people and the bit that was missing in the discussion above is that we generate snapshots every hour by default and hence this issue is unlikely to ever happen and the workaround is reasonably simple. I updated the code to remove the fallback and include a helpful message if this issue occurs.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18845:
URL: https://github.com/apache/kafka/pull/18845#issuecomment-2666840092
I resolved the outstanding item, so no known issues remain in this PR.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1956762973
##########
metadata/src/main/java/org/apache/kafka/image/node/MetadataNode.java:
##########
@@ -53,7 +53,8 @@ default void print(MetadataNodePrinter printer) {
for (String name : names) {
printer.enterNode(name);
MetadataNode child = child(name);
- child.print(printer);
+ if (child != null)
+ child.print(printer);
Review Comment:
Hm, it's definitely a bug, but I don't think it relates to this PR. The *Node classes are used by the metadata shell and some toString methods, so they are just informational. I think the exact side effect is that acls would appear at the wrong level of the metadata shell "tree".
I suggest we file a separate issue and fix it for 4.0.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1956525411
##########
metadata/src/main/java/org/apache/kafka/image/node/MetadataNode.java:
##########
@@ -53,7 +53,8 @@ default void print(MetadataNodePrinter printer) {
for (String name : names) {
printer.enterNode(name);
MetadataNode child = child(name);
- child.print(printer);
+ if (child != null)
+ child.print(printer);
Review Comment:
In reading the code of child, I notice that it seems the `AclsImageByIdNode` is misplaced.
```
private static final Map<String, Function<MetadataImage, MetadataNode>> CHILDREN = Map.of(
...
// AclsImageByIdNode should be replaced by AclsImageNode
AclsImageNode.NAME, image -> new AclsImageByIdNode(image.acls()),
ScramImageNode.NAME, image -> new ScramImageNode(image.scram()),
DelegationTokenImageNode.NAME, image -> new DelegationTokenImageNode(image.delegationTokens())
);
```
https://github.com/apache/kafka/blob/48283ad2e54133ebb604769dc8a72706e07f36e7/metadata/src/main/java/org/apache/kafka/image/node/MetadataImageNode.java#L46
and `AclsImage#toString` has similar issue.
```
@Override
public String toString() {
// AclsImageByIdNode should be replaced by AclsImageNode
return new AclsImageByIdNode(this).stringify();
}
```
https://github.com/apache/kafka/blob/trunk/metadata/src/main/java/org/apache/kafka/image/AclsImage.java#L78
@ijuma @mumrah WDYT?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953117041
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1652,9 +1652,7 @@ private void registerWriteNoOpRecord(long maxIdleIntervalNs) {
periodicControl.registerTask(new PeriodicTask("writeNoOpRecord",
() -> {
ArrayList<ApiMessageAndVersion> records = new ArrayList<>(1);
- if (featureControl.metadataVersion().isNoOpRecordSupported()) {
- records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
- }
+ records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
Review Comment:
@ijuma , we do not mutate the records list that gets returned from these functions later on. @chia7712 is correct that we return immutable objects in many other cases. Unfortunately, due to Java's lack of `const` I guess that isn't obvious.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949970524
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadataTest.java:
##########
@@ -126,7 +125,7 @@ public void testFeatureLevelForFeature() {
static final List<ApiMessageAndVersion> RECORDS_WITH_OLD_METADATA_VERSION = Collections.singletonList(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(FEATURE_NAME).
- setFeatureLevel(IBP_3_0_IV1.featureLevel()), (short) 0));
+ setFeatureLevel((short) 1), (short) 0));
Review Comment:
This is intended - see the name of the field (RECORDS_WITH_OLD_METADATA_VERSION). I did update this to refer to a constant.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18845:
URL: https://github.com/apache/kafka/pull/18845#issuecomment-2649590091
@chia7712 I addressed your comments except one (and left a reply explaining why).
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949489365
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java:
##########
@@ -128,18 +126,6 @@ public void testReassignment() throws Exception {
executeAndVerifyReassignment();
}
- @ClusterTests({
- @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, metadataVersion = IBP_3_3_IV0)
- })
- public void testReassignmentWithAlterPartitionDisabled() throws Exception {
- // Test reassignment when the IBP is on an older version which does not use
- // the `AlterPartition` API. In this case, the controller will register individual
- // watches for each reassigning partition so that the reassignment can be
- // completed as soon as the ISR is expanded.
Review Comment:
@cmccabe Do you know?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950200566
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectoryTest.java:
##########
@@ -81,18 +81,18 @@ public void testReadFromEmptyConfiguration() throws Exception {
@Test
public void testReadFromConfigurationWithAncientVersion() throws Exception {
try (BootstrapTestDirectory testDirectory = new BootstrapTestDirectory().createDirectory()) {
- assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.MINIMUM_BOOTSTRAP_VERSION,
- "the minimum version bootstrap with metadata.version 3.3-IV0"),
+ assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.MINIMUM_VERSION,
+ "the minimum version bootstrap with metadata.version 3.3-IV3"),
Review Comment:
Subsequent commits simplified this file and I think the issue you outlined is no longer present.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950199486
##########
server-common/src/main/java/org/apache/kafka/server/common/UnitTestFeatureVersion.java:
##########
@@ -321,7 +321,7 @@ public Map<String, Short> dependencies() {
* The feature is used to test the default value has MV dependency that is behind the bootstrap MV.
*/
public enum FV7 implements FeatureVersion {
- UT_FV7_0(0, MetadataVersion.MINIMUM_KRAFT_VERSION, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
+ UT_FV7_0(0, MetadataVersion.MINIMUM_VERSION, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
Review Comment:
Interesting - that test is passing, so I'll have to check why.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1948439772
##########
metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java:
##########
@@ -77,59 +77,6 @@ static ApiMessageAndVersion metadataVersionRecord(MetadataVersion metadataVersio
setFeatureLevel(metadataVersion.featureLevel()), (short) 0);
}
- /**
- * Test downgrading to a MetadataVersion that doesn't support FeatureLevelRecord.
- */
- @Test
- public void testPremodernVersion() {
- writeWithExpectedLosses(MetadataVersion.IBP_3_2_IV0,
- Collections.singletonList(
- "feature flag(s): foo.feature"),
- Arrays.asList(
- metadataVersionRecord(MetadataVersion.IBP_3_3_IV0),
- TEST_RECORDS.get(0),
- TEST_RECORDS.get(1),
- new ApiMessageAndVersion(new FeatureLevelRecord().
- setName("foo.feature").
- setFeatureLevel((short) 4), (short) 0)),
- Arrays.asList(
- TEST_RECORDS.get(0),
- TEST_RECORDS.get(1))
- );
- }
-
- /**
- * Test downgrading to a MetadataVersion that doesn't support inControlledShutdown.
- */
- @Test
- public void testPreControlledShutdownStateVersion() {
Review Comment:
We now always have the controlled shutdown state.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950210516
##########
core/src/test/scala/integration/kafka/coordinator/transaction/ProducerIntegrationTest.scala:
##########
@@ -55,7 +55,7 @@ import scala.jdk.CollectionConverters._
class ProducerIntegrationTest {
@ClusterTests(Array(
- new ClusterTest(metadataVersion = MetadataVersion.IBP_3_3_IV0)
+ new ClusterTest(metadataVersion = MetadataVersion.IBP_3_3_IV3)
Review Comment:
I had the same idea, but unfortunately it's not allowed:
> annotation argument needs to be a constant; found: org.apache.kafka.server.common.MetadataVersion.MINIMUM_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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950090651
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1652,9 +1652,7 @@ private void registerWriteNoOpRecord(long maxIdleIntervalNs) {
periodicControl.registerTask(new PeriodicTask("writeNoOpRecord",
() -> {
ArrayList<ApiMessageAndVersion> records = new ArrayList<>(1);
- if (featureControl.metadataVersion().isNoOpRecordSupported()) {
- records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
- }
+ records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
Review Comment:
The new version would make `records` immutable while the current code exposes the mutable `records` via `ControllerResult.records()`. Do you know if this change is safe? My cursory look indicates it may be ok, but I would have to look at this more carefully. Given that, I think we should also do this via a different PR (to avoid including a potential behavior change in the middle of this clean-up PR).
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949554971
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1813,7 +1813,7 @@ class Partition(val topicPartition: TopicPartition,
private def submitAlterPartition(proposedIsrState: PendingPartitionChange): CompletableFuture[LeaderAndIsr] = {
debug(s"Submitting ISR state change $proposedIsrState")
val future = alterIsrManager.submit(
- new TopicIdPartition(topicId.getOrElse(Uuid.ZERO_UUID), topicPartition),
+ new org.apache.kafka.server.common.TopicIdPartition(topicId.getOrElse(throw new IllegalStateException("Topic id not set")), topicPartition.partition),
Review Comment:
I think this makes sense. I'm not sure exactly though what the error handling is here when an error is thrown. It seems like the futures here are not always checked. (Also wondering if we want to include more information about the partition in the error message)
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949680492
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1813,7 +1813,7 @@ class Partition(val topicPartition: TopicPartition,
private def submitAlterPartition(proposedIsrState: PendingPartitionChange): CompletableFuture[LeaderAndIsr] = {
debug(s"Submitting ISR state change $proposedIsrState")
val future = alterIsrManager.submit(
- new TopicIdPartition(topicId.getOrElse(Uuid.ZERO_UUID), topicPartition),
+ new org.apache.kafka.server.common.TopicIdPartition(topicId.getOrElse(throw new IllegalStateException("Topic id not set")), topicPartition.partition),
Review Comment:
> so the error thrown is just a way to make sure we are not missing something.
Yeah, that makes sense. The main thing I was wanting to check is that the error doesn't get swallowed somewhere. 👍
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953658601
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -495,8 +495,13 @@ BrokerFeature processRegistrationFeature(
FinalizedControllerFeatures finalizedFeatures,
BrokerRegistrationRequestData.Feature feature
) {
- int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ? 1 : 0; // The default value for MetadataVersion is 1 not 0.
- short finalized = finalizedFeatures.versionOrDefault(feature.name(), (short) defaultVersion);
+ // MetadataVersion has no default while the other features default to `0`
+ short finalized;
+ if (feature.name().equals(MetadataVersion.FEATURE_NAME))
+ finalized = finalizedFeatures.get(feature.name()).orElseThrow(() -> new RuntimeException());
Review Comment:
I think we should throw a more specific exception or include a clear error message.
Update: This is already resolved :)
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953416344
##########
server-common/src/main/java/org/apache/kafka/server/common/UnitTestFeatureVersion.java:
##########
@@ -321,7 +321,7 @@ public Map<String, Short> dependencies() {
* The feature is used to test the default value has MV dependency that is behind the bootstrap MV.
*/
public enum FV7 implements FeatureVersion {
- UT_FV7_0(0, MetadataVersion.MINIMUM_KRAFT_VERSION, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
+ UT_FV7_0(0, MetadataVersion.MINIMUM_VERSION, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
Review Comment:
I'll change the test to check the error message so that this doesn't happen in the future.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949581070
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1813,7 +1813,7 @@ class Partition(val topicPartition: TopicPartition,
private def submitAlterPartition(proposedIsrState: PendingPartitionChange): CompletableFuture[LeaderAndIsr] = {
debug(s"Submitting ISR state change $proposedIsrState")
val future = alterIsrManager.submit(
- new TopicIdPartition(topicId.getOrElse(Uuid.ZERO_UUID), topicPartition),
+ new org.apache.kafka.server.common.TopicIdPartition(topicId.getOrElse(throw new IllegalStateException("Topic id not set")), topicPartition.partition),
Review Comment:
> I think this makes sense. I'm not sure exactly though what the error handling is here when an error is thrown. It seems like the futures here are not always checked. (Also wondering if we want to include more information about the partition in the error message)
My assumption is that the topic id is actually always set when this method is called, so the error thrown is just a way to make sure we are not missing something. Once we refactor things, this will be more obvious, but it would be nice to avoid blocking 4.0 on 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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950083279
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java:
##########
@@ -128,18 +126,6 @@ public void testReassignment() throws Exception {
executeAndVerifyReassignment();
}
- @ClusterTests({
- @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, metadataVersion = IBP_3_3_IV0)
- })
- public void testReassignmentWithAlterPartitionDisabled() throws Exception {
- // Test reassignment when the IBP is on an older version which does not use
- // the `AlterPartition` API. In this case, the controller will register individual
- // watches for each reassigning partition so that the reassignment can be
- // completed as soon as the ISR is expanded.
Review Comment:
Makes sense. To confirm, you're saying we can remove this test _and_ we don't need any other follow-up since the comment was for the zk controller. Right?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950212078
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadataTest.java:
##########
@@ -126,13 +126,13 @@ public void testFeatureLevelForFeature() {
static final List<ApiMessageAndVersion> RECORDS_WITH_OLD_METADATA_VERSION = Collections.singletonList(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(FEATURE_NAME).
- setFeatureLevel(IBP_3_0_IV1.featureLevel()), (short) 0));
+ setFeatureLevel(MetadataVersionTestUtils.IBP_3_0_IV1_FEATURE_LEVEL), (short) 0));
@Test
public void testFromRecordsListWithOldMetadataVersion() {
RuntimeException exception = assertThrows(RuntimeException.class,
() -> BootstrapMetadata.fromRecords(RECORDS_WITH_OLD_METADATA_VERSION, "quux"));
- assertEquals("Bootstrap metadata.version before 3.3-IV0 are not supported. Can't load " +
- "metadata from quux", exception.getMessage());
+ assertEquals("No MetadataVersion with feature level 1. Valid feature levels are from 7 to 25.",
Review Comment:
It's a good question and I wondered the same. One challenge is that the test ends up pretty similar to the actual code. But I don't feel strongly. Perhaps a middle ground is to use the constant for the latest version - which changes frequently. The minimum version changes very infrequently and it's probably good to find out if we accidentally change it. Thoughts?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950212652
##########
core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java:
##########
@@ -380,7 +380,7 @@ public void testUpdateBrokerConfigNotAffectedByInvalidConfig() {
@ClusterTest(
// Must be at greater than 1MB per cleaner thread, set to 2M+2 so that we can set 2 cleaner threads.
serverProperties = {@ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "2097154")},
- metadataVersion = MetadataVersion.IBP_3_3_IV0
+ metadataVersion = MetadataVersion.IBP_3_9_IV0
Review Comment:
Since we cannot use `MINIMUM_VERSION` (see my other comment), I picked a recent version so there is less chance of this code changing. If it's confusing, I can change it to `IBP_3_3_IV3`.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1956608864
##########
metadata/src/main/java/org/apache/kafka/image/node/MetadataNode.java:
##########
@@ -53,7 +53,8 @@ default void print(MetadataNodePrinter printer) {
for (String name : names) {
printer.enterNode(name);
MetadataNode child = child(name);
- child.print(printer);
+ if (child != null)
+ child.print(printer);
Review Comment:
Interesting. Shall we file a separate JIRA ticket for that? Perhaps we can wait for @mumrah to confirm.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950206038
##########
server-common/src/main/java/org/apache/kafka/server/common/Feature.java:
##########
@@ -155,18 +155,16 @@ public FeatureVersion fromFeatureLevel(short level,
* For example, say feature X level x relies on feature Y level y:
* if feature X >= x then throw an error if feature Y < y.
*
- * All feature levels above 0 in kraft require metadata.version=4 (IBP_3_3_IV0) in order to write the feature records to the cluster.
- *
* @param feature the feature we are validating
* @param features the feature versions we have (or want to set)
* @throws IllegalArgumentException if the feature is not valid
*/
public static void validateVersion(FeatureVersion feature, Map<String, Short> features) {
Short metadataVersion = features.get(MetadataVersion.FEATURE_NAME);
- if (feature.featureLevel() >= 1 && (metadataVersion == null || metadataVersion < MetadataVersion.IBP_3_3_IV0.featureLevel()))
+ if (feature.featureLevel() >= 1 && (metadataVersion == null || metadataVersion < MetadataVersion.MINIMUM_VERSION.featureLevel()))
Review Comment:
Since we are not actually using `MetadataVersion` here, it seems possible for callers to pass invalid values. That is why I kept the validation.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950204878
##########
metadata/src/main/java/org/apache/kafka/controller/ActivationRecordsGenerator.java:
##########
@@ -137,10 +137,10 @@ static ControllerResult<Void> recordsForNonEmptyLog(
}
}
- if (curMetadataVersion.equals(MetadataVersion.MINIMUM_KRAFT_VERSION)) {
+ if (curMetadataVersion.equals(MetadataVersion.MINIMUM_VERSION)) {
Review Comment:
Good catch. We have two options: (1) Remove this log or (2) Pass a parameter indicating whether `metadata.version` feature level record is present. Do you have an opinion on which way is better? I'll also take a closer look to understand whether the second option is worth it.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950222735
##########
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java:
##########
@@ -1524,31 +1507,6 @@ public void close() throws Exception {
(short) 0)),
"test bootstrap");
- @Test
- public void testUpgradeFromPreProductionVersion() throws Exception {
Review Comment:
Upgrades from pre production versions are no longer supported.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949583609
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1813,7 +1813,7 @@ class Partition(val topicPartition: TopicPartition,
private def submitAlterPartition(proposedIsrState: PendingPartitionChange): CompletableFuture[LeaderAndIsr] = {
debug(s"Submitting ISR state change $proposedIsrState")
val future = alterIsrManager.submit(
- new TopicIdPartition(topicId.getOrElse(Uuid.ZERO_UUID), topicPartition),
+ new org.apache.kafka.server.common.TopicIdPartition(topicId.getOrElse(throw new IllegalStateException("Topic id not set")), topicPartition.partition),
Review Comment:
> if we want to include more information about the partition in the error message
Good idea - I will add the `topicPartition` to the error message.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949970524
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadataTest.java:
##########
@@ -126,7 +125,7 @@ public void testFeatureLevelForFeature() {
static final List<ApiMessageAndVersion> RECORDS_WITH_OLD_METADATA_VERSION = Collections.singletonList(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(FEATURE_NAME).
- setFeatureLevel(IBP_3_0_IV1.featureLevel()), (short) 0));
+ setFeatureLevel((short) 1), (short) 0));
Review Comment:
This is intended - see the name of the records (RECORDS_WITH_OLD_METADATA_VERSION). I did update this to refer to a constant.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1951022774
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -378,7 +331,8 @@ public static MetadataVersion fromFeatureLevel(short version) {
return metadataVersion;
}
}
- throw new IllegalArgumentException("No MetadataVersion with feature level " + version);
+ throw new IllegalArgumentException("No MetadataVersion with feature level " + version + ". Valid feature levels are from "
+ + MINIMUM_VERSION.featureLevel + " to " + latestTesting().featureLevel + ".");
Review Comment:
The code actually considers all metadata values - the error message s consistent with 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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1952871969
##########
metadata/src/main/java/org/apache/kafka/image/FeaturesDelta.java:
##########
@@ -58,7 +59,14 @@ public void finishSnapshot() {
public void replay(FeatureLevelRecord record) {
if (record.name().equals(MetadataVersion.FEATURE_NAME)) {
- metadataVersionChange = MetadataVersion.fromFeatureLevel(record.featureLevel());
+ // Support for the `metadata.version` feature flag was added in IBP_3_3_IV0, so it's possible (but unlikely) that we read
+ // records with a feature level that is no longer supported for clusters that used a pre-release version of 3.3.0.
+ // We automatically fallback to IBP_3_3_IV3 in that case. We use explicit versions instead of `MINIMUM_VERSION` because
+ // we want to force an explicit decision if we change `MetadataVersion.MINIMUM_VERSION` in the future.
+ if (record.featureLevel() >= MINIMUM_PERSISTED_FEATURE_LEVEL && record.featureLevel() <= MetadataVersion.IBP_3_3_IV3.featureLevel())
+ metadataVersionChange = MetadataVersion.IBP_3_3_IV3;
Review Comment:
@junrao @cmccabe Thoughts on this fallback? It seems like quite a rare edge case (since it requires usage of a pre-release version and for the older metadata version not to be compacted away if I understand correctly), but it seems fair to never fail while reading the log and the fallback code is pretty simple.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953404839
##########
metadata/src/main/java/org/apache/kafka/image/FeaturesDelta.java:
##########
@@ -58,7 +59,14 @@ public void finishSnapshot() {
public void replay(FeatureLevelRecord record) {
if (record.name().equals(MetadataVersion.FEATURE_NAME)) {
- metadataVersionChange = MetadataVersion.fromFeatureLevel(record.featureLevel());
+ // Support for the `metadata.version` feature flag was added in IBP_3_3_IV0, so it's possible (but unlikely) that we read
+ // records with a feature level that is no longer supported for clusters that used a pre-release version of 3.3.0.
+ // We automatically fallback to IBP_3_3_IV3 in that case. We use explicit versions instead of `MINIMUM_VERSION` because
+ // we want to force an explicit decision if we change `MetadataVersion.MINIMUM_VERSION` in the future.
+ if (record.featureLevel() >= MINIMUM_PERSISTED_FEATURE_LEVEL && record.featureLevel() <= MetadataVersion.IBP_3_3_IV3.featureLevel())
+ metadataVersionChange = MetadataVersion.IBP_3_3_IV3;
Review Comment:
There is no guarantee, but the expected MV is IBP_3_3_IV3 and above. If this is not the case, it seems easier to just fail instead of changing it silently.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on PR #18845:
URL: https://github.com/apache/kafka/pull/18845#issuecomment-2668703940
Thanks for pushing this over the line @ijuma. It's great to see these cleanups and removals. "IBP" is all but a memory now :)
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18845:
URL: https://github.com/apache/kafka/pull/18845#issuecomment-2667217579
@chia7712 I addressed your comments.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1948437894
##########
metadata/src/main/java/org/apache/kafka/image/node/MetadataNode.java:
##########
@@ -53,7 +53,8 @@ default void print(MetadataNodePrinter printer) {
for (String name : names) {
printer.enterNode(name);
MetadataNode child = child(name);
- child.print(printer);
+ if (child != null)
+ child.print(printer);
Review Comment:
Fixes a NPE I noticed while debugging a failing test.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953658601
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -495,8 +495,13 @@ BrokerFeature processRegistrationFeature(
FinalizedControllerFeatures finalizedFeatures,
BrokerRegistrationRequestData.Feature feature
) {
- int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ? 1 : 0; // The default value for MetadataVersion is 1 not 0.
- short finalized = finalizedFeatures.versionOrDefault(feature.name(), (short) defaultVersion);
+ // MetadataVersion has no default while the other features default to `0`
+ short finalized;
+ if (feature.name().equals(MetadataVersion.FEATURE_NAME))
+ finalized = finalizedFeatures.get(feature.name()).orElseThrow(() -> new RuntimeException());
Review Comment:
I think we should throw a more specific exception or include a clear error message.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950087565
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -86,12 +84,7 @@ BootstrapMetadata readFromConfiguration() {
return BootstrapMetadata.fromVersion(MetadataVersion.latestProduction(), "the default bootstrap");
}
MetadataVersion version = MetadataVersion.fromVersionString(ibp.get());
- if (version.isLessThan(MINIMUM_BOOTSTRAP_VERSION)) {
Review Comment:
I guess this means the answer to my question is that the fallback is not important.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950998881
##########
tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java:
##########
@@ -250,66 +250,77 @@ public void testHandleDescribe() {
}
});
assertEquals(format("Feature: foo.bar\tSupportedMinVersion: 0\tSupportedMaxVersion: 10\tFinalizedVersionLevel: 5\tEpoch: 123%n" +
- "Feature: metadata.version\tSupportedMinVersion: 3.3-IV0\tSupportedMaxVersion: 3.3-IV3\tFinalizedVersionLevel: 3.3-IV2\tEpoch: 123"), describeResult);
+ "Feature: metadata.version\tSupportedMinVersion: 3.3-IV3\tSupportedMaxVersion: 3.5-IV0\tFinalizedVersionLevel: 3.4-IV0\tEpoch: 123"), describeResult);
}
@Test
- public void testHandleUpgrade() {
+ public void testHandleUpgradeToUnsupportedMetadataVersion() {
Map<String, Object> namespace = new HashMap<>();
namespace.put("metadata", "3.3-IV1");
Review Comment:
I don't think we need to do that - the feature levels are opaque, but the metadata versions are clear and don't need a constant for usage in tests (they would need a constant within non test code but we don't use this in non test code).
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950994148
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -368,7 +320,8 @@ public static MetadataVersion fromVersionString(String versionString) {
key = String.join(".", Arrays.copyOfRange(versionSegments, 0, numSegments));
}
return Optional.ofNullable(IBP_VERSIONS.get(key)).orElseThrow(() ->
- new IllegalArgumentException("Version " + versionString + " is not a valid version")
+ new IllegalArgumentException("Version " + versionString + " is not a valid version. The minimum version is " + MINIMUM_VERSION
+ + " and the maximum version is " + latestTesting())
Review Comment:
As you said, that seems to be inconsistent with the docs for `StorageTool` - perhaps we should file an issue to fix that in its own PR.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949883386
##########
server-common/src/test/java/org/apache/kafka/server/common/MetadataVersionTest.java:
##########
@@ -170,41 +140,17 @@ public void testPrevious() {
@Test
public void testMetadataChanged() {
- assertFalse(MetadataVersion.checkIfMetadataChanged(IBP_3_2_IV0, IBP_3_2_IV0));
- assertTrue(MetadataVersion.checkIfMetadataChanged(IBP_3_2_IV0, IBP_3_1_IV0));
- assertTrue(MetadataVersion.checkIfMetadataChanged(IBP_3_2_IV0, IBP_3_0_IV1));
- assertTrue(MetadataVersion.checkIfMetadataChanged(IBP_3_2_IV0, IBP_3_0_IV1));
- assertTrue(MetadataVersion.checkIfMetadataChanged(IBP_3_3_IV1, IBP_3_3_IV0));
+ assertTrue(MetadataVersion.checkIfMetadataChanged(IBP_3_4_IV0, IBP_3_3_IV3));
+ assertFalse(MetadataVersion.checkIfMetadataChanged(IBP_3_5_IV0, IBP_3_4_IV0));
+ assertFalse(MetadataVersion.checkIfMetadataChanged(IBP_3_5_IV1, IBP_3_5_IV0));
+ assertTrue(MetadataVersion.checkIfMetadataChanged(IBP_3_6_IV0, IBP_3_5_IV1));
+ assertTrue(MetadataVersion.checkIfMetadataChanged(IBP_3_6_IV1, IBP_3_6_IV0));
// Check that argument order doesn't matter
- assertTrue(MetadataVersion.checkIfMetadataChanged(IBP_3_1_IV0, IBP_3_2_IV0));
- assertTrue(MetadataVersion.checkIfMetadataChanged(IBP_3_0_IV1, IBP_3_2_IV0));
- }
-
- @Test
- public void testKRaftVersions() {
Review Comment:
This did not seem to add much value after the changes in this PR.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949577245
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1813,7 +1813,7 @@ class Partition(val topicPartition: TopicPartition,
private def submitAlterPartition(proposedIsrState: PendingPartitionChange): CompletableFuture[LeaderAndIsr] = {
debug(s"Submitting ISR state change $proposedIsrState")
val future = alterIsrManager.submit(
- new TopicIdPartition(topicId.getOrElse(Uuid.ZERO_UUID), topicPartition),
+ new org.apache.kafka.server.common.TopicIdPartition(topicId.getOrElse(throw new IllegalStateException("Topic id not set")), topicPartition.partition),
Review Comment:
@dajac I agree with you, but I thought that should be done in a separate PR as I think it will require some refactoring given the way various things interact right now. And perhaps it can be done in trunk only (no backport to 4.0 required).
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1956516319
##########
metadata/src/main/java/org/apache/kafka/image/loader/MetadataBatchLoader.java:
##########
@@ -100,7 +100,7 @@ public boolean hasSeenRecord() {
*/
public final void resetToImage(MetadataImage image) {
this.image = image;
- this.hasSeenRecord = true;
+ this.hasSeenRecord = !image.isEmpty();
Review Comment:
This was an existing issue that somehow became easier to trigger due to these changes. Thanks to @mumrah for figuring out that this was the reason for the remaining 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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1956777095
##########
metadata/src/main/java/org/apache/kafka/image/node/MetadataNode.java:
##########
@@ -53,7 +53,8 @@ default void print(MetadataNodePrinter printer) {
for (String name : names) {
printer.enterNode(name);
MetadataNode child = child(name);
- child.print(printer);
+ if (child != null)
+ child.print(printer);
Review Comment:
@ijuma @mumrah thanks for your response. open https://issues.apache.org/jira/browse/KAFKA-18803 for it
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1960706340
##########
metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java:
##########
@@ -718,30 +710,14 @@ public void testRegistrationWithUnsupportedMetadataVersion() {
setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
Collections.singleton(new BrokerRegistrationRequestData.Feature().
setName(MetadataVersion.FEATURE_NAME).
- setMinSupportedVersion(MetadataVersion.IBP_3_3_IV3.featureLevel()).
- setMaxSupportedVersion(MetadataVersion.IBP_3_3_IV3.featureLevel())).iterator())).
+ setMinSupportedVersion(MetadataVersion.IBP_3_4_IV0.featureLevel()).
+ setMaxSupportedVersion(MetadataVersion.IBP_3_4_IV0.featureLevel())).iterator())).
setIncarnationId(Uuid.fromString("0H4fUu1xQEKXFYwB1aBjhg")),
123L,
featureControl.finalizedFeatures(Long.MAX_VALUE),
false)).getMessage());
}
- @Test
- public void testRegisterControlWithOlderMetadataVersion() {
- FeatureControlManager featureControl = new FeatureControlManager.Builder().
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
Review Comment:
Good catch, I had missed that. I'll add the test back with `IBP_3_6_IV2` since that's the newest IBP that still fails.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18845:
URL: https://github.com/apache/kafka/pull/18845#issuecomment-2668661378
I will go ahead and merge this. If @junrao has any additional comments, I will submit a separate PR to address those.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1949804134
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectoryTest.java:
##########
@@ -81,18 +81,18 @@ public void testReadFromEmptyConfiguration() throws Exception {
@Test
public void testReadFromConfigurationWithAncientVersion() throws Exception {
try (BootstrapTestDirectory testDirectory = new BootstrapTestDirectory().createDirectory()) {
- assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.MINIMUM_BOOTSTRAP_VERSION,
- "the minimum version bootstrap with metadata.version 3.3-IV0"),
+ assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.MINIMUM_VERSION,
+ "the minimum version bootstrap with metadata.version 3.3-IV3"),
Review Comment:
The usage of 3.0 in the next line and in `testReadFromConfigurationFile` need to be adjusted.
##########
metadata/src/main/java/org/apache/kafka/image/FeaturesImage.java:
##########
@@ -56,7 +53,7 @@ public FeaturesImage(
public boolean isEmpty() {
return finalizedVersions.isEmpty() &&
- metadataVersion.equals(MetadataVersion.MINIMUM_KRAFT_VERSION);
+ metadataVersion.equals(MetadataVersion.MINIMUM_VERSION);
Review Comment:
Hmm, metadata version 3.3 can be set explicitly in KRaft. Is this still correct @cmccabe ?
##########
server-common/src/main/java/org/apache/kafka/server/common/Feature.java:
##########
@@ -298,7 +296,7 @@ public static void validateDefaultValueAndLatestProductionValue(
for (MetadataVersion metadataVersion: MetadataVersion.values()) {
// Only checking the kraft metadata versions.
- if (metadataVersion.compareTo(MetadataVersion.MINIMUM_KRAFT_VERSION) < 0) {
+ if (metadataVersion.compareTo(MetadataVersion.MINIMUM_VERSION) < 0) {
Review Comment:
This is no longer needed since all passed in metadataVersion are >= `MetadataVersion.MINIMUM_VERSION`.
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectoryTest.java:
##########
@@ -81,18 +81,17 @@ public void testReadFromEmptyConfiguration() throws Exception {
@Test
public void testReadFromConfigurationWithAncientVersion() throws Exception {
try (BootstrapTestDirectory testDirectory = new BootstrapTestDirectory().createDirectory()) {
- assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.MINIMUM_BOOTSTRAP_VERSION,
- "the minimum version bootstrap with metadata.version 3.3-IV0"),
+ assertThrows(IllegalArgumentException.class, () ->
new BootstrapDirectory(testDirectory.path(), Optional.of("3.0")).read());
}
}
@Test
public void testReadFromConfiguration() throws Exception {
try (BootstrapTestDirectory testDirectory = new BootstrapTestDirectory().createDirectory()) {
- assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_3_IV2,
- "the configured bootstrap with metadata.version 3.3-IV2"),
- new BootstrapDirectory(testDirectory.path(), Optional.of("3.3-IV2")).read());
+ assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_3_IV3,
+ "the configured bootstrap with metadata.version 3.3-IV3"),
+ new BootstrapDirectory(testDirectory.path(), Optional.of("3.3-IV3")).read());
Review Comment:
Should we change `3.0-IV0` in `testReadFromConfigurationFile` below?
##########
core/src/test/scala/integration/kafka/coordinator/transaction/ProducerIntegrationTest.scala:
##########
@@ -55,7 +55,7 @@ import scala.jdk.CollectionConverters._
class ProducerIntegrationTest {
@ClusterTests(Array(
- new ClusterTest(metadataVersion = MetadataVersion.IBP_3_3_IV0)
+ new ClusterTest(metadataVersion = MetadataVersion.IBP_3_3_IV3)
Review Comment:
Should we use MINIMUM_VERSION to avoid keeping changing it in the future?
##########
server-common/src/main/java/org/apache/kafka/server/common/UnitTestFeatureVersion.java:
##########
@@ -321,7 +321,7 @@ public Map<String, Short> dependencies() {
* The feature is used to test the default value has MV dependency that is behind the bootstrap MV.
*/
public enum FV7 implements FeatureVersion {
- UT_FV7_0(0, MetadataVersion.MINIMUM_KRAFT_VERSION, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
+ UT_FV7_0(0, MetadataVersion.MINIMUM_VERSION, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
Review Comment:
The error message in FeatureTest with respect to 3.0 needs to be adjusted.
```
"Feature UNIT_TEST_VERSION_7 has default FeatureVersion UT_FV7_0 when MV=3.0-IV1 with " +
"MV dependency 3.7-IV0 that is behind its bootstrap MV 3.0-IV1.");
```
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadataTest.java:
##########
@@ -126,13 +126,13 @@ public void testFeatureLevelForFeature() {
static final List<ApiMessageAndVersion> RECORDS_WITH_OLD_METADATA_VERSION = Collections.singletonList(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(FEATURE_NAME).
- setFeatureLevel(IBP_3_0_IV1.featureLevel()), (short) 0));
+ setFeatureLevel(MetadataVersionTestUtils.IBP_3_0_IV1_FEATURE_LEVEL), (short) 0));
@Test
public void testFromRecordsListWithOldMetadataVersion() {
RuntimeException exception = assertThrows(RuntimeException.class,
() -> BootstrapMetadata.fromRecords(RECORDS_WITH_OLD_METADATA_VERSION, "quux"));
- assertEquals("Bootstrap metadata.version before 3.3-IV0 are not supported. Can't load " +
- "metadata from quux", exception.getMessage());
+ assertEquals("No MetadataVersion with feature level 1. Valid feature levels are from 7 to 25.",
Review Comment:
Could we use defined constants for feature level to avoid keeping changing it in the future?
##########
core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java:
##########
@@ -380,7 +380,7 @@ public void testUpdateBrokerConfigNotAffectedByInvalidConfig() {
@ClusterTest(
// Must be at greater than 1MB per cleaner thread, set to 2M+2 so that we can set 2 cleaner threads.
serverProperties = {@ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "2097154")},
- metadataVersion = MetadataVersion.IBP_3_3_IV0
+ metadataVersion = MetadataVersion.IBP_3_9_IV0
Review Comment:
Hmm, why do we choose IBP_3_9_IV0 here?
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -44,25 +44,6 @@
*/
public enum MetadataVersion {
- // Introduce ListOffsets V7 which supports listing offsets by max timestamp (KIP-734)
- // Assume message format version is 3.0 (KIP-724)
- IBP_3_0_IV1(1, "3.0", "IV1", true),
-
- // Adds topic IDs to Fetch requests/responses (KIP-516)
- IBP_3_1_IV0(2, "3.1", "IV0", false),
-
- // Support for leader recovery for unclean leader election (KIP-704)
- IBP_3_2_IV0(3, "3.2", "IV0", true),
-
- // Support for metadata.version feature flag and Removes min_version_level from the finalized version range that is written to ZooKeeper (KIP-778)
- IBP_3_3_IV0(4, "3.3", "IV0", false),
-
- // Support NoopRecord for the cluster metadata log (KIP-835)
- IBP_3_3_IV1(5, "3.3", "IV1", true),
-
- // In KRaft mode, use BrokerRegistrationChangeRecord instead of UnfenceBrokerRecord and FenceBrokerRecord.
- IBP_3_3_IV2(6, "3.3", "IV2", true),
Review Comment:
The references of the old MV in AclsImage need to be adjusted.
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -220,18 +187,6 @@ public boolean isElrSupported() {
return this.isAtLeast(IBP_4_0_IV1);
}
- public boolean isKRaftSupported() {
- return this.featureLevel > 0;
- }
-
- public boolean isBrokerRegistrationChangeRecordSupported() {
Review Comment:
The comment in BrokerRegistrationTrackerTest.java with respect to 3.0 needs to be adjusted.
` // No calls are made because MetadataVersion is 3.0-IV1 initially`
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -495,7 +495,9 @@ BrokerFeature processRegistrationFeature(
FinalizedControllerFeatures finalizedFeatures,
BrokerRegistrationRequestData.Feature feature
) {
- int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ? 1 : 0; // The default value for MetadataVersion is 1 not 0.
+ // The default value for MetadataVersion changes over time while other features start at `0`
+ int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ?
+ MetadataVersion.MINIMUM_VERSION.featureLevel() : 0;
Review Comment:
@cmccabe : Should the finalized MetadataVersion always be present in any release upgradable to 4.0?
##########
core/src/test/scala/unit/kafka/server/AlterPartitionManagerTest.scala:
##########
@@ -552,24 +535,10 @@ class AlterPartitionManagerTest {
}
object AlterPartitionManagerTest {
- def provideMetadataVersions(): JStream[MetadataVersion] = {
+ def provideLeaderRecoveryState(): JStream[Arguments] = {
JStream.of(
- // Supports KIP-903: include broker epoch in AlterPartition request
- IBP_3_5_IV1,
Review Comment:
Should we change to test both IBP_3_3_IV3 and IBP_3_5_IV1 here, @CalvinConfluent ?
##########
server-common/src/main/java/org/apache/kafka/server/common/Feature.java:
##########
@@ -155,18 +155,16 @@ public FeatureVersion fromFeatureLevel(short level,
* For example, say feature X level x relies on feature Y level y:
* if feature X >= x then throw an error if feature Y < y.
*
- * All feature levels above 0 in kraft require metadata.version=4 (IBP_3_3_IV0) in order to write the feature records to the cluster.
- *
* @param feature the feature we are validating
* @param features the feature versions we have (or want to set)
* @throws IllegalArgumentException if the feature is not valid
*/
public static void validateVersion(FeatureVersion feature, Map<String, Short> features) {
Short metadataVersion = features.get(MetadataVersion.FEATURE_NAME);
- if (feature.featureLevel() >= 1 && (metadataVersion == null || metadataVersion < MetadataVersion.IBP_3_3_IV0.featureLevel()))
+ if (feature.featureLevel() >= 1 && (metadataVersion == null || metadataVersion < MetadataVersion.MINIMUM_VERSION.featureLevel()))
Review Comment:
It seems that we could remove this statement completely since the callers will always have metadataVersion set to >= 3.3. Is that correct @jolshan ?
##########
core/src/main/scala/kafka/server/AlterPartitionManager.scala:
##########
@@ -224,53 +215,37 @@ class DefaultAlterPartitionManager(
private def buildRequest(
inflightAlterPartitionItems: Seq[AlterPartitionItem],
brokerEpoch: Long
- ): (AlterPartitionRequest.Builder, mutable.Map[Uuid, String]) = {
- val metadataVersion = metadataVersionSupplier()
- // We build this mapping in order to map topic id back to their name when we
- // receive the response. We cannot rely on the metadata cache for this because
- // the metadata cache is updated after the partition state so it might not know
- // yet about a topic id already used here.
- val topicNamesByIds = mutable.HashMap[Uuid, String]()
-
+ ): AlterPartitionRequest.Builder = {
val message = new AlterPartitionRequestData()
.setBrokerId(brokerId)
.setBrokerEpoch(brokerEpoch)
- inflightAlterPartitionItems.groupBy(_.topicIdPartition.topic).foreach { case (topicName, items) =>
- val topicId = items.head.topicIdPartition.topicId
- topicNamesByIds(topicId) = topicName
-
+ inflightAlterPartitionItems.groupBy(_.topicIdPartition.topicId).foreach { case (topicId, items) =>
// Both the topic name and the topic id are set here because at this stage
// we don't know which version of the request will be used.
- val topicData = new AlterPartitionRequestData.TopicData()
- .setTopicName(topicName)
- .setTopicId(topicId)
+ val topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId)
Review Comment:
Could we adjust the comment on topic name above?
##########
metadata/src/main/java/org/apache/kafka/controller/ActivationRecordsGenerator.java:
##########
@@ -137,10 +137,10 @@ static ControllerResult<Void> recordsForNonEmptyLog(
}
}
- if (curMetadataVersion.equals(MetadataVersion.MINIMUM_KRAFT_VERSION)) {
+ if (curMetadataVersion.equals(MetadataVersion.MINIMUM_VERSION)) {
Review Comment:
MetadataVersion.MINIMUM_VERSION now supports metadata.version feature level record. So, the following log message could be confusing.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1948421007
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1813,7 +1813,7 @@ class Partition(val topicPartition: TopicPartition,
private def submitAlterPartition(proposedIsrState: PendingPartitionChange): CompletableFuture[LeaderAndIsr] = {
debug(s"Submitting ISR state change $proposedIsrState")
val future = alterIsrManager.submit(
- new TopicIdPartition(topicId.getOrElse(Uuid.ZERO_UUID), topicPartition),
+ new org.apache.kafka.server.common.TopicIdPartition(topicId.getOrElse(throw new IllegalStateException("Topic id not set")), topicPartition.partition),
Review Comment:
It's not safe to use `Uuid.ZERO_UUID` with the changes in this PR and hence we now throw an exception is the topic is is not known. @jolshan Is this OK?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953113246
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -495,7 +495,9 @@ BrokerFeature processRegistrationFeature(
FinalizedControllerFeatures finalizedFeatures,
BrokerRegistrationRequestData.Feature feature
) {
- int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ? 1 : 0; // The default value for MetadataVersion is 1 not 0.
+ // The default value for MetadataVersion changes over time while other features start at `0`
+ int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ?
+ MetadataVersion.MINIMUM_VERSION.featureLevel() : 0;
Review Comment:
If we're dropping support for everything before 3.3 we can always count on having a FeatureRecord for `metadata.version`. However doing this may require a lot of code changes in tests, since many junit tests may initialize with an empty log. (Or maybe not... I would have to check?)
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953113246
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -495,7 +495,9 @@ BrokerFeature processRegistrationFeature(
FinalizedControllerFeatures finalizedFeatures,
BrokerRegistrationRequestData.Feature feature
) {
- int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ? 1 : 0; // The default value for MetadataVersion is 1 not 0.
+ // The default value for MetadataVersion changes over time while other features start at `0`
+ int defaultVersion = feature.name().equals(MetadataVersion.FEATURE_NAME) ?
+ MetadataVersion.MINIMUM_VERSION.featureLevel() : 0;
Review Comment:
If we're dropping support for everything before 3.3 we can always count on having a FeatureRecord for `metadata.version`. However doing this may require a lot of code changes since many junit tests may initialize with an empty log. (Or maybe not... I would have to check?)
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1960688656
##########
metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java:
##########
@@ -718,30 +710,14 @@ public void testRegistrationWithUnsupportedMetadataVersion() {
setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
Collections.singleton(new BrokerRegistrationRequestData.Feature().
setName(MetadataVersion.FEATURE_NAME).
- setMinSupportedVersion(MetadataVersion.IBP_3_3_IV3.featureLevel()).
- setMaxSupportedVersion(MetadataVersion.IBP_3_3_IV3.featureLevel())).iterator())).
+ setMinSupportedVersion(MetadataVersion.IBP_3_4_IV0.featureLevel()).
+ setMaxSupportedVersion(MetadataVersion.IBP_3_4_IV0.featureLevel())).iterator())).
setIncarnationId(Uuid.fromString("0H4fUu1xQEKXFYwB1aBjhg")),
123L,
featureControl.finalizedFeatures(Long.MAX_VALUE),
false)).getMessage());
}
- @Test
- public void testRegisterControlWithOlderMetadataVersion() {
- FeatureControlManager featureControl = new FeatureControlManager.Builder().
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
Review Comment:
I deleted that because we have `testRegistrationWithUnsupportedMetadataVersion`. Does that address your concern?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18845:
URL: https://github.com/apache/kafka/pull/18845#issuecomment-2666525097
@junrao @chia7712 The tests for JDK 23 were all passing for the last run (the JDK 17 build failed, but it didn't look related to this PR).
I think there's only one outstanding issue: how to handle the case if we read an old metadata version from the log. I'll update the relevant comment later today. Does the rest look good to you?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950210926
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectoryTest.java:
##########
@@ -81,18 +81,17 @@ public void testReadFromEmptyConfiguration() throws Exception {
@Test
public void testReadFromConfigurationWithAncientVersion() throws Exception {
try (BootstrapTestDirectory testDirectory = new BootstrapTestDirectory().createDirectory()) {
- assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.MINIMUM_BOOTSTRAP_VERSION,
- "the minimum version bootstrap with metadata.version 3.3-IV0"),
+ assertThrows(IllegalArgumentException.class, () ->
new BootstrapDirectory(testDirectory.path(), Optional.of("3.0")).read());
}
}
@Test
public void testReadFromConfiguration() throws Exception {
try (BootstrapTestDirectory testDirectory = new BootstrapTestDirectory().createDirectory()) {
- assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_3_IV2,
- "the configured bootstrap with metadata.version 3.3-IV2"),
- new BootstrapDirectory(testDirectory.path(), Optional.of("3.3-IV2")).read());
+ assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_3_IV3,
+ "the configured bootstrap with metadata.version 3.3-IV3"),
+ new BootstrapDirectory(testDirectory.path(), Optional.of("3.3-IV3")).read());
Review Comment:
I think the code you're referencing was removed in a subsequent commit.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950208594
##########
core/src/test/scala/unit/kafka/server/AlterPartitionManagerTest.scala:
##########
@@ -552,24 +535,10 @@ class AlterPartitionManagerTest {
}
object AlterPartitionManagerTest {
- def provideMetadataVersions(): JStream[MetadataVersion] = {
+ def provideLeaderRecoveryState(): JStream[Arguments] = {
JStream.of(
- // Supports KIP-903: include broker epoch in AlterPartition request
- IBP_3_5_IV1,
Review Comment:
I removed that method altogether since none of the tests used it anymore.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950727327
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -44,25 +44,6 @@
*/
Review Comment:
I think we should also update the java doc, replace old example like` IBP_0_10_0-IV0` to `IBP_3_3_IV3`
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -368,7 +320,8 @@ public static MetadataVersion fromVersionString(String versionString) {
key = String.join(".", Arrays.copyOfRange(versionSegments, 0, numSegments));
}
return Optional.ofNullable(IBP_VERSIONS.get(key)).orElseThrow(() ->
- new IllegalArgumentException("Version " + versionString + " is not a valid version")
+ new IllegalArgumentException("Version " + versionString + " is not a valid version. The minimum version is " + MINIMUM_VERSION
+ + " and the maximum version is " + latestTesting())
Review Comment:
This method is used in production code in `StorageTool.scala` and others, maybe we should not expose `latestTesting()` version, should use `latestProduction()`.
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -378,7 +331,8 @@ public static MetadataVersion fromFeatureLevel(short version) {
return metadataVersion;
}
}
- throw new IllegalArgumentException("No MetadataVersion with feature level " + version);
+ throw new IllegalArgumentException("No MetadataVersion with feature level " + version + ". Valid feature levels are from "
+ + MINIMUM_VERSION.featureLevel + " to " + latestTesting().featureLevel + ".");
Review Comment:
ditto, we should use latestProduction().
##########
tools/src/test/java/org/apache/kafka/tools/FeatureCommandTest.java:
##########
@@ -250,66 +250,77 @@ public void testHandleDescribe() {
}
});
assertEquals(format("Feature: foo.bar\tSupportedMinVersion: 0\tSupportedMaxVersion: 10\tFinalizedVersionLevel: 5\tEpoch: 123%n" +
- "Feature: metadata.version\tSupportedMinVersion: 3.3-IV0\tSupportedMaxVersion: 3.3-IV3\tFinalizedVersionLevel: 3.3-IV2\tEpoch: 123"), describeResult);
+ "Feature: metadata.version\tSupportedMinVersion: 3.3-IV3\tSupportedMaxVersion: 3.5-IV0\tFinalizedVersionLevel: 3.4-IV0\tEpoch: 123"), describeResult);
}
@Test
- public void testHandleUpgrade() {
+ public void testHandleUpgradeToUnsupportedMetadataVersion() {
Map<String, Object> namespace = new HashMap<>();
namespace.put("metadata", "3.3-IV1");
Review Comment:
nit: This `3.3-IV1` variable also can add to `MetadataVersionTestUtils` class.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1951007012
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -44,25 +44,6 @@
*/
Review Comment:
Good point - I can also simplify the parsing code since we no longer have metadata versions starting with `0.`.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1951432926
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -379,7 +373,7 @@ public void testCannotDowngradeBefore3_3_IV0() {
public void testCreateFeatureLevelRecords() {
Map<String, VersionRange> localSupportedFeatures = new HashMap<>();
localSupportedFeatures.put(MetadataVersion.FEATURE_NAME, VersionRange.of(
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.latestTesting().featureLevel()));
+ MetadataVersionTestUtils.IBP_3_0_IV1_FEATURE_LEVEL, MetadataVersion.latestTesting().featureLevel()));
Review Comment:
We probably should set the min to MetadataVersion.MINIMUM_VERSION?
##########
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java:
##########
@@ -310,7 +310,7 @@ public void testRemovePublisher() throws Exception {
singletonList(singletonList(new ApiMessageAndVersion(
new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(IBP_3_3_IV2.featureLevel()), (short) 0))));
+ setFeatureLevel(IBP_3_3_IV3.featureLevel()), (short) 0))));
Review Comment:
Since this can be any MV, could we just use MetadataVersion.MINIMUM_VERSION?
##########
metadata/src/test/java/org/apache/kafka/image/MetadataVersionChangeTest.java:
##########
@@ -30,31 +30,31 @@
@Timeout(value = 40)
public class MetadataVersionChangeTest {
- private static final MetadataVersionChange CHANGE_3_0_IV1_TO_3_3_IV0 =
- new MetadataVersionChange(IBP_3_0_IV1, IBP_3_3_IV0);
+ private static final MetadataVersionChange CHANGE_3_3_IV3_TO_3_6_IV0 =
Review Comment:
We just need two different MVs here. Could we use MetadataVersion.MINIMUM_VERSION and MetadataVersion.latestProduction?
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -240,38 +241,29 @@ public void testReplayRecords() {
private static final FeatureControlManager.Builder TEST_MANAGER_BUILDER1 =
new FeatureControlManager.Builder().
setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_3_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV2);
+ MetadataVersion.MINIMUM_VERSION.featureLevel(), MetadataVersion.IBP_3_6_IV0.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_4_IV0);
@Test
public void testApplyMetadataVersionChangeRecord() {
FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
manager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel()));
- assertEquals(MetadataVersion.IBP_3_3_IV3, manager.metadataVersion());
- }
-
- @Test
- public void testCannotDowngradeToVersionBeforeMinimumSupportedKraftVersion() {
- FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
- assertEquals(ControllerResult.of(Collections.emptyList(), new ApiError(Errors.INVALID_UPDATE_VERSION,
- "Invalid update version 3 for feature metadata.version. Local controller 0 only " +
- "supports versions 4-7")),
- manager.updateFeatures(
- singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_2_IV0.featureLevel()),
- singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.UNSAFE_DOWNGRADE),
- true));
+ setFeatureLevel(MetadataVersion.IBP_3_4_IV0.featureLevel()));
Review Comment:
Should we use MetadataVersion.MINIMUM_VERSION?
##########
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java:
##########
@@ -349,11 +349,11 @@ public void testLoadEmptySnapshot() throws Exception {
assertEquals(300L, loader.lastAppliedOffset());
assertEquals(new SnapshotManifest(new MetadataProvenance(300, 100, 4000, true), 3000000L),
publishers.get(0).latestSnapshotManifest);
- assertEquals(MetadataVersion.MINIMUM_KRAFT_VERSION,
+ assertEquals(MetadataVersion.MINIMUM_VERSION,
loader.metrics().currentMetadataVersion());
}
assertTrue(publishers.get(0).closed);
- assertEquals(MetadataVersion.IBP_3_0_IV1,
+ assertEquals(IBP_3_3_IV3,
Review Comment:
Could we just use MetadataVersion.MINIMUM_VERSION?
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -349,28 +344,27 @@ public void testCanUseUnsafeDowngradeIfMetadataChanged() {
public void testCanUseSafeDowngradeIfMetadataDidNotChange() {
FeatureControlManager manager = new FeatureControlManager.Builder().
setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV1.featureLevel())).
- setMetadataVersion(MetadataVersion.IBP_3_1_IV0).
- setMinimumBootstrapVersion(MetadataVersion.IBP_3_0_IV1).
+ MetadataVersion.MINIMUM_VERSION.featureLevel(), MetadataVersion.IBP_3_6_IV0.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_5_IV0).
build();
assertEquals(ControllerResult.of(Collections.emptyList(), ApiError.NONE),
manager.updateFeatures(
- singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_0_IV1.featureLevel()),
+ singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_4_IV0.featureLevel()),
singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.SAFE_DOWNGRADE),
true));
}
@Test
- public void testCannotDowngradeBefore3_3_IV0() {
+ public void testCannotDowngradeBeforeMinimumKraftVersion() {
FeatureControlManager manager = new FeatureControlManager.Builder().
setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
+ MetadataVersionTestUtils.IBP_3_0_IV1_FEATURE_LEVEL, MetadataVersion.MINIMUM_VERSION.featureLevel())).
Review Comment:
We probably should set the min/max to `MetadataVersion.MINIMUM_VERSION` and `MetadataVersion.latestTesting()`.
##########
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java:
##########
@@ -511,7 +511,7 @@ public void testLastAppliedOffset() throws Exception {
new MetadataProvenance(200, 100, 4000, true), asList(
singletonList(new ApiMessageAndVersion(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(IBP_3_3_IV1.featureLevel()), (short) 0)),
+ setFeatureLevel(IBP_3_3_IV3.featureLevel()), (short) 0)),
Review Comment:
Since this can be any MV, could we just use MetadataVersion.MINIMUM_VERSION?
##########
metadata/src/test/java/org/apache/kafka/image/FeaturesImageTest.java:
##########
@@ -163,7 +163,7 @@ public void testEmpty() {
assertFalse(new FeaturesImage(Collections.singletonMap("foo", (short) 1),
FeaturesImage.EMPTY.metadataVersion()).isEmpty());
assertFalse(new FeaturesImage(FeaturesImage.EMPTY.finalizedVersions(),
- MetadataVersion.IBP_3_3_IV0).isEmpty());
+ MetadataVersion.IBP_3_4_IV0).isEmpty());
Review Comment:
Since this can be any MV, could we just use MetadataVersion.MINIMUM_VERSION?
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -240,38 +241,29 @@ public void testReplayRecords() {
private static final FeatureControlManager.Builder TEST_MANAGER_BUILDER1 =
new FeatureControlManager.Builder().
setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_3_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV2);
+ MetadataVersion.MINIMUM_VERSION.featureLevel(), MetadataVersion.IBP_3_6_IV0.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_4_IV0);
@Test
public void testApplyMetadataVersionChangeRecord() {
FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
manager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel()));
- assertEquals(MetadataVersion.IBP_3_3_IV3, manager.metadataVersion());
- }
-
- @Test
- public void testCannotDowngradeToVersionBeforeMinimumSupportedKraftVersion() {
- FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
- assertEquals(ControllerResult.of(Collections.emptyList(), new ApiError(Errors.INVALID_UPDATE_VERSION,
- "Invalid update version 3 for feature metadata.version. Local controller 0 only " +
- "supports versions 4-7")),
- manager.updateFeatures(
- singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_2_IV0.featureLevel()),
- singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.UNSAFE_DOWNGRADE),
- true));
+ setFeatureLevel(MetadataVersion.IBP_3_4_IV0.featureLevel()));
+ assertEquals(MetadataVersion.IBP_3_4_IV0, manager.metadataVersion());
}
@Test
public void testCannotDowngradeToHigherVersion() {
- FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+ FeatureControlManager manager = new FeatureControlManager.Builder().
Review Comment:
Could we use TEST_MANAGER_BUILDER1? This will make this test the same as `testCannotUnsafeDowngradeToHigherVersion` except for the unsafe flag, which is a bit easier to understand.
##########
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java:
##########
@@ -597,7 +597,7 @@ private void loadTestSnapshot2(
new MetadataProvenance(offset, 100, 4000, true), asList(
singletonList(new ApiMessageAndVersion(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(IBP_3_3_IV2.featureLevel()), (short) 0)),
+ setFeatureLevel(IBP_3_4_IV0.featureLevel()), (short) 0)),
Review Comment:
Since this can be any MV larger than MetadataVersion.MINIMUM_VERSION, could we just use MetadataVersion.latestProduction?
##########
server-common/src/test/java/org/apache/kafka/server/common/MetadataVersionTest.java:
##########
@@ -29,38 +29,20 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
class MetadataVersionTest {
- @Test
- public void testKRaftFeatureLevelsBefore3_0_IV1() {
- for (int i = 0; i < MetadataVersion.IBP_3_0_IV1.ordinal(); i++) {
- assertEquals(-1, MetadataVersion.VERSIONS[i].featureLevel());
- }
- }
@Test
- public void testKRaftFeatureLevelsAtAndAfter3_0_IV1() {
- for (int i = MetadataVersion.IBP_3_0_IV1.ordinal(); i < MetadataVersion.VERSIONS.length; i++) {
- int expectedLevel = i - MetadataVersion.IBP_3_0_IV1.ordinal() + 1;
+ public void testFeatureLevels() {
+ for (int i = MINIMUM_VERSION.ordinal(); i < MetadataVersion.VERSIONS.length; i++) {
+ int expectedLevel = i + 7;
Review Comment:
Could we use MINIMUM_VERSION.featureLevel() instead of 7?
##########
metadata/src/test/java/org/apache/kafka/image/loader/MetadataLoaderTest.java:
##########
@@ -581,7 +581,7 @@ private void loadTestSnapshot(
new MetadataProvenance(offset, 100, 4000, true), asList(
singletonList(new ApiMessageAndVersion(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(IBP_3_3_IV1.featureLevel()), (short) 0)),
+ setFeatureLevel(IBP_3_3_IV3.featureLevel()), (short) 0)),
Review Comment:
Since this can be any MV, could we just use MetadataVersion.MINIMUM_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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1948187511
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsCommandTest.java:
##########
@@ -128,18 +126,6 @@ public void testReassignment() throws Exception {
executeAndVerifyReassignment();
}
- @ClusterTests({
- @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, metadataVersion = IBP_3_3_IV0)
- })
- public void testReassignmentWithAlterPartitionDisabled() throws Exception {
- // Test reassignment when the IBP is on an older version which does not use
- // the `AlterPartition` API. In this case, the controller will register individual
- // watches for each reassigning partition so that the reassignment can be
- // completed as soon as the ISR is expanded.
Review Comment:
Has the code for this fallback been removed? If not, do we have a JIRA ticket for it? If not, I'll file one.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950095803
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1652,9 +1652,7 @@ private void registerWriteNoOpRecord(long maxIdleIntervalNs) {
periodicControl.registerTask(new PeriodicTask("writeNoOpRecord",
() -> {
ArrayList<ApiMessageAndVersion> records = new ArrayList<>(1);
- if (featureControl.metadataVersion().isNoOpRecordSupported()) {
- records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
- }
+ records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
Review Comment:
> Do you know if this change is safe? My cursory look indicates it may be ok, but I would have to look at this more carefully.
Since other similar cases pass immutable lists to `ControllerResult`, it seems safe to do so here as well.
> I think we should also do this via a different PR (to avoid including a potential behavior change in the middle of this clean-up PR).
make sense to me
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1951108501
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -44,25 +44,6 @@
*/
Review Comment:
Done.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1950100652
##########
metadata/src/test/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectoryTest.java:
##########
@@ -74,39 +73,21 @@ public void testReadFromEmptyConfiguration() throws Exception {
try (BootstrapTestDirectory testDirectory = new BootstrapTestDirectory().createDirectory()) {
assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.latestProduction(),
"the default bootstrap"),
- new BootstrapDirectory(testDirectory.path(), Optional.empty()).read());
- }
- }
-
- @Test
- public void testReadFromConfigurationWithAncientVersion() throws Exception {
- try (BootstrapTestDirectory testDirectory = new BootstrapTestDirectory().createDirectory()) {
- assertEquals(BootstrapMetadata.fromVersion(MetadataVersion.MINIMUM_BOOTSTRAP_VERSION,
- "the minimum version bootstrap with metadata.version 3.3-IV0"),
- new BootstrapDirectory(testDirectory.path(), Optional.of("3.0")).read());
- }
- }
-
- @Test
- public void testReadFromConfiguration() throws Exception {
Review Comment:
`testReadFromConfiguration` is the same as `testReadFromEmptyConfiguration` once the `MetadataVersion` parameter is removed.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1960678323
##########
metadata/src/test/java/org/apache/kafka/image/writer/ImageWriterOptionsTest.java:
##########
@@ -48,40 +48,33 @@
public class ImageWriterOptionsTest {
@Test
public void testDefaultLossHandler() {
- ImageWriterOptions options = new ImageWriterOptions.Builder().build();
+ ImageWriterOptions options = new ImageWriterOptions.Builder(MetadataVersion.latestProduction()).build();
assertEquals("stuff", assertThrows(UnwritableMetadataException.class,
() -> options.handleLoss("stuff")).loss());
}
@Test
public void testSetMetadataVersion() {
Review Comment:
I agree that this test isn't useful since the special logic for the setter was removed - even without removing the builder. I removed it.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1960696997
##########
metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java:
##########
@@ -718,30 +710,14 @@ public void testRegistrationWithUnsupportedMetadataVersion() {
setFeatures(new BrokerRegistrationRequestData.FeatureCollection(
Collections.singleton(new BrokerRegistrationRequestData.Feature().
setName(MetadataVersion.FEATURE_NAME).
- setMinSupportedVersion(MetadataVersion.IBP_3_3_IV3.featureLevel()).
- setMaxSupportedVersion(MetadataVersion.IBP_3_3_IV3.featureLevel())).iterator())).
+ setMinSupportedVersion(MetadataVersion.IBP_3_4_IV0.featureLevel()).
+ setMaxSupportedVersion(MetadataVersion.IBP_3_4_IV0.featureLevel())).iterator())).
setIncarnationId(Uuid.fromString("0H4fUu1xQEKXFYwB1aBjhg")),
123L,
featureControl.finalizedFeatures(Long.MAX_VALUE),
false)).getMessage());
}
- @Test
- public void testRegisterControlWithOlderMetadataVersion() {
- FeatureControlManager featureControl = new FeatureControlManager.Builder().
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
Review Comment:
It appears that `testRegistrationWithUnsupportedMetadataVersion` is designed to test broker registration, while `testRegisterControlWithOlderMetadataVersion` is for controller 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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953369257
##########
metadata/src/main/java/org/apache/kafka/image/FeaturesDelta.java:
##########
@@ -58,7 +59,14 @@ public void finishSnapshot() {
public void replay(FeatureLevelRecord record) {
if (record.name().equals(MetadataVersion.FEATURE_NAME)) {
- metadataVersionChange = MetadataVersion.fromFeatureLevel(record.featureLevel());
+ // Support for the `metadata.version` feature flag was added in IBP_3_3_IV0, so it's possible (but unlikely) that we read
+ // records with a feature level that is no longer supported for clusters that used a pre-release version of 3.3.0.
+ // We automatically fallback to IBP_3_3_IV3 in that case. We use explicit versions instead of `MINIMUM_VERSION` because
+ // we want to force an explicit decision if we change `MetadataVersion.MINIMUM_VERSION` in the future.
+ if (record.featureLevel() >= MINIMUM_PERSISTED_FEATURE_LEVEL && record.featureLevel() <= MetadataVersion.IBP_3_3_IV3.featureLevel())
+ metadataVersionChange = MetadataVersion.IBP_3_3_IV3;
Review Comment:
Is it guaranteed that we won't have a record with the older metadata version in the log in that case?
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1952741967
##########
metadata/src/test/java/org/apache/kafka/image/FeaturesImageTest.java:
##########
@@ -163,7 +163,7 @@ public void testEmpty() {
assertFalse(new FeaturesImage(Collections.singletonMap("foo", (short) 1),
FeaturesImage.EMPTY.metadataVersion()).isEmpty());
assertFalse(new FeaturesImage(FeaturesImage.EMPTY.finalizedVersions(),
- MetadataVersion.IBP_3_3_IV0).isEmpty());
+ MetadataVersion.IBP_3_4_IV0).isEmpty());
Review Comment:
That causes the test to fail. This is related to your question of whether `FeatuesImage.isEmpty` should be changed.
--
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-18601: Assume a baseline of 3.3 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18845:
URL: https://github.com/apache/kafka/pull/18845#discussion_r1953225545
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1652,9 +1652,7 @@ private void registerWriteNoOpRecord(long maxIdleIntervalNs) {
periodicControl.registerTask(new PeriodicTask("writeNoOpRecord",
() -> {
ArrayList<ApiMessageAndVersion> records = new ArrayList<>(1);
- if (featureControl.metadataVersion().isNoOpRecordSupported()) {
- records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
- }
+ records.add(new ApiMessageAndVersion(new NoOpRecord(), (short) 0));
Review Comment:
OK, I'll include this change in my next push.
--
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