From f215091a7375ce423527f914ca07582c2c6de346 Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Wed, 6 May 2026 12:39:36 +0530 Subject: [PATCH 01/14] CCDB-4523: Strip config.providers from kafka-ready preflight check When the Connect worker config includes config.providers entries (e.g. for AWS Secrets Manager), the kafka-ready preflight check fails with ClassNotFoundException because the config provider plugin JARs are on the worker's plugin.path but not on the CUB/UB classpath. The kafka-ready check only needs Kafka client properties (bootstrap servers, security config) to verify broker connectivity. Config providers are irrelevant for this check, so strip them before creating the AdminClient. Removed entries are logged at WARN level. --- .../admin/utils/cli/KafkaReadyCommand.java | 31 ++++++ .../utils/cli/KafkaReadyCommandTest.java | 98 +++++++++++++++++++ 2 files changed, 129 insertions(+) create mode 100644 utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 014b031529..7ebef8e30f 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -28,6 +28,7 @@ import org.apache.logging.log4j.Logger; import java.util.HashMap; +import java.util.Iterator; import java.util.Map; import io.confluent.admin.utils.ClusterStatus; @@ -52,6 +53,7 @@ public class KafkaReadyCommand { private static final Logger log = LogManager.getLogger(KafkaReadyCommand.class); public static final String KAFKA_READY = "kafka-ready"; + private static final String CONFIG_PROVIDERS_PREFIX = "config.providers"; private static ArgumentParser createArgsParser() { ArgumentParser kafkaReady = ArgumentParsers @@ -118,6 +120,7 @@ public static void main(String[] args) { } else { if (res.getString("config") != null) { workerProps = Utils.propsToStringMap(Utils.loadProps(res.getString("config"))); + stripConfigProviders(workerProps); } if (res.getString("bootstrap_servers") != null) { workerProps.put( @@ -156,4 +159,32 @@ public static void main(String[] args) { System.exit(1); } } + + /** + * Removes config.providers entries from the properties map. These entries cause + * ClassNotFoundException during AdminClient creation when the config provider + * plugin JARs are not on the kafka-ready classpath (they live on the worker's + * plugin.path instead). The kafka-ready preflight check does not need config + * providers, so stripping them is safe. + */ + static void stripConfigProviders(Map props) { + Iterator> it = props.entrySet().iterator(); + boolean found = false; + while (it.hasNext()) { + String key = it.next().getKey(); + if (key.equals(CONFIG_PROVIDERS_PREFIX) || key.startsWith(CONFIG_PROVIDERS_PREFIX + ".")) { + log.warn("Removing property '{}' from kafka-ready config — config provider classes " + + "are not needed for the preflight broker connectivity check and may not be on " + + "the classpath.", key); + it.remove(); + found = true; + } + } + if (found) { + log.warn("Config provider properties were stripped from the kafka-ready check. " + + "If your Connect worker config uses config providers to resolve security " + + "properties (e.g. passwords from a secrets manager), ensure those values are " + + "also available as literal values or environment variables for the preflight check."); + } + } } diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java new file mode 100644 index 0000000000..dcea5fcd68 --- /dev/null +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -0,0 +1,98 @@ +/* + * Copyright 2017 Confluent Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.confluent.admin.utils.cli; + +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +public class KafkaReadyCommandTest { + + @Test + public void stripConfigProviders_removesAllConfigProviderEntries() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("security.protocol", "SASL_SSL"); + props.put("sasl.mechanism", "PLAIN"); + props.put("config.providers", "secretmanager"); + props.put("config.providers.secretmanager.class", + "io.confluent.csid.config.provider.aws.SecretsManagerConfigProvider"); + props.put("config.providers.secretmanager.param.aws.region", "us-east-1"); + + KafkaReadyCommand.stripConfigProviders(props); + + assertThat(props).containsOnlyKeys( + "bootstrap.servers", "security.protocol", "sasl.mechanism"); + assertThat(props).doesNotContainKey("config.providers"); + assertThat(props).doesNotContainKey("config.providers.secretmanager.class"); + assertThat(props).doesNotContainKey("config.providers.secretmanager.param.aws.region"); + } + + @Test + public void stripConfigProviders_noopWhenNoConfigProviders() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("security.protocol", "PLAINTEXT"); + + KafkaReadyCommand.stripConfigProviders(props); + + assertThat(props).hasSize(2); + assertThat(props).containsEntry("bootstrap.servers", "localhost:9092"); + assertThat(props).containsEntry("security.protocol", "PLAINTEXT"); + } + + @Test + public void stripConfigProviders_handlesMultipleProviders() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("config.providers", "secretmanager,vault"); + props.put("config.providers.secretmanager.class", "com.example.SecretsProvider"); + props.put("config.providers.secretmanager.param.region", "us-east-1"); + props.put("config.providers.vault.class", "com.example.VaultProvider"); + props.put("config.providers.vault.param.address", "https://vault:8200"); + props.put("ssl.truststore.location", "/etc/ssl/truststore.jks"); + + KafkaReadyCommand.stripConfigProviders(props); + + assertThat(props).containsOnlyKeys("bootstrap.servers", "ssl.truststore.location"); + } + + @Test + public void stripConfigProviders_handlesEmptyMap() { + Map props = new HashMap<>(); + + KafkaReadyCommand.stripConfigProviders(props); + + assertThat(props).isEmpty(); + } + + @Test + public void stripConfigProviders_doesNotRemoveSimilarKeys() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("config.providers", "secretmanager"); + props.put("config.providers.secretmanager.class", "com.example.Provider"); + // This key should NOT be removed — it doesn't match the prefix + props.put("some.config.providers.unrelated", "value"); + + KafkaReadyCommand.stripConfigProviders(props); + + assertThat(props).containsOnlyKeys("bootstrap.servers", "some.config.providers.unrelated"); + } +} From 2c758d7218755d2b04d77fb3ba2f3602a0d48b09 Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Wed, 6 May 2026 12:58:40 +0530 Subject: [PATCH 02/14] CCDB-4523: Make kafka-ready preflight resilient to config provider class loading errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the Connect worker config includes config.providers entries (e.g. for AWS Secrets Manager), the kafka-ready preflight check fails with ClassNotFoundException because the config provider plugin JARs are on the worker's plugin.path but not on the CUB/UB classpath. The fix uses a try/catch/retry strategy: 1. Try the check with the full config (works if providers are on classpath) 2. On ConfigException from class loading, strip config.providers entries 3. If remaining properties have unresolved ${provider:...} variable references (security props depend on config providers), skip the check with a WARN — the Connect worker will verify connectivity on startup 4. Otherwise, retry the check without config.providers --- .../admin/utils/cli/KafkaReadyCommand.java | 101 ++++++++++++++---- .../utils/cli/KafkaReadyCommandTest.java | 89 ++++++++++++--- 2 files changed, 158 insertions(+), 32 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 7ebef8e30f..6ec3351dc2 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -29,7 +29,12 @@ import java.util.HashMap; import java.util.Iterator; +import java.util.List; import java.util.Map; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +import org.apache.kafka.common.config.ConfigException; import io.confluent.admin.utils.ClusterStatus; @@ -54,6 +59,8 @@ public class KafkaReadyCommand { private static final Logger log = LogManager.getLogger(KafkaReadyCommand.class); public static final String KAFKA_READY = "kafka-ready"; private static final String CONFIG_PROVIDERS_PREFIX = "config.providers"; + // Matches Kafka config provider variable syntax: ${provider-name:path:key} + private static final Pattern CONFIG_PROVIDER_VAR_PATTERN = Pattern.compile("\\$\\{[^}]+}"); private static ArgumentParser createArgsParser() { ArgumentParser kafkaReady = ArgumentParsers @@ -120,7 +127,6 @@ public static void main(String[] args) { } else { if (res.getString("config") != null) { workerProps = Utils.propsToStringMap(Utils.loadProps(res.getString("config"))); - stripConfigProviders(workerProps); } if (res.getString("bootstrap_servers") != null) { workerProps.put( @@ -133,7 +139,7 @@ public static void main(String[] args) { "Bootstrap servers should be provided through config or bootstrap_servers" ); } - success = ClusterStatus.isKafkaReady( + success = checkKafkaReadyWithConfigProviderResilience( workerProps, res.getInt("min_expected_brokers"), res.getInt("timeout") @@ -161,30 +167,89 @@ public static void main(String[] args) { } /** - * Removes config.providers entries from the properties map. These entries cause - * ClassNotFoundException during AdminClient creation when the config provider - * plugin JARs are not on the kafka-ready classpath (they live on the worker's - * plugin.path instead). The kafka-ready preflight check does not need config - * providers, so stripping them is safe. + * Attempts the kafka-ready check, handling config provider class loading failures gracefully. + * + *

Strategy: + * 1. Try with the full config (config providers may be loadable if on the classpath). + * 2. If config provider class loading fails (ConfigException wrapping ClassNotFoundException), + * strip config.providers entries and check for unresolved variable references. + * 3. If remaining properties contain unresolved ${provider:...} references (i.e. security + * properties depend on config providers), skip the check with a warning — the Connect + * worker will verify connectivity itself on startup. + * 4. Otherwise, retry the check without config.providers. + */ + static boolean checkKafkaReadyWithConfigProviderResilience( + Map workerProps, + int minBrokerCount, + int timeoutMs + ) { + if (!hasConfigProviders(workerProps)) { + return ClusterStatus.isKafkaReady(workerProps, minBrokerCount, timeoutMs); + } + + try { + return ClusterStatus.isKafkaReady(workerProps, minBrokerCount, timeoutMs); + } catch (ConfigException e) { + if (!isConfigProviderLoadFailure(e)) { + throw e; + } + log.warn("Config provider class could not be loaded during kafka-ready check: {}", + e.getMessage()); + + stripConfigProviders(workerProps); + + List unresolvedKeys = findUnresolvedConfigProviderVars(workerProps); + if (!unresolvedKeys.isEmpty()) { + log.warn("Skipping kafka-ready check — the following properties contain unresolved " + + "config provider variable references that cannot be resolved without the " + + "config provider plugin: {}. The Connect worker will verify broker connectivity " + + "on startup.", unresolvedKeys); + return true; + } + + log.warn("Retrying kafka-ready check without config provider properties."); + return ClusterStatus.isKafkaReady(workerProps, minBrokerCount, timeoutMs); + } + } + + /** + * Returns true if the properties map contains config.providers entries. + */ + static boolean hasConfigProviders(Map props) { + return props.containsKey(CONFIG_PROVIDERS_PREFIX); + } + + /** + * Returns true if the exception is a ConfigException caused by a config provider + * class loading failure. + */ + static boolean isConfigProviderLoadFailure(ConfigException e) { + String msg = e.getMessage(); + return msg != null && msg.contains("config.providers") && msg.contains("Could not load"); + } + + /** + * Removes config.providers entries from the properties map. */ static void stripConfigProviders(Map props) { Iterator> it = props.entrySet().iterator(); - boolean found = false; while (it.hasNext()) { String key = it.next().getKey(); if (key.equals(CONFIG_PROVIDERS_PREFIX) || key.startsWith(CONFIG_PROVIDERS_PREFIX + ".")) { - log.warn("Removing property '{}' from kafka-ready config — config provider classes " - + "are not needed for the preflight broker connectivity check and may not be on " - + "the classpath.", key); + log.warn("Removing property '{}' from kafka-ready config.", key); it.remove(); - found = true; } } - if (found) { - log.warn("Config provider properties were stripped from the kafka-ready check. " - + "If your Connect worker config uses config providers to resolve security " - + "properties (e.g. passwords from a secrets manager), ensure those values are " - + "also available as literal values or environment variables for the preflight check."); - } + } + + /** + * Returns property keys whose values contain unresolved config provider variable + * references (e.g. ${secretmanager:path:key}). + */ + static List findUnresolvedConfigProviderVars(Map props) { + return props.entrySet().stream() + .filter(e -> CONFIG_PROVIDER_VAR_PATTERN.matcher(e.getValue()).find()) + .map(Map.Entry::getKey) + .collect(Collectors.toList()); } } diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java index dcea5fcd68..4826f843e6 100644 --- a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -15,15 +15,19 @@ */ package io.confluent.admin.utils.cli; +import org.apache.kafka.common.config.ConfigException; import org.junit.jupiter.api.Test; import java.util.HashMap; +import java.util.List; import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; public class KafkaReadyCommandTest { + // --- stripConfigProviders --- + @Test public void stripConfigProviders_removesAllConfigProviderEntries() { Map props = new HashMap<>(); @@ -39,9 +43,6 @@ public void stripConfigProviders_removesAllConfigProviderEntries() { assertThat(props).containsOnlyKeys( "bootstrap.servers", "security.protocol", "sasl.mechanism"); - assertThat(props).doesNotContainKey("config.providers"); - assertThat(props).doesNotContainKey("config.providers.secretmanager.class"); - assertThat(props).doesNotContainKey("config.providers.secretmanager.param.aws.region"); } @Test @@ -54,7 +55,6 @@ public void stripConfigProviders_noopWhenNoConfigProviders() { assertThat(props).hasSize(2); assertThat(props).containsEntry("bootstrap.servers", "localhost:9092"); - assertThat(props).containsEntry("security.protocol", "PLAINTEXT"); } @Test @@ -63,7 +63,6 @@ public void stripConfigProviders_handlesMultipleProviders() { props.put("bootstrap.servers", "localhost:9092"); props.put("config.providers", "secretmanager,vault"); props.put("config.providers.secretmanager.class", "com.example.SecretsProvider"); - props.put("config.providers.secretmanager.param.region", "us-east-1"); props.put("config.providers.vault.class", "com.example.VaultProvider"); props.put("config.providers.vault.param.address", "https://vault:8200"); props.put("ssl.truststore.location", "/etc/ssl/truststore.jks"); @@ -74,25 +73,87 @@ public void stripConfigProviders_handlesMultipleProviders() { } @Test - public void stripConfigProviders_handlesEmptyMap() { + public void stripConfigProviders_doesNotRemoveSimilarKeys() { Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("config.providers", "secretmanager"); + props.put("config.providers.secretmanager.class", "com.example.Provider"); + props.put("some.config.providers.unrelated", "value"); KafkaReadyCommand.stripConfigProviders(props); - assertThat(props).isEmpty(); + assertThat(props).containsOnlyKeys("bootstrap.servers", "some.config.providers.unrelated"); } + // --- hasConfigProviders --- + @Test - public void stripConfigProviders_doesNotRemoveSimilarKeys() { + public void hasConfigProviders_trueWhenPresent() { Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); props.put("config.providers", "secretmanager"); - props.put("config.providers.secretmanager.class", "com.example.Provider"); - // This key should NOT be removed — it doesn't match the prefix - props.put("some.config.providers.unrelated", "value"); - KafkaReadyCommand.stripConfigProviders(props); + assertThat(KafkaReadyCommand.hasConfigProviders(props)).isTrue(); + } - assertThat(props).containsOnlyKeys("bootstrap.servers", "some.config.providers.unrelated"); + @Test + public void hasConfigProviders_falseWhenAbsent() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + + assertThat(KafkaReadyCommand.hasConfigProviders(props)).isFalse(); + } + + // --- isConfigProviderLoadFailure --- + + @Test + public void isConfigProviderLoadFailure_trueForClassLoadError() { + ConfigException e = new ConfigException( + "Invalid value io.confluent.csid.config.provider.aws.SecretsManagerConfigProvider " + + "for configuration config.providers.secretmanager.class: " + + "Could not load config provider class or one of its dependencies"); + + assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); + } + + @Test + public void isConfigProviderLoadFailure_falseForOtherConfigException() { + ConfigException e = new ConfigException("Unknown configuration key: bad.key"); + + assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isFalse(); + } + + // --- findUnresolvedConfigProviderVars --- + + @Test + public void findUnresolvedVars_detectsProviderReferences() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("ssl.keystore.password", "${secretmanager:my-secret:keystore-password}"); + props.put("sasl.jaas.config", "${vault:secret/kafka:jaas-config}"); + props.put("security.protocol", "SASL_SSL"); + + List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(props); + + assertThat(unresolved).hasSize(2); + assertThat(unresolved).contains("ssl.keystore.password", "sasl.jaas.config"); + } + + @Test + public void findUnresolvedVars_emptyWhenAllLiteral() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("ssl.keystore.password", "my-literal-password"); + props.put("security.protocol", "SSL"); + + List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(props); + + assertThat(unresolved).isEmpty(); + } + + @Test + public void findUnresolvedVars_handlesEmptyMap() { + List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(new HashMap<>()); + + assertThat(unresolved).isEmpty(); } } From 66c12009771261616a69ccbf748748a764b731ca Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Wed, 6 May 2026 15:05:50 +0530 Subject: [PATCH 03/14] Address review feedback: consolidate log lines, fix em dash, add stack trace - Consolidate per-key WARN logs into a single message with count + key list - Replace Unicode em dash with plain hyphen for ASCII-safe log parsing - Include exception stack trace in config provider load failure warning --- .../confluent/admin/utils/cli/KafkaReadyCommand.java | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 6ec3351dc2..3ebd24ba6b 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -194,13 +194,13 @@ static boolean checkKafkaReadyWithConfigProviderResilience( throw e; } log.warn("Config provider class could not be loaded during kafka-ready check: {}", - e.getMessage()); + e.getMessage(), e); stripConfigProviders(workerProps); List unresolvedKeys = findUnresolvedConfigProviderVars(workerProps); if (!unresolvedKeys.isEmpty()) { - log.warn("Skipping kafka-ready check — the following properties contain unresolved " + log.warn("Skipping kafka-ready check - the following properties contain unresolved " + "config provider variable references that cannot be resolved without the " + "config provider plugin: {}. The Connect worker will verify broker connectivity " + "on startup.", unresolvedKeys); @@ -232,14 +232,19 @@ static boolean isConfigProviderLoadFailure(ConfigException e) { * Removes config.providers entries from the properties map. */ static void stripConfigProviders(Map props) { + List removed = new java.util.ArrayList<>(); Iterator> it = props.entrySet().iterator(); while (it.hasNext()) { String key = it.next().getKey(); if (key.equals(CONFIG_PROVIDERS_PREFIX) || key.startsWith(CONFIG_PROVIDERS_PREFIX + ".")) { - log.warn("Removing property '{}' from kafka-ready config.", key); + removed.add(key); it.remove(); } } + if (!removed.isEmpty()) { + log.warn("Removed {} config provider properties from kafka-ready config: {}", + removed.size(), removed); + } } /** From 5fe8ad0e886c644dbeb8620426577f102d588068 Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Wed, 6 May 2026 16:38:22 +0530 Subject: [PATCH 04/14] Address Copilot review: tighten regex, check cause chain, add tests - Tighten CONFIG_PROVIDER_VAR_PATTERN to require at least one colon separator (${provider:path[:key]}), avoiding false matches on non-provider ${...} placeholders like ${ENV_VAR} - Check exception cause chain for ClassNotFoundException / NoClassDefFoundError instead of brittle message string matching; keep message check as fallback - Add 4 new tests: ClassNotFoundException cause, NoClassDefFoundError cause, nested cause chain, non-provider placeholder filtering --- .../admin/utils/cli/KafkaReadyCommand.java | 17 +++++-- .../utils/cli/KafkaReadyCommandTest.java | 47 ++++++++++++++++++- 2 files changed, 59 insertions(+), 5 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 3ebd24ba6b..0284824746 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -59,8 +59,10 @@ public class KafkaReadyCommand { private static final Logger log = LogManager.getLogger(KafkaReadyCommand.class); public static final String KAFKA_READY = "kafka-ready"; private static final String CONFIG_PROVIDERS_PREFIX = "config.providers"; - // Matches Kafka config provider variable syntax: ${provider-name:path:key} - private static final Pattern CONFIG_PROVIDER_VAR_PATTERN = Pattern.compile("\\$\\{[^}]+}"); + // Matches Kafka config provider variable syntax: ${provider-name:path[:key]} + // Requires at least one colon separator to distinguish from other ${...} placeholders. + private static final Pattern CONFIG_PROVIDER_VAR_PATTERN = + Pattern.compile("\\$\\{[^:}]+:[^}]+}"); private static ArgumentParser createArgsParser() { ArgumentParser kafkaReady = ArgumentParsers @@ -221,9 +223,18 @@ static boolean hasConfigProviders(Map props) { /** * Returns true if the exception is a ConfigException caused by a config provider - * class loading failure. + * class loading failure. Checks both the cause chain for ClassNotFoundException / + * NoClassDefFoundError and the message for config.providers references as a fallback. */ static boolean isConfigProviderLoadFailure(ConfigException e) { + Throwable cause = e.getCause(); + while (cause != null) { + if (cause instanceof ClassNotFoundException || cause instanceof NoClassDefFoundError) { + return true; + } + cause = cause.getCause(); + } + // Fallback: check message in case the cause chain is truncated String msg = e.getMessage(); return msg != null && msg.contains("config.providers") && msg.contains("Could not load"); } diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java index 4826f843e6..6f59ae42c1 100644 --- a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -106,9 +106,37 @@ public void hasConfigProviders_falseWhenAbsent() { // --- isConfigProviderLoadFailure --- @Test - public void isConfigProviderLoadFailure_trueForClassLoadError() { + public void isConfigProviderLoadFailure_trueForClassNotFoundException() { + ConfigException e = new ConfigException("Could not load config provider class"); + e.initCause(new ClassNotFoundException( + "io.confluent.csid.config.provider.aws.SecretsManagerConfigProvider")); + + assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); + } + + @Test + public void isConfigProviderLoadFailure_trueForNoClassDefFoundError() { + ConfigException e = new ConfigException("Could not instantiate config provider"); + e.initCause(new NoClassDefFoundError("software/amazon/awssdk/core/SdkClient")); + + assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); + } + + @Test + public void isConfigProviderLoadFailure_trueForNestedClassNotFound() { + ClassNotFoundException cnfe = new ClassNotFoundException("com.example.Provider"); + RuntimeException wrapper = new RuntimeException("wrapper", cnfe); + ConfigException e = new ConfigException("config error"); + e.initCause(wrapper); + + assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); + } + + @Test + public void isConfigProviderLoadFailure_trueForMessageFallback() { + // No ClassNotFoundException in cause chain, but message matches ConfigException e = new ConfigException( - "Invalid value io.confluent.csid.config.provider.aws.SecretsManagerConfigProvider " + "Invalid value com.example.Provider " + "for configuration config.providers.secretmanager.class: " + "Could not load config provider class or one of its dependencies"); @@ -138,6 +166,21 @@ public void findUnresolvedVars_detectsProviderReferences() { assertThat(unresolved).contains("ssl.keystore.password", "sasl.jaas.config"); } + @Test + public void findUnresolvedVars_ignoresNonProviderPlaceholders() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + // Shell-style ${ENV_VAR} without colons should NOT be matched + props.put("some.prop", "${SOME_ENV_VAR}"); + // But provider-style ${provider:path} should be matched + props.put("ssl.keystore.password", "${secretmanager:my-secret:password}"); + + List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(props); + + assertThat(unresolved).hasSize(1); + assertThat(unresolved).contains("ssl.keystore.password"); + } + @Test public void findUnresolvedVars_emptyWhenAllLiteral() { Map props = new HashMap<>(); From 344519ebf04d6dff13fbb1eac346817eecf5f761 Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Thu, 7 May 2026 11:57:29 +0530 Subject: [PATCH 05/14] Address Copilot review round 2: testable orchestration, tighter regex, cause chain check - Extract ClusterStatus.isKafkaReady behind KafkaReadyChecker functional interface so checkKafkaReadyWithConfigProviderResilience orchestration logic can be unit tested with mocked checkers - Add 5 orchestration tests covering all scenarios: no config providers, providers loadable (first try succeeds), class not found with literal props (strip + retry), class not found with unresolved vars (skip), non-provider ConfigException (rethrow) - Tighten CONFIG_PROVIDER_VAR_PATTERN to require at least one colon separator, avoiding false matches on non-provider ${...} placeholders - Check exception cause chain for ClassNotFoundException / NoClassDefFoundError instead of brittle message string matching; keep message check as fallback for truncated cause chains --- .../admin/utils/cli/KafkaReadyCommand.java | 22 +++- .../utils/cli/KafkaReadyCommandTest.java | 114 ++++++++++++++++++ 2 files changed, 133 insertions(+), 3 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 0284824746..3a501e9beb 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -59,6 +59,12 @@ public class KafkaReadyCommand { private static final Logger log = LogManager.getLogger(KafkaReadyCommand.class); public static final String KAFKA_READY = "kafka-ready"; private static final String CONFIG_PROVIDERS_PREFIX = "config.providers"; + + @FunctionalInterface + interface KafkaReadyChecker { + boolean isReady(Map config, int minBrokerCount, int timeoutMs); + } + // Matches Kafka config provider variable syntax: ${provider-name:path[:key]} // Requires at least one colon separator to distinguish from other ${...} placeholders. private static final Pattern CONFIG_PROVIDER_VAR_PATTERN = @@ -184,13 +190,23 @@ static boolean checkKafkaReadyWithConfigProviderResilience( Map workerProps, int minBrokerCount, int timeoutMs + ) { + return checkKafkaReadyWithConfigProviderResilience( + workerProps, minBrokerCount, timeoutMs, ClusterStatus::isKafkaReady); + } + + static boolean checkKafkaReadyWithConfigProviderResilience( + Map workerProps, + int minBrokerCount, + int timeoutMs, + KafkaReadyChecker checker ) { if (!hasConfigProviders(workerProps)) { - return ClusterStatus.isKafkaReady(workerProps, minBrokerCount, timeoutMs); + return checker.isReady(workerProps, minBrokerCount, timeoutMs); } try { - return ClusterStatus.isKafkaReady(workerProps, minBrokerCount, timeoutMs); + return checker.isReady(workerProps, minBrokerCount, timeoutMs); } catch (ConfigException e) { if (!isConfigProviderLoadFailure(e)) { throw e; @@ -210,7 +226,7 @@ static boolean checkKafkaReadyWithConfigProviderResilience( } log.warn("Retrying kafka-ready check without config provider properties."); - return ClusterStatus.isKafkaReady(workerProps, minBrokerCount, timeoutMs); + return checker.isReady(workerProps, minBrokerCount, timeoutMs); } } diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java index 6f59ae42c1..db2b6cf593 100644 --- a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -21,8 +21,10 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; public class KafkaReadyCommandTest { @@ -199,4 +201,116 @@ public void findUnresolvedVars_handlesEmptyMap() { assertThat(unresolved).isEmpty(); } + + // --- checkKafkaReadyWithConfigProviderResilience orchestration --- + + @Test + public void resilience_noConfigProviders_callsCheckerOnce() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + AtomicInteger callCount = new AtomicInteger(0); + KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { + callCount.incrementAndGet(); + return true; + }; + + boolean result = KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( + props, 1, 5000, checker); + + assertThat(result).isTrue(); + assertThat(callCount.get()).isEqualTo(1); + } + + @Test + public void resilience_configProvidersLoadable_succeedsFirstTry() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("config.providers", "secretmanager"); + props.put("config.providers.secretmanager.class", "com.example.Provider"); + AtomicInteger callCount = new AtomicInteger(0); + KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { + callCount.incrementAndGet(); + return true; + }; + + boolean result = KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( + props, 1, 5000, checker); + + assertThat(result).isTrue(); + assertThat(callCount.get()).isEqualTo(1); + // config.providers still present since first try succeeded + assertThat(props).containsKey("config.providers"); + } + + @Test + public void resilience_classNotFound_literalProps_stripsAndRetries() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("security.protocol", "SASL_SSL"); + props.put("sasl.mechanism", "PLAIN"); + props.put("config.providers", "secretmanager"); + props.put("config.providers.secretmanager.class", "com.example.MissingProvider"); + AtomicInteger callCount = new AtomicInteger(0); + KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { + int call = callCount.incrementAndGet(); + if (call == 1) { + ConfigException ex = new ConfigException("Could not load"); + ex.initCause(new ClassNotFoundException("com.example.MissingProvider")); + throw ex; + } + return true; + }; + + boolean result = KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( + props, 1, 5000, checker); + + assertThat(result).isTrue(); + assertThat(callCount.get()).isEqualTo(2); + // config.providers should have been stripped before retry + assertThat(props).doesNotContainKey("config.providers"); + assertThat(props).doesNotContainKey("config.providers.secretmanager.class"); + assertThat(props).containsKey("bootstrap.servers"); + } + + @Test + public void resilience_classNotFound_unresolvedVars_skipsWithWarning() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("ssl.keystore.password", "${secretmanager:secret:password}"); + props.put("config.providers", "secretmanager"); + props.put("config.providers.secretmanager.class", "com.example.MissingProvider"); + AtomicInteger callCount = new AtomicInteger(0); + KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { + callCount.incrementAndGet(); + ConfigException ex = new ConfigException("Could not load"); + ex.initCause(new ClassNotFoundException("com.example.MissingProvider")); + throw ex; + }; + + boolean result = KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( + props, 1, 5000, checker); + + // Should skip (return true) without retrying + assertThat(result).isTrue(); + assertThat(callCount.get()).isEqualTo(1); + } + + @Test + public void resilience_nonProviderConfigException_rethrows() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + props.put("config.providers", "secretmanager"); + props.put("config.providers.secretmanager.class", "com.example.Provider"); + KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { + throw new ConfigException("Unknown configuration key: bad.key"); + }; + + try { + KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( + props, 1, 5000, checker); + fail("Expected ConfigException to be rethrown"); + } catch (ConfigException e) { + assertThat(e.getMessage()).contains("bad.key"); + } + } } From 3a60c812f8cd0b3bfcb25abdd0a76c3ef6501f56 Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Thu, 7 May 2026 15:06:46 +0530 Subject: [PATCH 06/14] Address Copilot review round 3: import ordering, tighter CNFE check - Move ConfigException import into org.apache.kafka.* group for consistent import ordering - Require exception message to mention "config.providers" before checking cause chain for ClassNotFoundException/NoClassDefFoundError, so SASL/SSL handler class loading failures are not misclassified - Add tests for SASL and SSL CNFE cases that must NOT match --- .../admin/utils/cli/KafkaReadyCommand.java | 23 +++++++----- .../utils/cli/KafkaReadyCommandTest.java | 36 ++++++++++++++++--- 2 files changed, 46 insertions(+), 13 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 3a501e9beb..38d8e9453a 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -23,6 +23,7 @@ import net.sourceforge.argparse4j.inf.Namespace; import org.apache.kafka.clients.CommonClientConfigs; +import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.utils.Utils; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -34,8 +35,6 @@ import java.util.regex.Pattern; import java.util.stream.Collectors; -import org.apache.kafka.common.config.ConfigException; - import io.confluent.admin.utils.ClusterStatus; import static net.sourceforge.argparse4j.impl.Arguments.store; @@ -238,11 +237,20 @@ static boolean hasConfigProviders(Map props) { } /** - * Returns true if the exception is a ConfigException caused by a config provider - * class loading failure. Checks both the cause chain for ClassNotFoundException / - * NoClassDefFoundError and the message for config.providers references as a fallback. + * Returns true if the exception is a ConfigException specifically caused by a config + * provider class loading failure. Requires both: + * - The exception message references "config.providers" (to distinguish from SASL/SSL + * handler class loading failures which also throw ConfigException with CNFE) + * - Either a ClassNotFoundException/NoClassDefFoundError in the cause chain, or + * "Could not load" in the message (fallback for truncated cause chains) */ static boolean isConfigProviderLoadFailure(ConfigException e) { + String msg = e.getMessage(); + boolean mentionsConfigProviders = msg != null && msg.contains("config.providers"); + if (!mentionsConfigProviders) { + return false; + } + Throwable cause = e.getCause(); while (cause != null) { if (cause instanceof ClassNotFoundException || cause instanceof NoClassDefFoundError) { @@ -250,9 +258,8 @@ static boolean isConfigProviderLoadFailure(ConfigException e) { } cause = cause.getCause(); } - // Fallback: check message in case the cause chain is truncated - String msg = e.getMessage(); - return msg != null && msg.contains("config.providers") && msg.contains("Could not load"); + // Fallback: message-only check when cause chain is truncated + return msg.contains("Could not load"); } /** diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java index db2b6cf593..c7e9c39ac1 100644 --- a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -109,16 +109,19 @@ public void hasConfigProviders_falseWhenAbsent() { @Test public void isConfigProviderLoadFailure_trueForClassNotFoundException() { - ConfigException e = new ConfigException("Could not load config provider class"); - e.initCause(new ClassNotFoundException( - "io.confluent.csid.config.provider.aws.SecretsManagerConfigProvider")); + ConfigException e = new ConfigException( + "Invalid value com.example.Provider for configuration " + + "config.providers.secretmanager.class: Could not load class"); + e.initCause(new ClassNotFoundException("com.example.Provider")); assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); } @Test public void isConfigProviderLoadFailure_trueForNoClassDefFoundError() { - ConfigException e = new ConfigException("Could not instantiate config provider"); + ConfigException e = new ConfigException( + "Invalid value com.example.Provider for configuration " + + "config.providers.secretmanager.class: Could not load class"); e.initCause(new NoClassDefFoundError("software/amazon/awssdk/core/SdkClient")); assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); @@ -128,7 +131,8 @@ public void isConfigProviderLoadFailure_trueForNoClassDefFoundError() { public void isConfigProviderLoadFailure_trueForNestedClassNotFound() { ClassNotFoundException cnfe = new ClassNotFoundException("com.example.Provider"); RuntimeException wrapper = new RuntimeException("wrapper", cnfe); - ConfigException e = new ConfigException("config error"); + ConfigException e = new ConfigException( + "config.providers.vault.class: Could not load"); e.initCause(wrapper); assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); @@ -152,6 +156,28 @@ public void isConfigProviderLoadFailure_falseForOtherConfigException() { assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isFalse(); } + @Test + public void isConfigProviderLoadFailure_falseForSaslClassNotFound() { + // CNFE for a SASL handler class should NOT be treated as a config provider failure + ConfigException e = new ConfigException( + "Invalid value com.example.SaslHandler for configuration " + + "sasl.client.callback.handler.class: Could not load class"); + e.initCause(new ClassNotFoundException("com.example.SaslHandler")); + + assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isFalse(); + } + + @Test + public void isConfigProviderLoadFailure_falseForSslClassNotFound() { + // CNFE for an SSL engine factory should NOT match + ConfigException e = new ConfigException( + "Invalid value com.example.SslFactory for configuration " + + "ssl.engine.factory.class: Could not load class"); + e.initCause(new ClassNotFoundException("com.example.SslFactory")); + + assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isFalse(); + } + // --- findUnresolvedConfigProviderVars --- @Test From 2f3c3e8a6f84ce20aaba7230ddd1268a3e9f7d4d Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Thu, 7 May 2026 20:04:08 +0530 Subject: [PATCH 07/14] Fix test failures and address Copilot review round 4 - Fix 2 test failures: mock ConfigException messages now include "config.providers" to match the tightened isConfigProviderLoadFailure predicate from round 3 - Limit findUnresolvedConfigProviderVars to Kafka client connectivity keys only (security.*, sasl.*, ssl.*, bootstrap.servers) so unresolved references in non-client keys (e.g. connector configs) don't cause the preflight to be skipped unnecessarily - Add test for non-client keys being ignored --- .../admin/utils/cli/KafkaReadyCommand.java | 24 +++++++++++++++++-- .../utils/cli/KafkaReadyCommandTest.java | 20 +++++++++++++++- 2 files changed, 41 insertions(+), 3 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 38d8e9453a..3a99dd36b7 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -69,6 +69,14 @@ interface KafkaReadyChecker { private static final Pattern CONFIG_PROVIDER_VAR_PATTERN = Pattern.compile("\\$\\{[^:}]+:[^}]+}"); + // Property key prefixes that affect Kafka client connectivity. Only these are checked + // for unresolved config provider variable references when deciding whether to skip + // the kafka-ready check. Non-client keys (e.g. connector configs) are irrelevant + // since AdminClient ignores them. + private static final String[] KAFKA_CLIENT_KEY_PREFIXES = { + "security.", "sasl.", "ssl.", "bootstrap.servers" + }; + private static ArgumentParser createArgsParser() { ArgumentParser kafkaReady = ArgumentParsers .newArgumentParser(KAFKA_READY) @@ -282,13 +290,25 @@ static void stripConfigProviders(Map props) { } /** - * Returns property keys whose values contain unresolved config provider variable - * references (e.g. ${secretmanager:path:key}). + * Returns Kafka client connectivity property keys whose values contain unresolved + * config provider variable references (e.g. ${secretmanager:path:key}). Only checks + * keys that affect broker connectivity (security.*, sasl.*, ssl.*, bootstrap.servers) + * since non-client keys are ignored by AdminClient and won't cause failures. */ static List findUnresolvedConfigProviderVars(Map props) { return props.entrySet().stream() + .filter(e -> isKafkaClientKey(e.getKey())) .filter(e -> CONFIG_PROVIDER_VAR_PATTERN.matcher(e.getValue()).find()) .map(Map.Entry::getKey) .collect(Collectors.toList()); } + + private static boolean isKafkaClientKey(String key) { + for (String prefix : KAFKA_CLIENT_KEY_PREFIXES) { + if (key.startsWith(prefix)) { + return true; + } + } + return false; + } } diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java index c7e9c39ac1..18c862813a 100644 --- a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -228,6 +228,22 @@ public void findUnresolvedVars_handlesEmptyMap() { assertThat(unresolved).isEmpty(); } + @Test + public void findUnresolvedVars_ignoresNonClientKeys() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", "localhost:9092"); + // Non-client key with provider reference should be ignored + props.put("connector.password", "${secretmanager:secret:connector-pw}"); + props.put("group.id", "${secretmanager:secret:group}"); + // Client key with provider reference should be detected + props.put("ssl.keystore.password", "${secretmanager:secret:keystore-pw}"); + + List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(props); + + assertThat(unresolved).hasSize(1); + assertThat(unresolved).contains("ssl.keystore.password"); + } + // --- checkKafkaReadyWithConfigProviderResilience orchestration --- @Test @@ -280,7 +296,9 @@ public void resilience_classNotFound_literalProps_stripsAndRetries() { KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { int call = callCount.incrementAndGet(); if (call == 1) { - ConfigException ex = new ConfigException("Could not load"); + ConfigException ex = new ConfigException( + "Invalid value com.example.MissingProvider for configuration " + + "config.providers.secretmanager.class: Could not load"); ex.initCause(new ClassNotFoundException("com.example.MissingProvider")); throw ex; } From a8c2d5a0ca25a7f4dbe11dfd67fc466c8256c1f4 Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Mon, 11 May 2026 12:32:33 +0530 Subject: [PATCH 08/14] Fix remaining test: mock ConfigException must include config.providers in message The resilience_classNotFound_unresolvedVars_skipsWithWarning test was still using a bare "Could not load" message without "config.providers", which the tightened isConfigProviderLoadFailure predicate rejects. --- .../io/confluent/admin/utils/cli/KafkaReadyCommandTest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java index 18c862813a..b5ce50cf20 100644 --- a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -326,7 +326,9 @@ public void resilience_classNotFound_unresolvedVars_skipsWithWarning() { AtomicInteger callCount = new AtomicInteger(0); KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { callCount.incrementAndGet(); - ConfigException ex = new ConfigException("Could not load"); + ConfigException ex = new ConfigException( + "Invalid value com.example.MissingProvider for configuration " + + "config.providers.secretmanager.class: Could not load"); ex.initCause(new ClassNotFoundException("com.example.MissingProvider")); throw ex; }; From 6b7535a57330c9fb952f24f5209df2992659579f Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Mon, 11 May 2026 13:17:31 +0530 Subject: [PATCH 09/14] Switch test to JUnit 4 and add junit-vintage-engine Convert KafkaReadyCommandTest from JUnit 5 to JUnit 4 for consistency with the rest of the module. Add junit-vintage-engine dependency so JUnit 4 tests are discovered by the JUnit Platform surefire provider. --- utility-belt/pom.xml | 7 +++++++ .../confluent/admin/utils/cli/KafkaReadyCommandTest.java | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/utility-belt/pom.xml b/utility-belt/pom.xml index 9485b3b81a..170422622a 100644 --- a/utility-belt/pom.xml +++ b/utility-belt/pom.xml @@ -221,6 +221,13 @@ test + + org.junit.vintage + junit-vintage-engine + ${junit.jupiter.version} + test + + org.apache.logging.log4j log4j-api diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java index b5ce50cf20..552e44abbe 100644 --- a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -16,7 +16,7 @@ package io.confluent.admin.utils.cli; import org.apache.kafka.common.config.ConfigException; -import org.junit.jupiter.api.Test; +import org.junit.Test; import java.util.HashMap; import java.util.List; From 969429816b71e1a45487188cd498b381aec4adb4 Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Mon, 11 May 2026 14:27:28 +0530 Subject: [PATCH 10/14] Remove junit-vintage-engine to avoid surfacing pre-existing test failures The vintage engine causes surefire to discover existing JUnit 4 tests that have a broken EmbeddedKafkaCluster setup. Removing it keeps the PR scoped to the kafka-ready fix without inheriting unrelated failures. --- utility-belt/pom.xml | 7 ------- 1 file changed, 7 deletions(-) diff --git a/utility-belt/pom.xml b/utility-belt/pom.xml index 170422622a..9485b3b81a 100644 --- a/utility-belt/pom.xml +++ b/utility-belt/pom.xml @@ -221,13 +221,6 @@ test - - org.junit.vintage - junit-vintage-engine - ${junit.jupiter.version} - test - - org.apache.logging.log4j log4j-api From 22ddc747ac08de9e63cb21ec6969a009ee268c0f Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Mon, 11 May 2026 15:24:17 +0530 Subject: [PATCH 11/14] Use defensive copy to avoid mutating caller's map, update javadoc - Operate on a copy of workerProps when stripping config.providers, so the caller's map is never mutated - Update javadoc to document fast-path and no-mutation guarantee - Update test assertion to verify original map is untouched --- .../admin/utils/cli/KafkaReadyCommand.java | 20 +++++++++++-------- .../utils/cli/KafkaReadyCommandTest.java | 6 +++--- 2 files changed, 15 insertions(+), 11 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 3a99dd36b7..224fc8e726 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -185,13 +185,16 @@ public static void main(String[] args) { * Attempts the kafka-ready check, handling config provider class loading failures gracefully. * *

Strategy: + * 0. If no config.providers entries are present, run the check directly (fast path). * 1. Try with the full config (config providers may be loadable if on the classpath). * 2. If config provider class loading fails (ConfigException wrapping ClassNotFoundException), - * strip config.providers entries and check for unresolved variable references. - * 3. If remaining properties contain unresolved ${provider:...} references (i.e. security - * properties depend on config providers), skip the check with a warning — the Connect - * worker will verify connectivity itself on startup. - * 4. Otherwise, retry the check without config.providers. + * strip config.providers entries from a copy and check for unresolved variable references + * in Kafka client connectivity properties (security.*, sasl.*, ssl.*). + * 3. If client connectivity properties contain unresolved ${provider:path:key} references, + * skip the check with a warning - the Connect worker will verify connectivity on startup. + * 4. Otherwise, retry the check with the stripped copy (without config.providers). + * + *

The caller's workerProps map is never mutated. */ static boolean checkKafkaReadyWithConfigProviderResilience( Map workerProps, @@ -221,9 +224,10 @@ static boolean checkKafkaReadyWithConfigProviderResilience( log.warn("Config provider class could not be loaded during kafka-ready check: {}", e.getMessage(), e); - stripConfigProviders(workerProps); + Map strippedProps = new HashMap<>(workerProps); + stripConfigProviders(strippedProps); - List unresolvedKeys = findUnresolvedConfigProviderVars(workerProps); + List unresolvedKeys = findUnresolvedConfigProviderVars(strippedProps); if (!unresolvedKeys.isEmpty()) { log.warn("Skipping kafka-ready check - the following properties contain unresolved " + "config provider variable references that cannot be resolved without the " @@ -233,7 +237,7 @@ static boolean checkKafkaReadyWithConfigProviderResilience( } log.warn("Retrying kafka-ready check without config provider properties."); - return checker.isReady(workerProps, minBrokerCount, timeoutMs); + return checker.isReady(strippedProps, minBrokerCount, timeoutMs); } } diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java index 552e44abbe..5390bbe738 100644 --- a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -310,9 +310,9 @@ public void resilience_classNotFound_literalProps_stripsAndRetries() { assertThat(result).isTrue(); assertThat(callCount.get()).isEqualTo(2); - // config.providers should have been stripped before retry - assertThat(props).doesNotContainKey("config.providers"); - assertThat(props).doesNotContainKey("config.providers.secretmanager.class"); + // caller's map should not be mutated + assertThat(props).containsKey("config.providers"); + assertThat(props).containsKey("config.providers.secretmanager.class"); assertThat(props).containsKey("bootstrap.servers"); } From 31c1bb6b9e0c36cf13db5383352c9d540308f40f Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Tue, 26 May 2026 12:12:52 +0530 Subject: [PATCH 12/14] Simplify: gate config.providers stripping behind env var flag Replace the complex try/catch/retry/unresolved-var logic with a simple approach: when CUB_KAFKA_READY_SKIP_CONFIG_PROVIDERS=true, strip all config.providers entries from the config before running the check. When not set (default), original behavior is preserved. This is safe because the kafka-ready preflight only needs Kafka client properties (bootstrap.servers, security.*, ssl.*, sasl.*) to verify broker connectivity. Config providers are only needed by the Connect worker at runtime. --- .../admin/utils/cli/KafkaReadyCommand.java | 168 ++-------- .../utils/cli/KafkaReadyCommandTest.java | 300 +----------------- 2 files changed, 35 insertions(+), 433 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 224fc8e726..38ae489a2a 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -23,17 +23,13 @@ import net.sourceforge.argparse4j.inf.Namespace; import org.apache.kafka.clients.CommonClientConfigs; -import org.apache.kafka.common.config.ConfigException; import org.apache.kafka.common.utils.Utils; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import java.util.HashMap; import java.util.Iterator; -import java.util.List; import java.util.Map; -import java.util.regex.Pattern; -import java.util.stream.Collectors; import io.confluent.admin.utils.ClusterStatus; @@ -59,23 +55,11 @@ public class KafkaReadyCommand { public static final String KAFKA_READY = "kafka-ready"; private static final String CONFIG_PROVIDERS_PREFIX = "config.providers"; - @FunctionalInterface - interface KafkaReadyChecker { - boolean isReady(Map config, int minBrokerCount, int timeoutMs); - } - - // Matches Kafka config provider variable syntax: ${provider-name:path[:key]} - // Requires at least one colon separator to distinguish from other ${...} placeholders. - private static final Pattern CONFIG_PROVIDER_VAR_PATTERN = - Pattern.compile("\\$\\{[^:}]+:[^}]+}"); - - // Property key prefixes that affect Kafka client connectivity. Only these are checked - // for unresolved config provider variable references when deciding whether to skip - // the kafka-ready check. Non-client keys (e.g. connector configs) are irrelevant - // since AdminClient ignores them. - private static final String[] KAFKA_CLIENT_KEY_PREFIXES = { - "security.", "sasl.", "ssl.", "bootstrap.servers" - }; + // When set to "true", config.providers entries are stripped from the worker config + // before running the kafka-ready check. This prevents ClassNotFoundException when + // config provider plugin JARs are on the worker's plugin.path but not on the + // CUB_CLASSPATH used by kafka-ready. Default: disabled (original behavior). + static final String SKIP_CONFIG_PROVIDERS_ENV = "CUB_KAFKA_READY_SKIP_CONFIG_PROVIDERS"; private static ArgumentParser createArgsParser() { ArgumentParser kafkaReady = ArgumentParsers @@ -154,7 +138,12 @@ public static void main(String[] args) { "Bootstrap servers should be provided through config or bootstrap_servers" ); } - success = checkKafkaReadyWithConfigProviderResilience( + + if (isConfigProviderResilienceEnabled()) { + workerProps = stripConfigProviders(workerProps); + } + + success = ClusterStatus.isKafkaReady( workerProps, res.getInt("min_expected_brokers"), res.getInt("timeout") @@ -182,137 +171,28 @@ public static void main(String[] args) { } /** - * Attempts the kafka-ready check, handling config provider class loading failures gracefully. - * - *

Strategy: - * 0. If no config.providers entries are present, run the check directly (fast path). - * 1. Try with the full config (config providers may be loadable if on the classpath). - * 2. If config provider class loading fails (ConfigException wrapping ClassNotFoundException), - * strip config.providers entries from a copy and check for unresolved variable references - * in Kafka client connectivity properties (security.*, sasl.*, ssl.*). - * 3. If client connectivity properties contain unresolved ${provider:path:key} references, - * skip the check with a warning - the Connect worker will verify connectivity on startup. - * 4. Otherwise, retry the check with the stripped copy (without config.providers). - * - *

The caller's workerProps map is never mutated. - */ - static boolean checkKafkaReadyWithConfigProviderResilience( - Map workerProps, - int minBrokerCount, - int timeoutMs - ) { - return checkKafkaReadyWithConfigProviderResilience( - workerProps, minBrokerCount, timeoutMs, ClusterStatus::isKafkaReady); - } - - static boolean checkKafkaReadyWithConfigProviderResilience( - Map workerProps, - int minBrokerCount, - int timeoutMs, - KafkaReadyChecker checker - ) { - if (!hasConfigProviders(workerProps)) { - return checker.isReady(workerProps, minBrokerCount, timeoutMs); - } - - try { - return checker.isReady(workerProps, minBrokerCount, timeoutMs); - } catch (ConfigException e) { - if (!isConfigProviderLoadFailure(e)) { - throw e; - } - log.warn("Config provider class could not be loaded during kafka-ready check: {}", - e.getMessage(), e); - - Map strippedProps = new HashMap<>(workerProps); - stripConfigProviders(strippedProps); - - List unresolvedKeys = findUnresolvedConfigProviderVars(strippedProps); - if (!unresolvedKeys.isEmpty()) { - log.warn("Skipping kafka-ready check - the following properties contain unresolved " - + "config provider variable references that cannot be resolved without the " - + "config provider plugin: {}. The Connect worker will verify broker connectivity " - + "on startup.", unresolvedKeys); - return true; - } - - log.warn("Retrying kafka-ready check without config provider properties."); - return checker.isReady(strippedProps, minBrokerCount, timeoutMs); - } - } - - /** - * Returns true if the properties map contains config.providers entries. - */ - static boolean hasConfigProviders(Map props) { - return props.containsKey(CONFIG_PROVIDERS_PREFIX); - } - - /** - * Returns true if the exception is a ConfigException specifically caused by a config - * provider class loading failure. Requires both: - * - The exception message references "config.providers" (to distinguish from SASL/SSL - * handler class loading failures which also throw ConfigException with CNFE) - * - Either a ClassNotFoundException/NoClassDefFoundError in the cause chain, or - * "Could not load" in the message (fallback for truncated cause chains) + * Returns a copy of the properties map with config.providers entries removed. + * If no config.providers entries are found, returns the original map. */ - static boolean isConfigProviderLoadFailure(ConfigException e) { - String msg = e.getMessage(); - boolean mentionsConfigProviders = msg != null && msg.contains("config.providers"); - if (!mentionsConfigProviders) { - return false; + static Map stripConfigProviders(Map props) { + if (!props.containsKey(CONFIG_PROVIDERS_PREFIX)) { + return props; } - Throwable cause = e.getCause(); - while (cause != null) { - if (cause instanceof ClassNotFoundException || cause instanceof NoClassDefFoundError) { - return true; - } - cause = cause.getCause(); - } - // Fallback: message-only check when cause chain is truncated - return msg.contains("Could not load"); - } - - /** - * Removes config.providers entries from the properties map. - */ - static void stripConfigProviders(Map props) { - List removed = new java.util.ArrayList<>(); - Iterator> it = props.entrySet().iterator(); + Map result = new HashMap<>(props); + Iterator it = result.keySet().iterator(); while (it.hasNext()) { - String key = it.next().getKey(); + String key = it.next(); if (key.equals(CONFIG_PROVIDERS_PREFIX) || key.startsWith(CONFIG_PROVIDERS_PREFIX + ".")) { - removed.add(key); + log.warn("Stripping property '{}' from kafka-ready config " + + "({}=true).", key, SKIP_CONFIG_PROVIDERS_ENV); it.remove(); } } - if (!removed.isEmpty()) { - log.warn("Removed {} config provider properties from kafka-ready config: {}", - removed.size(), removed); - } + return result; } - /** - * Returns Kafka client connectivity property keys whose values contain unresolved - * config provider variable references (e.g. ${secretmanager:path:key}). Only checks - * keys that affect broker connectivity (security.*, sasl.*, ssl.*, bootstrap.servers) - * since non-client keys are ignored by AdminClient and won't cause failures. - */ - static List findUnresolvedConfigProviderVars(Map props) { - return props.entrySet().stream() - .filter(e -> isKafkaClientKey(e.getKey())) - .filter(e -> CONFIG_PROVIDER_VAR_PATTERN.matcher(e.getValue()).find()) - .map(Map.Entry::getKey) - .collect(Collectors.toList()); - } - - private static boolean isKafkaClientKey(String key) { - for (String prefix : KAFKA_CLIENT_KEY_PREFIXES) { - if (key.startsWith(prefix)) { - return true; - } - } - return false; + static boolean isConfigProviderResilienceEnabled() { + return "true".equalsIgnoreCase(System.getenv(SKIP_CONFIG_PROVIDERS_ENV)); } } diff --git a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java index 5390bbe738..76e36631cf 100644 --- a/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java +++ b/utility-belt/src/test/java/io/confluent/admin/utils/cli/KafkaReadyCommandTest.java @@ -15,21 +15,15 @@ */ package io.confluent.admin.utils.cli; -import org.apache.kafka.common.config.ConfigException; import org.junit.Test; import java.util.HashMap; -import java.util.List; import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; public class KafkaReadyCommandTest { - // --- stripConfigProviders --- - @Test public void stripConfigProviders_removesAllConfigProviderEntries() { Map props = new HashMap<>(); @@ -41,22 +35,23 @@ public void stripConfigProviders_removesAllConfigProviderEntries() { "io.confluent.csid.config.provider.aws.SecretsManagerConfigProvider"); props.put("config.providers.secretmanager.param.aws.region", "us-east-1"); - KafkaReadyCommand.stripConfigProviders(props); + Map result = KafkaReadyCommand.stripConfigProviders(props); - assertThat(props).containsOnlyKeys( + assertThat(result).containsOnlyKeys( "bootstrap.servers", "security.protocol", "sasl.mechanism"); + // original map is not mutated + assertThat(props).hasSize(6); } @Test - public void stripConfigProviders_noopWhenNoConfigProviders() { + public void stripConfigProviders_returnsOriginalWhenNoProviders() { Map props = new HashMap<>(); props.put("bootstrap.servers", "localhost:9092"); props.put("security.protocol", "PLAINTEXT"); - KafkaReadyCommand.stripConfigProviders(props); + Map result = KafkaReadyCommand.stripConfigProviders(props); - assertThat(props).hasSize(2); - assertThat(props).containsEntry("bootstrap.servers", "localhost:9092"); + assertThat(result).isSameAs(props); } @Test @@ -69,9 +64,9 @@ public void stripConfigProviders_handlesMultipleProviders() { props.put("config.providers.vault.param.address", "https://vault:8200"); props.put("ssl.truststore.location", "/etc/ssl/truststore.jks"); - KafkaReadyCommand.stripConfigProviders(props); + Map result = KafkaReadyCommand.stripConfigProviders(props); - assertThat(props).containsOnlyKeys("bootstrap.servers", "ssl.truststore.location"); + assertThat(result).containsOnlyKeys("bootstrap.servers", "ssl.truststore.location"); } @Test @@ -82,281 +77,8 @@ public void stripConfigProviders_doesNotRemoveSimilarKeys() { props.put("config.providers.secretmanager.class", "com.example.Provider"); props.put("some.config.providers.unrelated", "value"); - KafkaReadyCommand.stripConfigProviders(props); - - assertThat(props).containsOnlyKeys("bootstrap.servers", "some.config.providers.unrelated"); - } - - // --- hasConfigProviders --- - - @Test - public void hasConfigProviders_trueWhenPresent() { - Map props = new HashMap<>(); - props.put("config.providers", "secretmanager"); - - assertThat(KafkaReadyCommand.hasConfigProviders(props)).isTrue(); - } - - @Test - public void hasConfigProviders_falseWhenAbsent() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - - assertThat(KafkaReadyCommand.hasConfigProviders(props)).isFalse(); - } - - // --- isConfigProviderLoadFailure --- - - @Test - public void isConfigProviderLoadFailure_trueForClassNotFoundException() { - ConfigException e = new ConfigException( - "Invalid value com.example.Provider for configuration " - + "config.providers.secretmanager.class: Could not load class"); - e.initCause(new ClassNotFoundException("com.example.Provider")); - - assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); - } - - @Test - public void isConfigProviderLoadFailure_trueForNoClassDefFoundError() { - ConfigException e = new ConfigException( - "Invalid value com.example.Provider for configuration " - + "config.providers.secretmanager.class: Could not load class"); - e.initCause(new NoClassDefFoundError("software/amazon/awssdk/core/SdkClient")); - - assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); - } - - @Test - public void isConfigProviderLoadFailure_trueForNestedClassNotFound() { - ClassNotFoundException cnfe = new ClassNotFoundException("com.example.Provider"); - RuntimeException wrapper = new RuntimeException("wrapper", cnfe); - ConfigException e = new ConfigException( - "config.providers.vault.class: Could not load"); - e.initCause(wrapper); - - assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); - } - - @Test - public void isConfigProviderLoadFailure_trueForMessageFallback() { - // No ClassNotFoundException in cause chain, but message matches - ConfigException e = new ConfigException( - "Invalid value com.example.Provider " - + "for configuration config.providers.secretmanager.class: " - + "Could not load config provider class or one of its dependencies"); - - assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isTrue(); - } - - @Test - public void isConfigProviderLoadFailure_falseForOtherConfigException() { - ConfigException e = new ConfigException("Unknown configuration key: bad.key"); - - assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isFalse(); - } - - @Test - public void isConfigProviderLoadFailure_falseForSaslClassNotFound() { - // CNFE for a SASL handler class should NOT be treated as a config provider failure - ConfigException e = new ConfigException( - "Invalid value com.example.SaslHandler for configuration " - + "sasl.client.callback.handler.class: Could not load class"); - e.initCause(new ClassNotFoundException("com.example.SaslHandler")); - - assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isFalse(); - } - - @Test - public void isConfigProviderLoadFailure_falseForSslClassNotFound() { - // CNFE for an SSL engine factory should NOT match - ConfigException e = new ConfigException( - "Invalid value com.example.SslFactory for configuration " - + "ssl.engine.factory.class: Could not load class"); - e.initCause(new ClassNotFoundException("com.example.SslFactory")); - - assertThat(KafkaReadyCommand.isConfigProviderLoadFailure(e)).isFalse(); - } - - // --- findUnresolvedConfigProviderVars --- - - @Test - public void findUnresolvedVars_detectsProviderReferences() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("ssl.keystore.password", "${secretmanager:my-secret:keystore-password}"); - props.put("sasl.jaas.config", "${vault:secret/kafka:jaas-config}"); - props.put("security.protocol", "SASL_SSL"); - - List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(props); - - assertThat(unresolved).hasSize(2); - assertThat(unresolved).contains("ssl.keystore.password", "sasl.jaas.config"); - } - - @Test - public void findUnresolvedVars_ignoresNonProviderPlaceholders() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - // Shell-style ${ENV_VAR} without colons should NOT be matched - props.put("some.prop", "${SOME_ENV_VAR}"); - // But provider-style ${provider:path} should be matched - props.put("ssl.keystore.password", "${secretmanager:my-secret:password}"); - - List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(props); - - assertThat(unresolved).hasSize(1); - assertThat(unresolved).contains("ssl.keystore.password"); - } - - @Test - public void findUnresolvedVars_emptyWhenAllLiteral() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("ssl.keystore.password", "my-literal-password"); - props.put("security.protocol", "SSL"); - - List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(props); - - assertThat(unresolved).isEmpty(); - } - - @Test - public void findUnresolvedVars_handlesEmptyMap() { - List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(new HashMap<>()); - - assertThat(unresolved).isEmpty(); - } - - @Test - public void findUnresolvedVars_ignoresNonClientKeys() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - // Non-client key with provider reference should be ignored - props.put("connector.password", "${secretmanager:secret:connector-pw}"); - props.put("group.id", "${secretmanager:secret:group}"); - // Client key with provider reference should be detected - props.put("ssl.keystore.password", "${secretmanager:secret:keystore-pw}"); - - List unresolved = KafkaReadyCommand.findUnresolvedConfigProviderVars(props); - - assertThat(unresolved).hasSize(1); - assertThat(unresolved).contains("ssl.keystore.password"); - } - - // --- checkKafkaReadyWithConfigProviderResilience orchestration --- - - @Test - public void resilience_noConfigProviders_callsCheckerOnce() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - AtomicInteger callCount = new AtomicInteger(0); - KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { - callCount.incrementAndGet(); - return true; - }; - - boolean result = KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( - props, 1, 5000, checker); - - assertThat(result).isTrue(); - assertThat(callCount.get()).isEqualTo(1); - } - - @Test - public void resilience_configProvidersLoadable_succeedsFirstTry() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("config.providers", "secretmanager"); - props.put("config.providers.secretmanager.class", "com.example.Provider"); - AtomicInteger callCount = new AtomicInteger(0); - KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { - callCount.incrementAndGet(); - return true; - }; - - boolean result = KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( - props, 1, 5000, checker); - - assertThat(result).isTrue(); - assertThat(callCount.get()).isEqualTo(1); - // config.providers still present since first try succeeded - assertThat(props).containsKey("config.providers"); - } - - @Test - public void resilience_classNotFound_literalProps_stripsAndRetries() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("security.protocol", "SASL_SSL"); - props.put("sasl.mechanism", "PLAIN"); - props.put("config.providers", "secretmanager"); - props.put("config.providers.secretmanager.class", "com.example.MissingProvider"); - AtomicInteger callCount = new AtomicInteger(0); - KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { - int call = callCount.incrementAndGet(); - if (call == 1) { - ConfigException ex = new ConfigException( - "Invalid value com.example.MissingProvider for configuration " - + "config.providers.secretmanager.class: Could not load"); - ex.initCause(new ClassNotFoundException("com.example.MissingProvider")); - throw ex; - } - return true; - }; - - boolean result = KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( - props, 1, 5000, checker); - - assertThat(result).isTrue(); - assertThat(callCount.get()).isEqualTo(2); - // caller's map should not be mutated - assertThat(props).containsKey("config.providers"); - assertThat(props).containsKey("config.providers.secretmanager.class"); - assertThat(props).containsKey("bootstrap.servers"); - } - - @Test - public void resilience_classNotFound_unresolvedVars_skipsWithWarning() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("ssl.keystore.password", "${secretmanager:secret:password}"); - props.put("config.providers", "secretmanager"); - props.put("config.providers.secretmanager.class", "com.example.MissingProvider"); - AtomicInteger callCount = new AtomicInteger(0); - KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { - callCount.incrementAndGet(); - ConfigException ex = new ConfigException( - "Invalid value com.example.MissingProvider for configuration " - + "config.providers.secretmanager.class: Could not load"); - ex.initCause(new ClassNotFoundException("com.example.MissingProvider")); - throw ex; - }; - - boolean result = KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( - props, 1, 5000, checker); - - // Should skip (return true) without retrying - assertThat(result).isTrue(); - assertThat(callCount.get()).isEqualTo(1); - } - - @Test - public void resilience_nonProviderConfigException_rethrows() { - Map props = new HashMap<>(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("config.providers", "secretmanager"); - props.put("config.providers.secretmanager.class", "com.example.Provider"); - KafkaReadyCommand.KafkaReadyChecker checker = (config, brokers, timeout) -> { - throw new ConfigException("Unknown configuration key: bad.key"); - }; + Map result = KafkaReadyCommand.stripConfigProviders(props); - try { - KafkaReadyCommand.checkKafkaReadyWithConfigProviderResilience( - props, 1, 5000, checker); - fail("Expected ConfigException to be rethrown"); - } catch (ConfigException e) { - assertThat(e.getMessage()).contains("bad.key"); - } + assertThat(result).containsOnlyKeys("bootstrap.servers", "some.config.providers.unrelated"); } } From 7dbecff979b379fdb8463b0e15feb8a6e061824d Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Tue, 26 May 2026 12:33:02 +0530 Subject: [PATCH 13/14] Address Copilot review: fix prefix-only keys, rename method, single log line - Check for both config.providers and config.providers.* keys before returning early, so prefix-only entries are also stripped - Rename isConfigProviderResilienceEnabled to isSkipConfigProvidersEnabled to match the env var name - Replace per-key WARN with single INFO summarizing count of stripped keys --- .../admin/utils/cli/KafkaReadyCommand.java | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 38ae489a2a..16db715abb 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -139,7 +139,7 @@ public static void main(String[] args) { ); } - if (isConfigProviderResilienceEnabled()) { + if (isSkipConfigProvidersEnabled()) { workerProps = stripConfigProviders(workerProps); } @@ -172,27 +172,36 @@ public static void main(String[] args) { /** * Returns a copy of the properties map with config.providers entries removed. - * If no config.providers entries are found, returns the original map. + * If no config.providers entries are found, returns the original map unmodified. */ static Map stripConfigProviders(Map props) { - if (!props.containsKey(CONFIG_PROVIDERS_PREFIX)) { + boolean hasProviderKeys = false; + for (String key : props.keySet()) { + if (key.equals(CONFIG_PROVIDERS_PREFIX) || key.startsWith(CONFIG_PROVIDERS_PREFIX + ".")) { + hasProviderKeys = true; + break; + } + } + if (!hasProviderKeys) { return props; } Map result = new HashMap<>(props); + int count = 0; Iterator it = result.keySet().iterator(); while (it.hasNext()) { String key = it.next(); if (key.equals(CONFIG_PROVIDERS_PREFIX) || key.startsWith(CONFIG_PROVIDERS_PREFIX + ".")) { - log.warn("Stripping property '{}' from kafka-ready config " - + "({}=true).", key, SKIP_CONFIG_PROVIDERS_ENV); it.remove(); + count++; } } + log.info("Stripped {} config.providers properties from kafka-ready config ({}=true).", + count, SKIP_CONFIG_PROVIDERS_ENV); return result; } - static boolean isConfigProviderResilienceEnabled() { + static boolean isSkipConfigProvidersEnabled() { return "true".equalsIgnoreCase(System.getenv(SKIP_CONFIG_PROVIDERS_ENV)); } } From 342160e2dbf6438142ca43451823877c1a40a44f Mon Sep 17 00:00:00 2001 From: Kunal Sevkani Date: Wed, 27 May 2026 17:13:20 +0530 Subject: [PATCH 14/14] Simplify: use startsWith for config.providers key matching --- .../java/io/confluent/admin/utils/cli/KafkaReadyCommand.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java index 16db715abb..56bf2aebc0 100644 --- a/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java +++ b/utility-belt/src/main/java/io/confluent/admin/utils/cli/KafkaReadyCommand.java @@ -177,7 +177,7 @@ public static void main(String[] args) { static Map stripConfigProviders(Map props) { boolean hasProviderKeys = false; for (String key : props.keySet()) { - if (key.equals(CONFIG_PROVIDERS_PREFIX) || key.startsWith(CONFIG_PROVIDERS_PREFIX + ".")) { + if (key.startsWith(CONFIG_PROVIDERS_PREFIX)) { hasProviderKeys = true; break; } @@ -191,7 +191,7 @@ static Map stripConfigProviders(Map props) { Iterator it = result.keySet().iterator(); while (it.hasNext()) { String key = it.next(); - if (key.equals(CONFIG_PROVIDERS_PREFIX) || key.startsWith(CONFIG_PROVIDERS_PREFIX + ".")) { + if (key.startsWith(CONFIG_PROVIDERS_PREFIX)) { it.remove(); count++; }