KAFKA-19851; Delete dynamic configs that were removed by Kafka

[0xffff-zhiyan]

When upgrading from Kafka 3.x to 4.0, the metadata log may contain
dynamic configurations that were removed in 4.0 (e.g.,
message.format.version per KIP-724). These removed configs cause
InvalidConfigurationException when users attempt to modify any
configuration, because validation checks all existing configs including
the removed ones.

Adds filtering to prevent unsupported or invalid configurations from
being applied during metadata replay. The filtering is implemented using
a SupportedConfigChecker interface that is injected via dependency
injection through Builder patterns. When a ConfigRecord is replayed, the
checker validates whether the configuration name is supported for the
given resource type. Unsupported configurations are silently ignored
during replay, ensuring that only valid configurations enter the
in-memory state.

The SupportedConfigChecker interface provides a default TRUE
implementation that accepts all configurations. The actual filtering
logic is implemented by DefaultSupportedConfigChecker, which maintains a
whitelist of valid configuration names per resource type (TOPIC,
CLIENT_METRICS, GROUP) based on the actual config definitions. The
filtering occurs in both ConfigurationDelta#replay and
ConfigurationControlManager#replay methods.

Added unit tests to ensure:

• Removed configurations are filtered during the replay operations
• Only supported configurations appear in the resulting metadata images
• The filtering works correctly for all resource types (TOPIC, BROKER,
  CLIENT_METRICS, GROUP)
• DefaultSupportedConfigChecker correctly identifies supported vs
  unsupported configurations for each resource type

Reviewers: José Armando García Sancio jsancio@apache.org, Jun Rao
junrao@apache.org, Alyssa Huang ahuang@confluent.io, Kevin Wu
kevin.wu2412@gmail.com, Andrew Grant agrant@confluent.io

[ahuang98]

partial review, thanks for working on this Tony!

I was wondering if we considered changing the behavior of validate(ConfigResource resource, Map<String, String> newConfigs, Map<String, String> existingConfigs); to log vs throw when there is an illegal existing config?

[ahuang98]

can we have this method handle UNKNOWN type to simplify logic elsewhere?

[ahuang98]

what about something like

return Optional.ofNullable(configSchemaSupplier)
    .map(Supplier::get)
    .map(schema -> schema.validConfigNames(resourceType))
    .orElse(Set.of());

[kevin-wu24]

Thanks for taking this on @0xffff-zhiyan. I just have some high level comments for your and @ahuang98's consideration. After doing some thinking on this, the issue is more complex than it appears if we want to actually remove these records from the metadata partition.

I was wondering if we considered changing the behavior of validate(ConfigResource resource, Map<String, String> newConfigs, Map<String, String> existingConfigs); to log vs throw when there is an illegal existing config?

I think the main question we want to answer is:

1. Do we want kafka to even remove these now-invalid configs from the cluster metadata partition as part of this PR?

What happens if the user wants to downgrade to a lower version? If we "clean up" config records, downgrading the software now results in lost metadata. In my opinion, the lossy downgrade case is enough to convince me kafka should not proactively clean up these configs.

Ultimately, what we want is the active controller to gracefully handle an ALTER_CONFIG after upgrading to a software version which may make some of its existing config state unknown/invalid to the new software version.

The most simple approach is to not validate dynamic configs that are not known to kafka. This matches what we do for static configs, as you can add unknown configs to the .properties file and it will not impact kafka. However, because we persist this config state to disk via the metadata partition, it is a problem to allow arbitrary config updates.

If the controller is to keep returning an error like it does today in this state, it should return an error that lists all the now-invalid configs so it is straightforward for the user to clean them up. This error should also let the user know these configs are invalid because they are unrecognized by the current kafka software version. This means the user becomes aware that a downgrade would be lossy if they delete these configs.

[kevin-wu24]

We shouldn't need to change this file at all. Remember that each ConfigurationDelta object will contain the current configuration image for a given ConfigResource, as well as its deltas. When we call ConfigurationDelta#apply, that is the only place we need to make changes.

[kevin-wu24]

Please remove the changes to this file.

[ahuang98]

