KAFKA-19964: API Versions Response returns minimum metadata version for finalized level when no quorum exists #21122

mannoopj · 2025-12-10T20:53:54Z

FeaturesPublisher's default finalizedFeatures assumes a metadata version of 7 when there is no finalized level for metadata.version. Instead of defaulting to 7, it should not report a value or report -1 indicating it is unknown.

See https://lists.apache.org/thread/ytmo7b9vmy46n6p47l1sxx6ftvdhht5b for the context as part of the KIP-1170 discussion.

Implemented by changing finalizedFeatures to Optional in FeaturesPublisher

kevin-wu24

Thanks for the changes @mannoopj. Left a review of src/main. Can we also add a FeaturesPublisher unit test to verify this new behavior?

kevin-wu24 · 2025-12-10T21:24:22Z

server/src/main/java/org/apache/kafka/server/SimpleApiVersionManager.java

    @Override
    public FinalizedFeatures features() {
-        return featuresProvider.get();
+        return featuresProvider.get().orElse(null);


We should avoid ever returning null, because this is not safe. Instead, we can change the ApiVersionManager interface's features() method to return an Optional<FinalizedFeatures>. The implementors can return the wrapped optional, and check things based on its presence.

For example, ControllerApis#handleDescribeCluster can now return a more useful error message if the optional is not present. We can also add a test for that in ControllerApisTest.

kevin-wu24 · 2025-12-10T21:59:31Z

core/src/main/scala/kafka/server/ControllerServer.scala

-        () => featuresPublisher.features().setFinalizedLevel(
-          KRaftVersion.FEATURE_NAME,
-          raftManager.client.kraftVersion().featureLevel())
+        () => featuresPublisher.features().map(f =>


There is a subtlety here with kraft.version. Previously, featuresPublisher.features was always "present", so we can always set the "finalized" kraft version known by the local KafkaRaftClient. However, now the featurePublisher.features() might be an empty optional, in which case the function in the .map call won't be applied until after featuresPublisher.features().isPresent().

This means if a controller node has kraft.version=1, but has not committed the bootstrap metadata version record (i.e. featuresPublisher.features().isEmpty()), the controller will not show that it is kraft.version=1 in the ApiVersionsResponse it sends, even though it previously would have.

I think the intention is, for better or for worse, that the "finalizedFeatures" section of the ApiVersionsResponse to show the local node's kraft.version level, even if it may be uncommitted. Is that correct @jsancio?

If so, one way to fix this is to add a NOT_FINALIZED enum value to MetadataVersion. This option makes the most sense to me, but it is weird that FinalizedFeatures has a NOT_FINALIZED value for metadata version...

Or we can make FinalizedFeatures#metadataVersion field an optional, and keep it so that FinalizedFeatures does not have to be an optional (although this seems confusing when the object is called "FinalizedFeatures").

Also it is weird we have FinalizedFeatures and FinalizedControllerFeatures records.

Looks like this behavior is not intentional: KAFKA-18865

kevin-wu24 · 2025-12-10T22:46:11Z

server/src/main/java/org/apache/kafka/server/SimpleApiVersionManager.java

                .setSupportedFeatures(brokerFeatures)
-                .setFinalizedFeatures(currentFeatures.finalizedFeatures())
-                .setFinalizedFeaturesEpoch(currentFeatures.finalizedFeaturesEpoch())
+                .setFinalizedFeatures(currentFeatures.map(FinalizedFeatures::finalizedFeatures).orElse(Map.of()))


Actually, if we pass the kraft.version information here as a Supplier, we can set the finalized features field with the kraft.version here whenever currentFeatures.isEmpty() (i.e. FeaturesPublisher#finalizedFeatures.isEmpty()).

val apiVersionManager = new SimpleApiVersionManager( ListenerType.CONTROLLER, config.unstableApiVersionsEnabled, () => featuresPublisher.features().setFinalizedLevel( KRaftVersion.FEATURE_NAME, raftManager.client.kraftVersion().featureLevel()) )

This code in ControllerServer.scala already adds the kraft.version. It might not appear in the output if kraft.version is 0 and hence gets filtered out.

the only other places SimpleApiVersionManager gets called are from tests and KRaftMetadataRequestBenchmark so changing the default to FinalizedFeatures.UNKNOWN_FINALIZED_FEATURES shouldn't change any behaviors here

I think if we want to expose the latest committed kraft version, we should do it in a separate PR.

This code in ControllerServer.scala already adds the kraft.version. It might not appear in the output if kraft.version is 0 and hence gets filtered out.

Yes but the current code will expose kraft.version=1 in the APIVersionsResponse even if it has not been committed yet. We can slightly improve on this by not adding the kraft.version if the metadata.version does not exist, but the feature's value in the API versions may still be uncommitted when doing upgrade.

This reverts commit 6d53ead.

kevin-wu24

Thanks for the changes @mannoopj . Left another review of src/main.

Can we add the suggestion from: #21122 (comment)? That way we can call setFinalizedFeatureLevel for KRaft in the APIVersionsResponse only when the FinalizedFeatures is is not "empty."

kevin-wu24 · 2025-12-11T17:51:09Z

metadata/src/main/java/org/apache/kafka/metadata/publisher/FeaturesPublisher.java

    private final Logger log;
    private final FaultHandler faultHandler;
-    private volatile FinalizedFeatures finalizedFeatures = FinalizedFeatures.fromKRaftVersion(MINIMUM_VERSION);
+    private volatile FinalizedFeatures finalizedFeatures = new FinalizedFeatures(


We should make new FinalizedFeatures(Optional.empty()...) a static constant in FinalizedFeatures. Maybe something like UNKNOWN_FINALIZED_FEATURES.

kevin-wu24 · 2025-12-11T19:13:26Z

core/src/main/scala/kafka/server/ControllerApis.scala

    // Nearly all RPCs should check MetadataVersion inside the QuorumController. However, this
    // RPC is consulting a cache which lives outside the QC. So we check MetadataVersion here.
-    if (!apiVersionManager.features.metadataVersion().isControllerRegistrationSupported) {
+    if (!apiVersionManager.features.metadataVersion().map(_.isControllerRegistrationSupported).orElse(false)) {


It would be better if we could return a more accurate error message in the case where metadataVersion().isEmpty(). Returning a message like "There is no finalized MetadataVersion, so direct-to-controller communication is not supported" is more accurate in this case.

kevin-wu24 · 2025-12-15T22:16:30Z

core/src/main/scala/kafka/server/ControllerApis.scala

-        "supported with the current MetadataVersion.")
+    if (!apiVersionManager.features.metadataVersion().map(_.isControllerRegistrationSupported).orElse(false)) {
+      throw new UnsupportedVersionException("There is no finalized MetadataVersion, so " +
+        "direct-to-controller communication is not supported.")


Please make a separate if/else if clause for if the metadata version is not finalized. We should not change this existing check.

kevin-wu24 · 2025-12-15T22:19:19Z

server/src/main/java/org/apache/kafka/server/SimpleApiVersionManager.java

                .setSupportedFeatures(brokerFeatures)
-                .setFinalizedFeatures(currentFeatures.finalizedFeatures())
-                .setFinalizedFeaturesEpoch(currentFeatures.finalizedFeaturesEpoch())
+                .setFinalizedFeatures(currentFeatures.map(FinalizedFeatures::finalizedFeatures).orElse(Map.of()))


I think if we want to expose the latest committed kraft version, we should do it in a separate PR.

Add optional FinalizedFeatures

6d53ead

github-actions bot added triage PRs from the community core Kafka Broker performance kraft small Small PRs labels Dec 10, 2025

kevin-wu24 reviewed Dec 10, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label Dec 11, 2025

mannoopj added 3 commits December 11, 2025 12:24

Revert "Add optional FinalizedFeatures"

3126d1c

This reverts commit 6d53ead.

Have MetadataVersion be optional rather than FinalizedFeatures

80de492

remove FinalizedFeatures.EMPTY

08fcd72

kevin-wu24 reviewed Dec 11, 2025

View reviewed changes

nit fixes

f072ffb

kevin-wu24 reviewed Dec 15, 2025

View reviewed changes

KAFKA-19964: API Versions Response returns minimum metadata version for finalized level when no quorum exists #21122

Are you sure you want to change the base?

KAFKA-19964: API Versions Response returns minimum metadata version for finalized level when no quorum exists #21122

Uh oh!

Conversation

mannoopj commented Dec 10, 2025

Uh oh!

kevin-wu24 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kevin-wu24 Dec 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kevin-wu24 Dec 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kevin-wu24 Dec 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kevin-wu24 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

kevin-wu24 Dec 10, 2025 •

edited

Loading

kevin-wu24 Dec 10, 2025 •

edited

Loading

kevin-wu24 Dec 10, 2025 •

edited

Loading