Skip to content

fix(controller): skip AlterConfigPolicy validation for newly created resources #506

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
gpkc wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

fix(controller): skip AlterConfigPolicy validation for newly created resources #506

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,7 +369,14 @@ private ApiError validateAlterConfig(
if (!newlyCreatedResource) {
existenceChecker.accept(configResource);
}
alterConfigPolicy.ifPresent(policy -> policy.validate(new RequestMetadata(configResource, alteredConfigsForAlterConfigPolicyCheck)));
// When a resource is being created (e.g. topic creation), we may generate ConfigRecords
// before the resource is fully visible to other controller components. In that flow, we
// intentionally do not apply AlterConfigPolicy validation to the CreateTopics payload
// (ZooKeeper mode historically only applied CreateTopicPolicy during topic creation).
if (!newlyCreatedResource) {
alterConfigPolicy.ifPresent(policy ->
policy.validate(new RequestMetadata(configResource, alteredConfigsForAlterConfigPolicyCheck)));
}
} catch (ConfigException e) {
return new ApiError(INVALID_CONFIG, e.getMessage());
} catch (Throwable e) {
Expand Down
41 changes: 38 additions & 3 deletions metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -363,7 +363,42 @@ public void testIncrementalAlterConfigsWithPolicy() {
entry("quux", entry(SET, "456")),
entry("broker.config.to.remove", entry(DELETE, null))
))),
true));
false));
}

private static class AlwaysFailPolicy implements AlterConfigPolicy {
@Override
public void validate(RequestMetadata actual) throws PolicyViolationException {
throw new PolicyViolationException("Always fail");
}

@Override
public void close() {
// empty
}

@Override
public void configure(Map<String, ?> configs) {
// empty
}
}

@Test
public void testIncrementalAlterConfigSkipsPolicyForNewlyCreatedResource() {
ConfigurationControlManager manager = new ConfigurationControlManager.Builder().
setFeatureControl(createFeatureControlManager()).
setKafkaConfigSchema(SCHEMA).
setAlterConfigPolicy(Optional.of(new AlwaysFailPolicy())).
build();

ControllerResult<ApiError> result = manager.incrementalAlterConfig(
MYTOPIC,
toMap(entry("abc", entry(SET, "123"))),
true
);

assertEquals(ApiError.NONE, result.response());
assertFalse(result.records().isEmpty());
}

private static class CheckForNullValuesPolicy implements AlterConfigPolicy {
Expand Down Expand Up @@ -405,7 +440,7 @@ public void testLegacyAlterConfigs() {
expectedRecords1, toMap(entry(MYTOPIC, ApiError.NONE))),
manager.legacyAlterConfigs(
toMap(entry(MYTOPIC, toMap(entry("abc", "456"), entry("def", "901")))),
true));
false));
for (ApiMessageAndVersion message : expectedRecords1) {
manager.replay((ConfigRecord) message.message());
}
Expand All @@ -419,7 +454,7 @@ expectedRecords1, toMap(entry(MYTOPIC, ApiError.NONE))),
CONFIG_RECORD.highestSupportedVersion())),
toMap(entry(MYTOPIC, ApiError.NONE))),
manager.legacyAlterConfigs(toMap(entry(MYTOPIC, toMap(entry("def", "901")))),
true));
false));
}

@ParameterizedTest
Expand Down
Loading