You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "cmccabe (via GitHub)" <gi...@apache.org> on 2025/04/23 22:39:47 UTC
[PR] KAFKA-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
cmccabe opened a new pull request, #19545:
URL: https://github.com/apache/kafka/pull/19545
Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers. When the server tries to read the bootstrap.metadata file, it will fail if the metadata.version is older than 3.3-IV3 (feature level 7). This causes problems when these clusters are upgraded.
This PR makes it possible to represent older MVs in BootstrapMetadata objects without causing an exception. An exception is thrown only if we attempt to access the BootstrapMetadata. This ensures that only the code path in which we start wtih an empty metadata log checks that the metadata version is 7 or newer.
--
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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2058834500
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -66,48 +66,48 @@ public static BootstrapMetadata fromVersions(
setFeatureLevel(level), (short) 0));
}
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
List<ApiMessageAndVersion> records = List.of(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
- MetadataVersion metadataVersion = null;
+ Optional<Short> metadataVersionLevel = Optional.empty();
for (ApiMessageAndVersion record : records) {
- Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
- if (version.isPresent()) {
- metadataVersion = version.get();
+ Optional<Short> level = recordToMetadataVersionLevel(record.message());
+ if (level.isPresent()) {
+ metadataVersionLevel = level;
}
}
- if (metadataVersion == null) {
+ if (!metadataVersionLevel.isPresent()) {
throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
" was found in the bootstrap metadata from " + source);
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersionLevel.get(), source);
}
- public static Optional<MetadataVersion> recordToMetadataVersion(ApiMessage record) {
+ public static Optional<Short> recordToMetadataVersionLevel(ApiMessage record) {
if (record instanceof FeatureLevelRecord featureLevel) {
if (featureLevel.name().equals(MetadataVersion.FEATURE_NAME)) {
- return Optional.of(MetadataVersion.fromFeatureLevel(featureLevel.featureLevel()));
+ return Optional.of(featureLevel.featureLevel());
Review Comment:
It's true that we might have to revisit this if we drop support for records previously used in bootstrapping.
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -1619,6 +1619,55 @@ class KRaftClusterTest {
}
}
+ /**
+ * Test that once a cluster is formatted, a bootstrap.metadata file that contains an unsupported
+ * MetadataVersion is not a problem. This is a regression test for KAFKA-19192.
+ */
+ @Test
+ def testOldBootstrapMetadataFile(): Unit = {
Review Comment:
yes, it does
--
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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2056983000
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -117,7 +117,7 @@ public List<ApiMessageAndVersion> records() {
}
public MetadataVersion metadataVersion() {
- return metadataVersion;
+ return MetadataVersion.fromFeatureLevel(metadataVersionLevel);
Review Comment:
this seems a bit dangerous - right now we call this when generating activation records and in a data loss scenario/new controller scenario - wouldn't we hit the bug again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2058871135
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -1619,6 +1619,55 @@ class KRaftClusterTest {
}
}
+ /**
+ * Test that once a cluster is formatted, a bootstrap.metadata file that contains an unsupported
+ * MetadataVersion is not a problem. This is a regression test for KAFKA-19192.
+ */
+ @Test
+ def testOldBootstrapMetadataFile(): Unit = {
+ val baseDirectory = TestUtils.tempDir().toPath()
+ val cluster = new KafkaClusterTestKit.Builder(
+ new TestKitNodes.Builder().
+ setNumBrokerNodes(1).
+ setNumControllerNodes(1).
+ setBaseDirectory(baseDirectory).
+ build()).
+ setDeleteOnClose(false).
+ build()
+ try {
+ cluster.format()
+ cluster.startup()
+ cluster.waitForReadyBrokers()
+ } finally {
+ cluster.close()
+ }
Review Comment:
Scala recently added the Using type. It has support for closing or releasing resources.
```java
Using.resource(new KafkaClusterTestKit.Builder(
new TestKitNodes.Builder().
setNumBrokerNodes(1).
setNumControllerNodes(1).
setBaseDirectory(baseDirectory).
build()).
setDeleteOnClose(false).
build()
) { cluster =>
cluster.format()
cluster.startup()
cluster.waitForReadyBrokers()
}
```
This comment also applies to the creation and usage of the `cluster2` object in this method.
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
Review Comment:
This comment applies to the PR description.
In the description you have "When the server tries to read the bootstrap.metadata file". Do you mean "the bootstrap.checkpoint" file?
--
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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2058913387
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -117,7 +117,7 @@ public List<ApiMessageAndVersion> records() {
}
public MetadataVersion metadataVersion() {
- return metadataVersion;
+ return MetadataVersion.fromFeatureLevel(metadataVersionLevel);
Review Comment:
got it - and there's no check that the bootstrap metadata file / bootstrap MV matches across all controllers? in this data loss scenario, it would work to re-format the directory with the current MV that the cluster is on before re-adding the voter?
--
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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio merged PR #19545:
URL: https://github.com/apache/kafka/pull/19545
--
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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2058561246
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -66,48 +66,48 @@ public static BootstrapMetadata fromVersions(
setFeatureLevel(level), (short) 0));
}
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
List<ApiMessageAndVersion> records = List.of(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
- MetadataVersion metadataVersion = null;
+ Optional<Short> metadataVersionLevel = Optional.empty();
for (ApiMessageAndVersion record : records) {
- Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
- if (version.isPresent()) {
- metadataVersion = version.get();
+ Optional<Short> level = recordToMetadataVersionLevel(record.message());
+ if (level.isPresent()) {
+ metadataVersionLevel = level;
}
}
- if (metadataVersion == null) {
+ if (!metadataVersionLevel.isPresent()) {
throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
" was found in the bootstrap metadata from " + source);
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersionLevel.get(), source);
}
- public static Optional<MetadataVersion> recordToMetadataVersion(ApiMessage record) {
+ public static Optional<Short> recordToMetadataVersionLevel(ApiMessage record) {
if (record instanceof FeatureLevelRecord featureLevel) {
if (featureLevel.name().equals(MetadataVersion.FEATURE_NAME)) {
- return Optional.of(MetadataVersion.fromFeatureLevel(featureLevel.featureLevel()));
+ return Optional.of(featureLevel.featureLevel());
Review Comment:
This works because, in the cluster metadata partition, Kafka 4.0 only deprecated and removed MV less than 7. If in the future Kafka drops supports for other metadata records or fields then we will need to revisit this issue, right?
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -117,7 +117,7 @@ public List<ApiMessageAndVersion> records() {
}
public MetadataVersion metadataVersion() {
- return metadataVersion;
+ return MetadataVersion.fromFeatureLevel(metadataVersionLevel);
Review Comment:
Right. Kafka assumes that if there is data loss it is because the entire disk failed. Kafka doesn't handle error in the filesystem were some data was lost after an fsync.
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -66,48 +66,48 @@ public static BootstrapMetadata fromVersions(
setFeatureLevel(level), (short) 0));
}
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
List<ApiMessageAndVersion> records = List.of(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
- MetadataVersion metadataVersion = null;
+ Optional<Short> metadataVersionLevel = Optional.empty();
for (ApiMessageAndVersion record : records) {
- Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
- if (version.isPresent()) {
- metadataVersion = version.get();
+ Optional<Short> level = recordToMetadataVersionLevel(record.message());
+ if (level.isPresent()) {
+ metadataVersionLevel = level;
}
}
- if (metadataVersion == null) {
+ if (!metadataVersionLevel.isPresent()) {
Review Comment:
Yes. Newer version of Java added `isEmpty`. We can use that in Kafka since we dropped support for Java 8.
--
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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2057607502
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -117,7 +117,7 @@ public List<ApiMessageAndVersion> records() {
}
public MetadataVersion metadataVersion() {
- return metadataVersion;
+ return MetadataVersion.fromFeatureLevel(metadataVersionLevel);
Review Comment:
In a data loss scenario with `kraft.version=1` we would need to re-format the directory and re-add the voter, so there would be no more `bootstrap.metadata` file.
(If you delete some files in the raft log directory but not others, bad things may happen, of course, but that is no different than today; it breaks the Raft invariants.)
--
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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "ahuang98 (via GitHub)" <gi...@apache.org>.
ahuang98 commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2058913387
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -117,7 +117,7 @@ public List<ApiMessageAndVersion> records() {
}
public MetadataVersion metadataVersion() {
- return metadataVersion;
+ return MetadataVersion.fromFeatureLevel(metadataVersionLevel);
Review Comment:
got it - and controllers don't check that the bootstrap metadata file / bootstrap MV matches? in this data loss scenario, it would work to re-format the directory with the current MV that the cluster is on before re-adding the voter?
--
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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2058643463
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -1619,6 +1619,55 @@ class KRaftClusterTest {
}
}
+ /**
+ * Test that once a cluster is formatted, a bootstrap.metadata file that contains an unsupported
+ * MetadataVersion is not a problem. This is a regression test for KAFKA-19192.
+ */
+ @Test
+ def testOldBootstrapMetadataFile(): Unit = {
Review Comment:
Thanks for the fix. I assume this fails without 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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2059108995
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -66,48 +66,48 @@ public static BootstrapMetadata fromVersions(
setFeatureLevel(level), (short) 0));
}
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
List<ApiMessageAndVersion> records = List.of(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
- MetadataVersion metadataVersion = null;
+ Optional<Short> metadataVersionLevel = Optional.empty();
for (ApiMessageAndVersion record : records) {
- Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
- if (version.isPresent()) {
- metadataVersion = version.get();
+ Optional<Short> level = recordToMetadataVersionLevel(record.message());
+ if (level.isPresent()) {
+ metadataVersionLevel = level;
}
}
- if (metadataVersion == null) {
+ if (!metadataVersionLevel.isPresent()) {
throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
" was found in the bootstrap metadata from " + source);
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersionLevel.get(), source);
}
- public static Optional<MetadataVersion> recordToMetadataVersion(ApiMessage record) {
+ public static Optional<Short> recordToMetadataVersionLevel(ApiMessage record) {
if (record instanceof FeatureLevelRecord featureLevel) {
if (featureLevel.name().equals(MetadataVersion.FEATURE_NAME)) {
- return Optional.of(MetadataVersion.fromFeatureLevel(featureLevel.featureLevel()));
+ return Optional.of(featureLevel.featureLevel());
Review Comment:
Other records are scram records take a look at KIP-900: https://cwiki.apache.org/confluence/x/f49bDg
A new cluster needs to copy all of the records in `bootstrap.checkpoint` to the cluster metadata partition. After a cluster has done that for the first time it doesn't need to read or process the bootstrap checkpoint again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2057712261
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -66,48 +66,48 @@ public static BootstrapMetadata fromVersions(
setFeatureLevel(level), (short) 0));
}
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
List<ApiMessageAndVersion> records = List.of(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
- MetadataVersion metadataVersion = null;
+ Optional<Short> metadataVersionLevel = Optional.empty();
for (ApiMessageAndVersion record : records) {
- Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
- if (version.isPresent()) {
- metadataVersion = version.get();
+ Optional<Short> level = recordToMetadataVersionLevel(record.message());
+ if (level.isPresent()) {
+ metadataVersionLevel = level;
}
}
- if (metadataVersion == null) {
+ if (!metadataVersionLevel.isPresent()) {
Review Comment:
```suggestion
if (metadataVersionLevel.isEmpty()) {
```
--
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-19192: Old bootstrap.metadata files cause problems with post-KAFKA-18601 servers [kafka]
Posted by "lzyLuke (via GitHub)" <gi...@apache.org>.
lzyLuke commented on code in PR #19545:
URL: https://github.com/apache/kafka/pull/19545#discussion_r2058890828
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -66,48 +66,48 @@ public static BootstrapMetadata fromVersions(
setFeatureLevel(level), (short) 0));
}
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
List<ApiMessageAndVersion> records = List.of(
new ApiMessageAndVersion(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersion.featureLevel(), source);
}
public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
- MetadataVersion metadataVersion = null;
+ Optional<Short> metadataVersionLevel = Optional.empty();
for (ApiMessageAndVersion record : records) {
- Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
- if (version.isPresent()) {
- metadataVersion = version.get();
+ Optional<Short> level = recordToMetadataVersionLevel(record.message());
+ if (level.isPresent()) {
+ metadataVersionLevel = level;
}
}
- if (metadataVersion == null) {
+ if (!metadataVersionLevel.isPresent()) {
throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
" was found in the bootstrap metadata from " + source);
}
- return new BootstrapMetadata(records, metadataVersion, source);
+ return new BootstrapMetadata(records, metadataVersionLevel.get(), source);
}
- public static Optional<MetadataVersion> recordToMetadataVersion(ApiMessage record) {
+ public static Optional<Short> recordToMetadataVersionLevel(ApiMessage record) {
if (record instanceof FeatureLevelRecord featureLevel) {
if (featureLevel.name().equals(MetadataVersion.FEATURE_NAME)) {
- return Optional.of(MetadataVersion.fromFeatureLevel(featureLevel.featureLevel()));
+ return Optional.of(featureLevel.featureLevel());
Review Comment:
qq: what are other typical records besides those feature level records in bootstrap.checkpoint file, why we cannot just skip doing metadata version check for the checkpoint file? So we do not worry about the support in 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