Skip to content

KAFKA-20506 kafka-configs.sh can't delete the config from a offline broker when using bootstrap controller #22113

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

Open
m1a2st wants to merge 3 commits into
base: 4.3
Choose a base branch
from
+52 −112
Open

KAFKA-20506 kafka-configs.sh can't delete the config from a offline broker when using bootstrap controller #22113

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6a8561a
to 4.3 branch
m1a2st Apr 22, 2026
ad5b9f4
addressed by comments
m1a2st Apr 23, 2026
2848fd4
Revert "addressed by comments"
m1a2st Apr 23, 2026
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
32 changes: 14 additions & 18 deletions core/src/main/scala/kafka/admin/ConfigCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,12 +184,15 @@ object ConfigCommand extends Logging {
val configResourceType = entityTypeHead match {
case TopicType => ConfigResource.Type.TOPIC
case ClientMetricsType => ConfigResource.Type.CLIENT_METRICS
case BrokerType => ConfigResource.Type.BROKER
case BrokerType =>
if (entityNameHead.nonEmpty)
validateBrokerId(entityNameHead, entityTypeHead)
ConfigResource.Type.BROKER
case GroupType => ConfigResource.Type.GROUP
case _ => throw new IllegalArgumentException(s"$entityNameHead is not a valid entity-type.")
}
try {
alterResourceConfig(adminClient, entityTypeHead, entityNameHead, configsToBeDeleted, configsToBeAdded, configResourceType)
alterResourceConfig(adminClient, entityNameHead, configsToBeDeleted, configsToBeAdded, configResourceType)
} catch {
case e: ExecutionException =>
e.getCause match {
Expand All @@ -202,7 +205,7 @@ object ConfigCommand extends Logging {
}

case BrokerLoggerConfigType =>
val validLoggers = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = true, describeAll = false).map(_.name)
val validLoggers = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false).map(_.name)
// fail the command if any of the configured broker loggers do not exist
val invalidBrokerLoggers = configsToBeDeleted.filterNot(validLoggers.contains) ++ configsToBeAdded.keys.filterNot(validLoggers.contains)
if (invalidBrokerLoggers.nonEmpty)
Expand Down Expand Up @@ -405,15 +408,7 @@ object ConfigCommand extends Logging {
}
}