I think I agree that it might have less intended side effects to not delete the unknown configs. About the downgrade scenario, I would assume most cases where we're introducing new topic config, it might involve a new MV, in which case we don't guarantee lossless downgrades (this is besides the point that MV downgrade isn't supported yet). We don't technically gate the actual topic config on MV or major versions, so it is quite possible we lose information unexpectedly on a downgrade.

The most simple approach is to not validate dynamic configs that are not known to kafka.

Seems similar to my question on if we can just avoid validating the existing configs and prevent new invalid configs from being added. I don't necessarily agree with allowing a user add a bad config - this could become a vulnerability if we don't have a cap on number of configs

[0xffff-zhiyan]

avoid validating the existing configs and prevent new invalid configs from being added

I agree that is a better way.

If the controller is to keep returning an error like it does today in this state, it should return an error that lists all the now-invalid configs so it is straightforward for the user to clean them up.

that way doesn't fix our current issue. the problem is that whenever users add or modify configurations, we throw an exception if there are any invalid configs. Users have to manually remove them all, which is tedious and exactly what we want to improve. Simply informing them about the invalid configs doesn’t really simplify the process, because they still need to clean them up one by one. And they will lose those configs permanently at last.

So based on the discussion above, we'd better stop validating existing configs and preventing users from adding invalid configs.

My only concern is: Is it really the right approach to let Kafka tolerate configurations that should no longer exist in the metadata? If we ever introduce new logic that handles existing configs in the future, we might have to keep adding code paths that explicitly ignore these existing but invalid configs. That seems like it could gradually accumulate technical debt. If we want the metadata to be clean without losing those configs permanently, is it possible we introduce a new config called REMOVED_CONFIG and move all those configs to there?
@ahuang98 @kevin-wu24

[kevin-wu24]

Thanks for the discussion @ahuang98 @0xffff-zhiyan:

I don't necessarily agree with allowing a user add a bad config - this could become a vulnerability if we don't have a cap on number of configs

I think I misunderstood your original comment. I agree that if we ignore the existing config metadata state and only validate what is contained in the ALTER_CONFIG request, a given version of kafka's dynamic config will be valid. When going between major versions, the removal from source code of config will not invalidate the existing dynamic config state on the new version of kafka, and allows ALTER_CONFIG to complete. This matches how the static .properties config is validated by kafka.

My only concern is: Is it really the right approach to let Kafka tolerate configurations that should no longer exist in the metadata? If we ever introduce new logic that handles existing configs in the future, we might have to keep adding code paths that explicitly ignore these existing but invalid configs

Dynamic configs that are not known by kafka, just like static configs, shouldn't invalidate the entire config. In this case, they are because ALTER_CONFIG will fail. The argument here is that we should not have been validating the existing dynamic config in the first place, since what is a "valid" (dynamic OR static) configuration depends only on the software version of kafka currently running. If I change software versions, fields in my static .properties file can go from valid -> unknown by kafka, and loading in those unknown configs into KafkaConfig does not throw an exception because they are ignored. We should apply this semantic to the dynamic configuration too.

[kevin-wu24]

Thanks for the changes @0xffff-zhiyan. Left a review of src/main.

[kevin-wu24]

Why is DynamicBrokerConfig#AllDynamicConfigs not used as part of the whitelist?

[kevin-wu24]

Please remove the changes to this file. Look at ControllerConfigurationValidator to see how dynamic config changes are validated.

[kevin-wu24]

Please remove the changes to this file.

[kevin-wu24]

We do not need this change. When upgrading the software version, the controller that becomes active must load its most recent snapshot from disk, and that will clean up the MetadataImage.

[kevin-wu24]

Look at ConfigCommandOptions#addConfig for the whitelist we should be using.

[0xffff-zhiyan]

KafkaConfigSchema is initialized in KafkaRaftServer and depends on config definitions from the core module (e.g., KafkaConfig.configDef, LogConfig.configDefCopy), we cannot build the whitelist in the metadata module due to module dependency constraints.

I found a solution to initialize the whitelist in ConfigurationDelta during KafkaRaftServer startup by calling ConfigurationDelta.initializeValidConfigs() with the set of valid dynamic config names collected from the core module. This whitelist is stored in a static AtomicReference<Set<String>> field, ensuring one-time initialization. The apply() method then uses this whitelist to filter out invalid/deprecated configurations.

@kevin-wu24 @jsancio @ahuang98

[ahuang98]

do we even need to filter what is being written to the metadata image? based on what we had discussed with allowing existing unknown configs to exist and only preventing new ones from being created, I would have thought there's no reason to change ConfigurationsDelta at all

[0xffff-zhiyan]

do we even need to filter what is being written to the metadata image? based on what we had discussed with allowing existing unknown configs to exist and only preventing new ones from being created, I would have thought there's no reason to change ConfigurationsDelta at all

Hmm... am I misunderstanding something? Didn't we agree to remove the existing invalid configs finally?

[ahuang98]

@0xffff-zhiyan thanks for clarifying, from reading Kevin's last comments I had assumed we were only going to validate new configs as they came in (and not sanitize existing configs)

[0xffff-zhiyan]

@ahuang98 Thanks!
I looked into validateAlterConfig, and the flow works like this: when we alter configs, we first read all existing configs from the snapshot into a map (existingConfigsSnapshot). Then we apply the config-altered records to update the configs in that map, and finally we validate the entire map. Because of this, it’s hard to cleanly separate validation of existing configs from validation of new configs.
So I added a filter when reading existing configs from the snapshot to ensure that all configs in the map are valid. I think this should prevent validation failures caused by existing invalid configs.

[ahuang98]

MetadataLoader is built in SharedServer, I'm wondering if it will be better practice to generate the set of dynamicConfigNames in sharedServer and then pass it all the way through form MetadataLoader to MetadataDelta to ConfigurationsDelta etc.

[ahuang98]

similarly here, wondering if it would be better to grab the set of valid dynamic configs from ControllerServer which already holds a reference to sharedServer. this could be passed all the way through to configurationControlManager. as an example, you could take a look at how sharedServer.metadataEncryptorFactory gets propagated to configurationControl in QuorumController.

[ahuang98]

guessing this is an atomicReference because we're worried about calling initializeValidConfigs more than once (presumably on accident). this could be avoided if we instantiate this value as part of the ConfigurationDelta constructor

[ahuang98]

Hey Tony, had a chance to discuss this a bit with @jsancio to improve my own understanding of how we might resolve some dependencies.

What do you think of the following?

We could create a validator that can check if a config name is valid for a given resource type (broker, topic, user, etc.) and then pass it through from SharedServer all the way down to where it's needed (ConfigurationDelta and ConfigurationControlManager).

We should probably define a functional interface for this validator which should live in the metadata module based on the fact the classes which will rely on it live in metadata. Since the actual knowledge of which configs are valid (e.g. DynamicConfig.Broker) is defined in the core module, the implementation of this interface can live in core.

[ahuang98]

there might be a better place to do this validation (you might want to take a look at FeaturesDelta and see how we handle filtering there for an old bug)

another good point from @jsancio is that all XXXDelta's apply methods are doing are applying deltas ontop of a base image to produce a new image. that new image then becomes the base image for the next delta to be applied ontop of. given that we always start with an empty base image, it follows that we only need to validate the deltas (referenced to as changes in this code) to ensure every image that is produced as a result is valid

[0xffff-zhiyan]

I’m a bit confused about how we construct the image from the snapshot. Do we keep applying deltas incrementally, or do we construct the image directly from the snapshot without going through deltas? If it’s the latter, then validating only the deltas wouldn’t be sufficient to remove invalid configs.

[kevin-wu24]

You can look at MetadataLoader#handleLoadSnapshot for how the metadata publishing pipeline (which generates the deltas we are working with here) deals with a snapshot received from the KRaft layer. The initial "image" is always empty, so the entire first snapshot loaded from KRaft for a given JVM lifetime is a delta. Since the SnapshotGenerator is a metadata publisher, the next snapshot that is written to disk after the "cleanup" code executes should no longer have dynamic configs unknown to kafka.

However, in order for the active controller to successfully handle an alter config request after updating its software, but before it loads a "cleaned" snapshot, I think we still need to clean up the ConfigurationControlManager's state, because its configData is the "dynamic config state of truth" for handling RPCs, not whatever exists in the MetadataLoader. Controller state for RPC handling uses QuoumController#QuorumMetaLogListener to load snapshots from KRaft, which is separate code from MetadataLoader.

I think I made an observation before that it seemed incorrect to do this kind of metadata state modification without explicitly committing records to KRaft. The MetadataLoader and QuoumController#QuorumMetaLogListener are both raft listeners, with each being responsible for different things (the former is responsible for generating the next snapshot on-disk, amongst other things, and the latter is indirectly responsible, via the ConfigurationControlManager, for handling alter config requests). Should we instead commit an explicit "delete these X unknown dynamic config records" to KRaft alongside the user's requested alter config upon receiving an alter config request that would otherwise be invalid because of unknown configs? This would avoid code duplication in the listeners.

@ahuang98 @0xffff-zhiyan @jsancio Let me know what you think of this approach, or if I am misunderstanding something.

[0xffff-zhiyan]

Should we instead commit an explicit "delete these X unknown dynamic config records" to KRaft alongside the user's requested alter config upon receiving an alter config request that would otherwise be invalid because of unknown configs?

So does that mean we only clean up invalid configs when handling alter config records?

I think maybe it’s acceptable to do cleanup in both listeners, even if that results in some duplicate logic. For the alter config failure scenario, in my current PR before validating existing configs, I simply filter out invalid ones so that validation can proceed. This is only to prevent validation failures, not to clean up the underlying config state.

[0xffff-zhiyan]

Hey Tony, had a chance to discuss this a bit with @jsancio to improve my own understanding of how we might resolve some dependencies.

What do you think of the following?

We could create a validator that can check if a config name is valid for a given resource type (broker, topic, user, etc.) and then pass it through from SharedServer all the way down to where it's needed (ConfigurationDelta and ConfigurationControlManager).

We should probably define a functional interface for this validator which should live in the metadata module based on the fact the classes which will rely on it live in metadata. Since the actual knowledge of which configs are valid (e.g. DynamicConfig.Broker) is defined in the core module, the implementation of this interface can live in core.

Thanks for the advice! This approach looks pretty good. I'll update my current implementation

[andrewgrantcflt]

Thanks for working on this @0xffff-zhiyan. The PR titles makes it seem like we are hard deleting the config. Actually, is it more the case the deprecated configs will stay in the log but we will just ignore them when replaying the log. Is that right? If so, maybe a PR title like "Ignore deprecated configs" would be clearer?

[0xffff-zhiyan]

Thanks for working on this @0xffff-zhiyan. The PR titles makes it seem like we are hard deleting the config. Actually, is it more the case the deprecated configs will stay in the log but we will just ignore them when replaying the log. Is that right? If so, maybe a PR title like "Ignore deprecated configs" would be clearer?

Yes, we just ignore them when replaying the log so they will be deleted from metadata image and the snapshot file built from that image. "Ignore deprecated configs" is more accurate! @andrewgrantcflt

[0xffff-zhiyan]

I have updated the PR @ahuang98
The ConfigValidator is instantiated in SharedServer and propagated through two distinct paths:

1. Used for filtering invalid configurations when loading metadata from the log or snapshots:
   SharedServer creates ConfigValidatorImpl instance
   SharedServer → MetadataLoader.Builder → MetadataLoader → MetadataDelta (when initializing new publishers) → ConfigurationsDelta → ConfigurationDelta
2. Used for filtering invalid configurations when validating configuration changes:
   SharedServer → ControllerServer → QuorumController → ConfigurationControlManager

I have a small concern about it. The metadata module already has an interface ConfigurationValidator. Now we define a similar one and implement it in core module. Will it cause some confusion?

And we can not use ConfigurationValidator because the checkstyle rules in import-control-metadata.xml disallow org.apache.kafka.image from importing org.apache.kafka.controller. Since ConfigurationValidator is in org.apache.kafka.controller, it cannot be used in the image subpackage.

[ahuang98]

@andrewgrantcflt @0xffff-zhiyan

The PR titles makes it seem like we are hard deleting the config. Actually, is it more the case the deprecated configs will stay in the log but we will just ignore them when replaying the log. Is that right? If so, maybe a PR title like "Ignore deprecated configs" would be clearer?

I think the important distinction is that the deprecated configs will be deleted as a result of ignoring them when replaying the log. At some point we'll delete the last snapshot that contained those deprecated configs.

@0xffff-zhiyan

I have a small concern about it. The core module already has an interface ConfigurationValidator. Now we define a similar one in metadata module and implement it in core module. Will it cause some confusion?

That's a good observation, would it maybe make sense to include your logic in ConfigurationValidator and/or ControllerConfigurationValidator?

[ahuang98]

so we're ignoring any invalid existing config by removing it from the existingConfigsMap prior to processing it in validator.validate(configResource, allConfigs, existingConfigsMap);

I'm thinking it might work out to just incorporate this logic directly into ControllerConfigurationValidator#isValidConfig

[0xffff-zhiyan]

Yes we can. But ControllerConfigurationValidator is initialized in ControllerServer.

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

Line 248 in e376801

setConfigurationValidator(new ControllerConfigurationValidator(sharedServer.brokerConfig)).

How can we pass it through from SharedServer all the way down to ConfigurationDelta?

[ahuang98]

now that isValidConfig() and validate() are methods in the same implementation, you can now have validate() check for isValidConfig() before verifying existingConfigs

[ahuang98]

or, there might be an argument for why we should just avoid throwing errors on existingConfigs when validating alter config requests to begin with. (ControllerConfigurationValidator#validate is only called when validating alter config requests - we can simply skip checking existingConfigs in that method altogether if folks agree on this)

[0xffff-zhiyan]

I think we still need to check existingConfigs because it's used for state transition validation, not just validity checks. For example, LogConfig.validate() needs the previous state to validate transitions like disabling remote storage (checking if remote.log.delete.on.disable=true is set).

[0xffff-zhiyan]

now that isValidConfig() and validate() are methods in the same implementation, you can now have validate() check for isValidConfig() before verifying existingConfigs

We filter existingConfigs in validateAlterConfig() rather than inside validate() because existingConfigsSnapshot is a reference to the map in configData. Removing invalid configs from it directly cleans up configData

[ahuang98]

oh I see the issue... existing configs are validated not because ControllerConfigurationValidator#validate enforces the same validation logic over oldConfigs as newConfigs but because ConfigurationControlManager passes the existingConfigs into the newConfigs map as well.

[jsancio]

@0xffff-zhiyan , you dont need this change now that you are skipping removed configs during the replay phase.

[junrao]

@0xffff-zhiyan : Thanks for the updated PR. One more comment.

[junrao]

Did existing tests catch this issue? If not, could we add another test to cover this?

[junrao]

Should this be image?

[0xffff-zhiyan]

fixed

[junrao]

@0xffff-zhiyan : Thanks for the updated PR. One more comment. Also, the test failures seem related to the PR?

[junrao]

Could we just use DefaultSupportedConfigChecker? Ditto below.

[0xffff-zhiyan]

DefaultSupportedConfigChecker lives in the server module, while MetadataLoaderTest is in the metadata module. Using DefaultSupportedConfigChecker in metadata tests would create a circular dependency so we can't use it

Also, I've updated DefaultSupportedConfigChecker to fix some other test failures caused by the BROKER whitelist. This is because valid BROKER configs include listener-specific prefixed configs and plugin-defined configs like listener.name.<name>.ssl.keystore.location and custom authorizer configs or quota callback configs that are dynamically named and cannot be pre-enumerated in DynamicConfig.Broker.names().

I changed BROKER configs to use a sentinel ALLOW_ALL set (whose contains() always returns true). And since no dynamic BROKER configs were deprecated in the 3.x → 4.x upgrade, I think there is nothing to filter for BROKER. WDYT?

[junrao]

Hmm, basically we are saying if a config can't be found, just ignore it during reply. Do you know where we hit the exception mentioned in the jira? Would it be simpler to just catch such exception during replay and ignore it?
Caused by: org.apache.kafka.common.errors.InvalidConfigurationException: Unknown topic config name: message.format.version

[0xffff-zhiyan]

The exception is thrown in validateAlterConfig() during ALTER, not in replay(). Specifically, when a user alters any topic config, the controller loads all existing configs from configData (including the deprecated message.format.version), merges them with the new config, and validates the combined set , which throws InvalidConfigurationException.

We shouldn't just catch or ignore the exception because it only papers over the symptom. The deprecated config would still exist in configData, be included in metadata snapshots, and be propagated to brokers.

The goal of this PR is to fix this at the root: we filter out deprecated configs at replay time so they never enter the in-memory state(configData). And the newest snapshot won't have it. We think this is a graceful way to delete them

[junrao]

Ok. My main concern with the current approach is with the handling of the BROKER configs. If we deprecate some broker configs in the future, they will still pass through the replay layer.

[0xffff-zhiyan]

Correct. Case (1) is expected behavior: AlterConfig for a removed config should fail. This PR only addresses case (2): deprecated configs that already exist in the metadata log from 3.x. They enter configData during replay and then block any subsequent valid config update

[junrao]

They enter configData during replay and then block any subsequent valid config update

Hmm, earlier you said "There is no validation before or during replay
And neither will take effect, they will be silently ignored when we try to apply the configs". Which one is correct?

[0xffff-zhiyan]

Sorry for the confusion. Let me clarify it again. Actually they're all correct and not contradictory
"Block subsequent valid config updates" refers to the user initiated ALTER path which is different from replay path. It will be blocked when invalid config entered in-memory state, right?

"Silently ignored when applied" refers to the broker's passive application of metadata changes via DynamicConfigPublisher → LogManager.updateTopicConfig() → LogConfig.fromProps() (this is the path how config takes effect, not ALTER path, not replay path)
https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/log/LogManager.scala#L972
Internally fromProps() calls ConfigDef.parse(), which only iterates known config keys and silently skips unknowns. So the deprecated config has no runtime effect.
@junrao

[jsancio]

The core issue is that during incremental alter config of the topic resource, the controller validates the materialized final config for the resource.

For example:

1. Assume that config X was removed from the list of supported configs for the TOPIC resource.
2. Assume that config X already exist for topic T. This was added to the topic before AK 4.0.

If the user tries to update config Y (different from X) for topic T using incremental alter config, it will fail because the controller tried to validate configs Y and X.

The workaround for this issue is that the user needs to delete all removed configs from topic T. They need to do this before they can update valid configs.

Given this issue and the workaround, why not have Kafka do it automatically? Kafka should automatically delete any and all configurations that were removed. Having said that, this can only be done for resources that have strict validation like TOPIC. Configs should not be deleted from resources that have more relax validations like BROKER.

[junrao]

Thanks for the explanation, Jose. I understand this now.

[jsancio]

Did you consider fixing this by fixing the type used? E.g.

    private final Map<ConfigResource.Type, Predicate<String>> validConfigsByType = Map.of(
        ConfigResource.Type.BROKER, ignore -> true,
        ConfigResource.Type.TOPIC, new SetContainsPredicate(new HashSet<>(LogConfig.configNames()))
        ...
    );

    ...

    final static class SetContainsPredicate implements Predicate<String> {
        private final Set<String> keys;

        ...

        @Override
        public boolean test(String key) { return keys.contains(key) }
    }

[0xffff-zhiyan]

Thanks for the suggestion. That's a better way! Updated

[jsancio]

Can we add tests for this functionality? We didn't add any new tests for this type.

[0xffff-zhiyan]

added

[junrao]

@0xffff-zhiyan : Thanks for the updated PR. Just a minor comment.

[junrao]

Should we log the skipped deprecated config here too?

[0xffff-zhiyan]

@jsancio pointed out that in kafka we should always use LogContext instead of creating loggers directly, but injecting LogContext all the way down to ConfigurationDelta would be quite invasive. So I removed the log statement here.

[junrao]

Thanks for the explanation, Jose. I understand this now.

[junrao]

@0xffff-zhiyan : Thanks for the explanation. LGTM. I will let Jose take another look.

[jsancio]

LGTM