private def alterResourceConfig(adminClient: Admin, entityTypeHead: String, entityNameHead: String, configsToBeDeleted: Seq[String], configsToBeAdded: Map[String, ConfigEntry], resourceType: ConfigResource.Type): Unit = {
val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
.map { entry => (entry.name, entry) }.toMap

// fail the command if any of the configs to be deleted does not exist
val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
if (invalidConfigs.nonEmpty)
throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")

private def alterResourceConfig(adminClient: Admin, entityNameHead: String, configsToBeDeleted: Seq[String], configsToBeAdded: Map[String, ConfigEntry], resourceType: ConfigResource.Type): Unit = {
val configResource = new ConfigResource(resourceType, entityNameHead)
val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
val addEntries = configsToBeAdded.values.map(k => new AlterConfigOp(k, AlterConfigOp.OpType.SET))
Expand All @@ -422,11 +417,12 @@ object ConfigCommand extends Logging {
adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
}

private def validateBrokerId(entityName: String, entityType: String): Unit = try entityName.toInt catch {
case _: NumberFormatException =>
throw new IllegalArgumentException(s"The entity name for $entityType must be a valid integer broker id, found: $entityName")
}

private def getResourceConfig(adminClient: Admin, entityType: String, entityName: String, includeSynonyms: Boolean, describeAll: Boolean) = {
def validateBrokerId(): Unit = try entityName.toInt catch {
case _: NumberFormatException =>
throw new IllegalArgumentException(s"The entity name for $entityType must be a valid integer broker id, found: $entityName")
}

val (configResourceType, dynamicConfigSource) = entityType match {
case TopicType =>
Expand All @@ -437,12 +433,12 @@ object ConfigCommand extends Logging {
case BrokerDefaultEntityName =>
(ConfigResource.Type.BROKER, Some(ConfigEntry.ConfigSource.DYNAMIC_DEFAULT_BROKER_CONFIG))
case _ =>
validateBrokerId()
validateBrokerId(entityName, entityType)
(ConfigResource.Type.BROKER, Some(ConfigEntry.ConfigSource.DYNAMIC_BROKER_CONFIG))
}
case BrokerLoggerConfigType =>
if (entityName.nonEmpty)
validateBrokerId()
validateBrokerId(entityName, entityType)
(ConfigResource.Type.BROKER_LOGGER, None)
case ClientMetricsType =>
(ConfigResource.Type.CLIENT_METRICS, Some(ConfigEntry.ConfigSource.DYNAMIC_CLIENT_METRICS_CONFIG))
Expand Down
1 change: 1 addition & 0 deletions docs/getting-started/upgrade.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ type: docs

### Notable changes in 4.3.0

* `kafka-configs.sh --alter --delete-config` no longer requires the specified config keys to exist on the target resource. Previously, attempting to delete a non-existent config key raised an `InvalidConfigurationException`. The deletion is now a no-op when the key does not exist, which allows managing configs for offline brokers via `--bootstrap-controller`. For further details, please refer to [KAFKA-20506](https://issues.apache.org/jira/browse/KAFKA-20506).
Copy link
Copy Markdown
Member

@chia7712 chia7712 Apr 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you please port this change to the trunk version as well?

* Support dynamically changing configs for dynamic quorum controllers. Previously only brokers and static quorum controllers were supported. For further details, please refer to [KAFKA-18928](https://issues.apache.org/jira/browse/KAFKA-18928).
* Two new configs have been introduced: `group.coordinator.cached.buffer.max.bytes` and `share.coordinator.cached.buffer.max.bytes`. They allow the respective coordinators to set the maximum buffer size retained for reuse. For further details, please refer to [KIP-1196](https://cwiki.apache.org/confluence/x/hA5JFg).
* The new config have been introduced: `remote.log.metadata.topic.min.isr` with 2 as default value. You can correct the min.insync.replicas for the existed __remote_log_metadata topic via kafka-configs.sh if needed. For further details, please refer to [KIP-1235](https://cwiki.apache.org/confluence/x/yommFw).
Expand Down
23 changes: 23 additions & 0 deletions tools/src/test/java/org/apache/kafka/tools/ConfigCommandIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -616,6 +616,29 @@ public void testUpdateBrokerConfigNotAffectedByInvalidConfig() {
}
}

@ClusterTest
public void testDeleteNonExistentConfigIsIdempotent() throws Exception {
String topicName = "test-delete-nonexistent-topic";
try (Admin client = cluster.admin()) {
client.createTopics(List.of(new NewTopic(topicName, 1, (short) 1))).all().get();

ConfigCommand.alterConfig(client, new ConfigCommand.ConfigCommandOptions(toArray(
List.of("--bootstrap-server", cluster.bootstrapServers(),
"--entity-type", "topics", "--entity-name", topicName,
"--alter", "--delete-config", "non.existent.config"))));

ConfigCommand.alterConfig(client, new ConfigCommand.ConfigCommandOptions(toArray(
List.of("--bootstrap-server", cluster.bootstrapServers(),
"--entity-type", "brokers", "--entity-name", defaultBrokerId,
"--alter", "--delete-config", "non.existent.config"))));

ConfigCommand.alterConfig(client, new ConfigCommand.ConfigCommandOptions(toArray(
List.of("--bootstrap-server", cluster.bootstrapServers(),
"--entity-type", "brokers", "--entity-default",
"--alter", "--delete-config", "non.existent.config"))));
}
}

@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")},
Expand Down
108 changes: 14 additions & 94 deletions tools/src/test/java/org/apache/kafka/tools/ConfigCommandTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -808,30 +808,13 @@ public void shouldAlterTopicConfig(boolean file) {
"--delete-config", "unclean.leader.election.enable"));
AtomicBoolean alteredConfigs = new AtomicBoolean();

ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, resourceName);
List<ConfigEntry> configEntries = List.of(newConfigEntry("min.insync.replicas", "1"), newConfigEntry("unclean.leader.election.enable", "1"));
KafkaFutureImpl<Map<ConfigResource, Config>> future = new KafkaFutureImpl<>();
future.complete(Map.of(resource, new Config(configEntries)));
DescribeConfigsResult describeResult = mock(DescribeConfigsResult.class);
when(describeResult.all()).thenReturn(future);

KafkaFutureImpl<Void> alterFuture = new KafkaFutureImpl<>();
alterFuture.complete(null);
AlterConfigsResult alterResult = mock(AlterConfigsResult.class);
when(alterResult.all()).thenReturn(alterFuture);

Node node = new Node(1, "localhost", 9092);
MockAdminClient mockAdminClient = new MockAdminClient(List.of(node), node) {
@Override
public synchronized DescribeConfigsResult describeConfigs(Collection<ConfigResource> resources, DescribeConfigsOptions options) {
assertFalse(options.includeSynonyms(), "Config synonyms requested unnecessarily");
assertEquals(1, resources.size());
ConfigResource res = resources.iterator().next();
assertEquals(ConfigResource.Type.TOPIC, res.type());
assertEquals(resourceName, res.name());
return describeResult;
}

@Override
public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs, AlterConfigsOptions options) {
assertEquals(1, configs.size());
Expand Down Expand Up @@ -861,7 +844,7 @@ public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResourc
};
ConfigCommand.alterConfig(mockAdminClient, alterOpts);
assertTrue(alteredConfigs.get());
verify(describeResult).all();
verify(alterResult).all();
}

public ConfigEntry newConfigEntry(String name, String value) {
Expand Down Expand Up @@ -974,16 +957,16 @@ public void shouldRaiseInvalidConfigurationExceptionWhenAddingInvalidBrokerLogge
@Test
public void shouldAddDefaultBrokerDynamicConfig() {
Node node = new Node(1, "localhost", 9092);
verifyAlterBrokerConfig(node, "", List.of("--entity-default"));
verifyAlterBrokerConfig(node, List.of("--entity-default"));
}

@Test
public void shouldAddBrokerDynamicConfig() {
Node node = new Node(1, "localhost", 9092);
verifyAlterBrokerConfig(node, "1", List.of("--entity-name", "1"));
verifyAlterBrokerConfig(node, List.of("--entity-name", "1"));
}

public void verifyAlterBrokerConfig(Node node, String resourceName, List<String> resourceOpts) {
public void verifyAlterBrokerConfig(Node node, List<String> resourceOpts) {
String[] optsList = toArray(List.of("--bootstrap-server", "localhost:9092",
"--entity-type", "brokers",
"--alter",
Expand All @@ -992,29 +975,12 @@ public void verifyAlterBrokerConfig(Node node, String resourceName, List<String>
Map<String, String> brokerConfigs = new HashMap<>();
brokerConfigs.put("num.io.threads", "5");

ConfigResource resource = new ConfigResource(ConfigResource.Type.BROKER, resourceName);
List<ConfigEntry> configEntries = List.of(new ConfigEntry("num.io.threads", "5"));
KafkaFutureImpl<Map<ConfigResource, Config>> future = new KafkaFutureImpl<>();
future.complete(Map.of(resource, new Config(configEntries)));
DescribeConfigsResult describeResult = mock(DescribeConfigsResult.class);
when(describeResult.all()).thenReturn(future);

KafkaFutureImpl<Void> alterFuture = new KafkaFutureImpl<>();
alterFuture.complete(null);
AlterConfigsResult alterResult = mock(AlterConfigsResult.class);
when(alterResult.all()).thenReturn(alterFuture);

MockAdminClient mockAdminClient = new MockAdminClient(List.of(node), node) {
@Override
public synchronized DescribeConfigsResult describeConfigs(Collection<ConfigResource> resources, DescribeConfigsOptions options) {
assertFalse(options.includeSynonyms(), "Config synonyms requested unnecessarily");
assertEquals(1, resources.size());
ConfigResource res = resources.iterator().next();
assertEquals(ConfigResource.Type.BROKER, res.type());
assertEquals(resourceName, res.name());
return describeResult;
}

@Override
public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs, AlterConfigsOptions options) {
assertEquals(1, configs.size());
Expand All @@ -1032,7 +998,7 @@ public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResourc
expected.put("num.io.threads", "5");
expected.put("leader.replication.throttled.rate", "10");
assertEquals(expected, brokerConfigs);
verify(describeResult).all();
verify(alterResult).all();
}

@Test
Expand Down Expand Up @@ -1161,35 +1127,29 @@ public void shouldNotUpdateBrokerConfigIfMalformedBracketConfig() {
}

@Test
public void shouldNotUpdateConfigIfNonExistingConfigIsDeleted() {
public void shouldAllowDeletingNonExistingConfig() throws Exception {
String resourceName = "my-topic";
ConfigCommand.ConfigCommandOptions createOpts = new ConfigCommand.ConfigCommandOptions(toArray("--bootstrap-server", "localhost:9092",
"--entity-name", resourceName,
"--entity-type", "topics",
"--alter",
"--delete-config", "missing_config1, missing_config2"));

ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, resourceName);
List<ConfigEntry> configEntries = List.of();
KafkaFutureImpl<Map<ConfigResource, Config>> future = new KafkaFutureImpl<>();
future.complete(Map.of(resource, new Config(configEntries)));
DescribeConfigsResult describeResult = mock(DescribeConfigsResult.class);
when(describeResult.all()).thenReturn(future);
KafkaFutureImpl<Void> alterFuture = new KafkaFutureImpl<>();
alterFuture.complete(null);
AlterConfigsResult alterResult = mock(AlterConfigsResult.class);
when(alterResult.all()).thenReturn(alterFuture);

Node node = new Node(1, "localhost", 9092);
MockAdminClient mockAdminClient = new MockAdminClient(List.of(node), node) {
@Override
public synchronized DescribeConfigsResult describeConfigs(Collection<ConfigResource> resources, DescribeConfigsOptions options) {
assertEquals(1, resources.size());
ConfigResource res = resources.iterator().next();
assertEquals(ConfigResource.Type.TOPIC, res.type());
assertEquals(resourceName, res.name());
return describeResult;
public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs, AlterConfigsOptions options) {
return alterResult;
}
};

assertThrows(InvalidConfigurationException.class, () -> ConfigCommand.alterConfig(mockAdminClient, createOpts));
verify(describeResult).all();
ConfigCommand.alterConfig(mockAdminClient, createOpts);
verify(alterResult).all();
}

@Test
Expand All @@ -1210,31 +1170,12 @@ private void verifyAlterClientMetricsConfig(Node node, String resourceName, List
"match=[client_software_name=kafka.python,client_software_version=1\\.2\\..*]"), resourceOpts);
ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray(optsList));

ConfigResource resource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, resourceName);
List<ConfigEntry> configEntries = List.of(new ConfigEntry("interval.ms", "1000",
ConfigEntry.ConfigSource.DYNAMIC_CLIENT_METRICS_CONFIG, false, false, List.of(),
ConfigEntry.ConfigType.UNKNOWN, null));
KafkaFutureImpl<Map<ConfigResource, Config>> future = new KafkaFutureImpl<>();
future.complete(Map.of(resource, new Config(configEntries)));
DescribeConfigsResult describeResult = mock(DescribeConfigsResult.class);
when(describeResult.all()).thenReturn(future);

KafkaFutureImpl<Void> alterFuture = new KafkaFutureImpl<>();
alterFuture.complete(null);
AlterConfigsResult alterResult = mock(AlterConfigsResult.class);
when(alterResult.all()).thenReturn(alterFuture);

MockAdminClient mockAdminClient = new MockAdminClient(List.of(node), node) {
@Override
public synchronized DescribeConfigsResult describeConfigs(Collection<ConfigResource> resources, DescribeConfigsOptions options) {
assertFalse(options.includeSynonyms(), "Config synonyms requested unnecessarily");
assertEquals(1, resources.size());
ConfigResource res = resources.iterator().next();
assertEquals(ConfigResource.Type.CLIENT_METRICS, res.type());
assertEquals(resourceName, res.name());
return describeResult;
}

@Override
public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs, AlterConfigsOptions options) {
assertEquals(1, configs.size());
Expand All @@ -1258,7 +1199,6 @@ public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResourc
}
};
ConfigCommand.alterConfig(mockAdminClient, alterOpts);
verify(describeResult).all();
verify(alterResult).all();
}

Expand Down Expand Up @@ -1320,31 +1260,12 @@ private void verifyAlterGroupConfig(Node node, String resourceName, List<String>
"--add-config", "consumer.heartbeat.interval.ms=6000"), resourceOpts);
ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray(optsList));

ConfigResource resource = new ConfigResource(ConfigResource.Type.GROUP, resourceName);
List<ConfigEntry> configEntries = List.of(new ConfigEntry("consumer.session.timeout.ms", "45000",
ConfigEntry.ConfigSource.DYNAMIC_GROUP_CONFIG, false, false, List.of(),
ConfigEntry.ConfigType.UNKNOWN, null));
KafkaFutureImpl<Map<ConfigResource, Config>> future = new KafkaFutureImpl<>();
future.complete(Map.of(resource, new Config(configEntries)));
DescribeConfigsResult describeResult = mock(DescribeConfigsResult.class);
when(describeResult.all()).thenReturn(future);

KafkaFutureImpl<Void> alterFuture = new KafkaFutureImpl<>();
alterFuture.complete(null);
AlterConfigsResult alterResult = mock(AlterConfigsResult.class);
when(alterResult.all()).thenReturn(alterFuture);

MockAdminClient mockAdminClient = new MockAdminClient(List.of(node), node) {
@Override
public synchronized DescribeConfigsResult describeConfigs(Collection<ConfigResource> resources, DescribeConfigsOptions options) {
assertFalse(options.includeSynonyms(), "Config synonyms requested unnecessarily");
assertEquals(1, resources.size());
ConfigResource res = resources.iterator().next();
assertEquals(ConfigResource.Type.GROUP, res.type());
assertEquals(resourceName, res.name());
return describeResult;
}

@Override
public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs, AlterConfigsOptions options) {
assertEquals(1, configs.size());
Expand All @@ -1367,7 +1288,6 @@ public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResourc
}
};
ConfigCommand.alterConfig(mockAdminClient, alterOpts);
verify(describeResult).all();
verify(alterResult).all();
}

Expand Down
Loading