Skip to content

MINOR: AbstractConfig cleanup #15597

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

Merged
gharris1727 merged 8 commits into from
Mar 28, 2024
Merged

MINOR: AbstractConfig cleanup #15597

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
a2077b0
MINOR: AbstractConfig cleanup
gharris1727 Mar 25, 2024
a12c19e
fixup: add tests for including invalid classes, lists of classes with…
gharris1727 Mar 26, 2024
d6da916
Use assertThrows in tests
gharris1727 Mar 26, 2024
568050b
fixup: remove test util fn, whitespace, refactor Worker prefix test
gharris1727 Mar 27, 2024
5efcf02
fixup: Override WorkerConfig.originals to hide already-processed conf…
gharris1727 Mar 27, 2024
fe0d42e
fixup: revert mirror changes
gharris1727 Mar 27, 2024
997f026
fixup: failing test due to Worker change
gharris1727 Mar 28, 2024
352d7e7
fixup: broken PartitionStateMachineTest mocks
gharris1727 Mar 28, 2024
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
51 changes: 39 additions & 12 deletions clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
Expand All @@ -33,6 +34,8 @@
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Predicate;
import java.util.stream.Collectors;

/**
* A convenient base class for configurations to extend.
Expand All @@ -58,6 +61,8 @@ public class AbstractConfig {

private final ConfigDef definition;

public static final String AUTOMATIC_CONFIG_PROVIDERS_PROPERTY = "org.apache.kafka.automatic.config.providers";
Comment thread
gharris1727 marked this conversation as resolved.

public static final String CONFIG_PROVIDERS_CONFIG = "config.providers";

private static final String CONFIG_PROVIDERS_PARAM = ".param.";
Expand Down Expand Up @@ -101,14 +106,11 @@ public class AbstractConfig {
* the constructor to resolve any variables in {@code originals}; may be null or empty
* @param doLog whether the configurations should be logged
*/
@SuppressWarnings({"unchecked", "this-escape"})
@SuppressWarnings({"this-escape"})
public AbstractConfig(ConfigDef definition, Map<?, ?> originals, Map<String, ?> configProviderProps, boolean doLog) {
/* check that all the keys are really strings */
for (Map.Entry<?, ?> entry : originals.entrySet())
if (!(entry.getKey() instanceof String))
throw new ConfigException(entry.getKey().toString(), entry.getValue(), "Key must be a string.");
Map<String, Object> originalMap = Utils.castToStringObjectMap(originals);

this.originals = resolveConfigVariables(configProviderProps, (Map<String, Object>) originals);
this.originals = resolveConfigVariables(configProviderProps, originalMap);
this.values = definition.parse(this.originals);
Map<String, Object> configUpdates = postProcessParsedConfig(Collections.unmodifiableMap(this.values));
for (Map.Entry<String, Object> update : configUpdates.entrySet()) {
Expand Down Expand Up @@ -521,6 +523,7 @@ private Map<String, String> extractPotentialVariables(Map<?, ?> configMap) {
private Map<String, ?> resolveConfigVariables(Map<String, ?> configProviderProps, Map<String, Object> originals) {
Map<String, String> providerConfigString;
Map<String, ?> configProperties;
Predicate<String> classNameFilter;
Map<String, Object> resolvedOriginals = new HashMap<>();
// As variable configs are strings, parse the originals and obtain the potential variable configs.
Map<String, String> indirectVariables = extractPotentialVariables(originals);
Expand All @@ -529,11 +532,13 @@ private Map<String, String> extractPotentialVariables(Map<?, ?> configMap) {
if (configProviderProps == null || configProviderProps.isEmpty()) {
providerConfigString = indirectVariables;
configProperties = originals;
classNameFilter = automaticConfigProvidersFilter();
} else {
providerConfigString = extractPotentialVariables(configProviderProps);
configProperties = configProviderProps;
classNameFilter = ignored -> true;
}
Map<String, ConfigProvider> providers = instantiateConfigProviders(providerConfigString, configProperties);
Map<String, ConfigProvider> providers = instantiateConfigProviders(providerConfigString, configProperties, classNameFilter);

if (!providers.isEmpty()) {
ConfigTransformer configTransformer = new ConfigTransformer(providers);
Expand All @@ -547,6 +552,17 @@ private Map<String, String> extractPotentialVariables(Map<?, ?> configMap) {
return new ResolvingMap<>(resolvedOriginals, originals);
}

private Predicate<String> automaticConfigProvidersFilter() {
String systemProperty = System.getProperty(AUTOMATIC_CONFIG_PROVIDERS_PROPERTY);
if (systemProperty == null) {
return ignored -> true;
} else {
return Arrays.stream(systemProperty.split(","))
Comment thread
C0urante marked this conversation as resolved.
.map(String::trim)
.collect(Collectors.toSet())::contains;
}
}

private Map<String, Object> configProviderProperties(String configProviderPrefix, Map<String, ?> providerConfigProperties) {
Map<String, Object> result = new HashMap<>();
for (Map.Entry<String, ?> entry : providerConfigProperties.entrySet()) {
Expand All @@ -567,9 +583,14 @@ private Map<String, Object> configProviderProperties(String configProviderPrefix
*
* @param indirectConfigs The map of potential variable configs
* @param providerConfigProperties The map of config provider configs
* @return map map of config provider name and its instance.
* @param classNameFilter Filter for config provider class names
* @return map of config provider name and its instance.
*/
private Map<String, ConfigProvider> instantiateConfigProviders(Map<String, String> indirectConfigs, Map<String, ?> providerConfigProperties) {
private Map<String, ConfigProvider> instantiateConfigProviders(
Map<String, String> indirectConfigs,
Map<String, ?> providerConfigProperties,
Predicate<String> classNameFilter
) {
final String configProviders = indirectConfigs.get(CONFIG_PROVIDERS_CONFIG);

if (configProviders == null || configProviders.isEmpty()) {
Expand All @@ -580,9 +601,15 @@ private Map<String, ConfigProvider> instantiateConfigProviders(Map<String, Strin

for (String provider : configProviders.split(",")) {
String providerClass = providerClassProperty(provider);
if (indirectConfigs.containsKey(providerClass))
providerMap.put(provider, indirectConfigs.get(providerClass));

if (indirectConfigs.containsKey(providerClass)) {
String providerClassName = indirectConfigs.get(providerClass);
if (classNameFilter.test(providerClassName)) {
Comment thread
gharris1727 marked this conversation as resolved.
providerMap.put(provider, providerClassName);
} else {
throw new ConfigException(providerClassName + " is not allowed. Update System property '"
+ AUTOMATIC_CONFIG_PROVIDERS_PROPERTY + "' to allow " + providerClassName);
}
}
}
// Instantiate Config Providers
Map<String, ConfigProvider> configProviderInstances = new HashMap<>();
Expand Down
18 changes: 14 additions & 4 deletions clients/src/main/java/org/apache/kafka/common/utils/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1502,13 +1502,23 @@ public static <K, V> Map<K, V> filterMap(final Map<K, V> map, final Predicate<En
* @return a map including all elements in properties
*/
public static Map<String, Object> propsToMap(Properties properties) {
Map<String, Object> map = new HashMap<>(properties.size());
for (Map.Entry<Object, Object> entry : properties.entrySet()) {
return castToStringObjectMap(properties);
}

/**
* Cast a map with arbitrary type keys to be keyed on String.
* @param inputMap A map with unknown type keys
* @return A map with the same contents as the input map, but with String keys
* @throws ConfigException if any key is not a String
*/
public static Map<String, Object> castToStringObjectMap(Map<?, ?> inputMap) {
Comment thread
gharris1727 marked this conversation as resolved.
Map<String, Object> map = new HashMap<>(inputMap.size());
for (Map.Entry<?, ?> entry : inputMap.entrySet()) {
if (entry.getKey() instanceof String) {
String k = (String) entry.getKey();
map.put(k, properties.get(k));
map.put(k, entry.getValue());
} else {
throw new ConfigException(entry.getKey().toString(), entry.getValue(), "Key must be a string.");
throw new ConfigException(String.valueOf(entry.getKey()), entry.getValue(), "Key must be a string.");
}
}
return map;
Expand Down
122 changes: 94 additions & 28 deletions clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,18 @@
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.config.provider.FileConfigProvider;
import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.metrics.FakeMetricsReporter;
import org.apache.kafka.common.metrics.JmxReporter;
import org.apache.kafka.common.metrics.MetricsReporter;
import org.apache.kafka.common.security.TestSecurityConfig;
import org.apache.kafka.common.config.provider.MockVaultConfigProvider;
import org.apache.kafka.common.config.provider.MockFileConfigProvider;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.test.MockConsumerInterceptor;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.util.Arrays;
Expand All @@ -46,6 +50,23 @@

public class AbstractConfigTest {

private String propertyValue;

@BeforeEach
public void setup() {
propertyValue = System.getProperty(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY);
System.clearProperty(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY);
}

@AfterEach
public void teardown() {
if (propertyValue != null) {
System.setProperty(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY, propertyValue);
} else {
System.clearProperty(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY);
}
}

@Test
public void testConfiguredInstances() {
testValidInputs(" ");
Expand Down Expand Up @@ -254,12 +275,7 @@ private void testInvalidInputs(String configValue) {
Properties props = new Properties();
props.put(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, configValue);
TestConfig config = new TestConfig(props);
try {
config.getConfiguredInstances(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class);
fail("Expected a config exception due to invalid props :" + props);
} catch (KafkaException e) {
// this is good
}
assertThrows(KafkaException.class, () -> config.getConfiguredInstances(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class));
}

@Test
Expand Down Expand Up @@ -349,16 +365,6 @@ protected Class<?> findClass(String name) throws ClassNotFoundException {
}
}

@SuppressWarnings("unchecked")
public Map<String, ?> convertPropertiesToMap(Map<?, ?> props) {
for (Map.Entry<?, ?> entry : props.entrySet()) {
if (!(entry.getKey() instanceof String))
throw new ConfigException(entry.getKey().toString(), entry.getValue(),
"Key must be a string.");
}
return (Map<String, ?>) props;
}

@Test
public void testOriginalWithOverrides() {
Properties props = new Properties();
Expand Down Expand Up @@ -389,6 +395,43 @@ public void testOriginalsWithConfigProvidersProps() {
MockFileConfigProvider.assertClosed(id);
}

@Test
public void testOriginalsWithConfigProvidersPropsExcluded() {
System.setProperty(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY, MockVaultConfigProvider.class.getName() + " , " + FileConfigProvider.class.getName());
Properties props = new Properties();

// Test Case: Config provider that is not an allowed class
props.put("config.providers", "file");
props.put("config.providers.file.class", MockFileConfigProvider.class.getName());
String id = UUID.randomUUID().toString();
props.put("config.providers.file.param.testId", id);
props.put("prefix.ssl.truststore.location.number", 5);
props.put("sasl.kerberos.service.name", "service name");
props.put("sasl.kerberos.key", "${file:/usr/kerberos:key}");
props.put("sasl.kerberos.password", "${file:/usr/kerberos:password}");
assertThrows(ConfigException.class, () -> new TestIndirectConfigResolution(props, Collections.emptyMap()));
}

@Test
public void testOriginalsWithConfigProvidersPropsIncluded() {
System.setProperty(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY, MockFileConfigProvider.class.getName() + " , " + FileConfigProvider.class.getName());
Properties props = new Properties();

// Test Case: Config provider that is an allowed class
props.put("config.providers", "file");
props.put("config.providers.file.class", MockFileConfigProvider.class.getName());
String id = UUID.randomUUID().toString();
props.put("config.providers.file.param.testId", id);
props.put("prefix.ssl.truststore.location.number", 5);
props.put("sasl.kerberos.service.name", "service name");
props.put("sasl.kerberos.key", "${file:/usr/kerberos:key}");
props.put("sasl.kerberos.password", "${file:/usr/kerberos:password}");
TestIndirectConfigResolution config = new TestIndirectConfigResolution(props, Collections.emptyMap());
assertEquals("testKey", config.originals().get("sasl.kerberos.key"));
assertEquals("randomPassword", config.originals().get("sasl.kerberos.password"));
MockFileConfigProvider.assertClosed(id);
}

@Test
public void testConfigProvidersPropsAsParam() {
// Test Case: Valid Test Case for ConfigProviders as a separate variable
Expand All @@ -400,7 +443,7 @@ public void testConfigProvidersPropsAsParam() {
Properties props = new Properties();
props.put("sasl.kerberos.key", "${file:/usr/kerberos:key}");
props.put("sasl.kerberos.password", "${file:/usr/kerberos:password}");
TestIndirectConfigResolution config = new TestIndirectConfigResolution(props, convertPropertiesToMap(providers));
TestIndirectConfigResolution config = new TestIndirectConfigResolution(props, Utils.castToStringObjectMap(providers));
assertEquals("testKey", config.originals().get("sasl.kerberos.key"));
assertEquals("randomPassword", config.originals().get("sasl.kerberos.password"));
MockFileConfigProvider.assertClosed(id);
Expand All @@ -417,7 +460,7 @@ public void testImmutableOriginalsWithConfigProvidersProps() {
Properties props = new Properties();
props.put("sasl.kerberos.key", "${file:/usr/kerberos:key}");
Map<?, ?> immutableMap = Collections.unmodifiableMap(props);
Map<String, ?> provMap = convertPropertiesToMap(providers);
Map<String, ?> provMap = Utils.castToStringObjectMap(providers);
TestIndirectConfigResolution config = new TestIndirectConfigResolution(immutableMap, provMap);
assertEquals("testKey", config.originals().get("sasl.kerberos.key"));
MockFileConfigProvider.assertClosed(id);
Expand All @@ -437,7 +480,7 @@ public void testAutoConfigResolutionWithMultipleConfigProviders() {
props.put("sasl.kerberos.password", "${file:/usr/kerberos:password}");
props.put("sasl.truststore.key", "${vault:/usr/truststore:truststoreKey}");
props.put("sasl.truststore.password", "${vault:/usr/truststore:truststorePassword}");
TestIndirectConfigResolution config = new TestIndirectConfigResolution(props, convertPropertiesToMap(providers));
TestIndirectConfigResolution config = new TestIndirectConfigResolution(props, Utils.castToStringObjectMap(providers));
assertEquals("testKey", config.originals().get("sasl.kerberos.key"));
assertEquals("randomPassword", config.originals().get("sasl.kerberos.password"));
assertEquals("testTruststoreKey", config.originals().get("sasl.truststore.key"));
Expand All @@ -453,12 +496,33 @@ public void testAutoConfigResolutionWithInvalidConfigProviderClass() {
props.put("config.providers.file.class",
"org.apache.kafka.common.config.provider.InvalidConfigProvider");
props.put("testKey", "${test:/foo/bar/testpath:testKey}");
try {
new TestIndirectConfigResolution(props);
fail("Expected a config exception due to invalid props :" + props);
} catch (KafkaException e) {
// this is good
}
assertThrows(KafkaException.class, () -> new TestIndirectConfigResolution(props));
}

@Test
public void testAutoConfigResolutionWithInvalidConfigProviderClassExcluded() {
String invalidConfigProvider = "org.apache.kafka.common.config.provider.InvalidConfigProvider";
System.setProperty(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY, "");
Comment thread
gharris1727 marked this conversation as resolved.
// Test Case: Any config provider specified while the system property is empty
Properties props = new Properties();
props.put("config.providers", "file");
props.put("config.providers.file.class", invalidConfigProvider);
props.put("testKey", "${test:/foo/bar/testpath:testKey}");
KafkaException e = assertThrows(KafkaException.class, () -> new TestIndirectConfigResolution(props, Collections.emptyMap()));
assertTrue(e.getMessage().contains(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY));
}

@Test
public void testAutoConfigResolutionWithInvalidConfigProviderClassIncluded() {
String invalidConfigProvider = "org.apache.kafka.common.config.provider.InvalidConfigProvider";
System.setProperty(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY, invalidConfigProvider);
// Test Case: Invalid config provider specified, but is also included in the system property
Properties props = new Properties();
props.put("config.providers", "file");
props.put("config.providers.file.class", invalidConfigProvider);
props.put("testKey", "${test:/foo/bar/testpath:testKey}");
KafkaException e = assertThrows(KafkaException.class, () -> new TestIndirectConfigResolution(props, Collections.emptyMap()));
assertFalse(e.getMessage().contains(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY));
}

@Test
Expand Down Expand Up @@ -496,13 +560,15 @@ public void testAutoConfigResolutionWithDuplicateConfigProvider() {
props.put("config.providers", "file");
props.put("config.providers.file.class", MockVaultConfigProvider.class.getName());

TestIndirectConfigResolution config = new TestIndirectConfigResolution(props, convertPropertiesToMap(providers));
TestIndirectConfigResolution config = new TestIndirectConfigResolution(props, Utils.castToStringObjectMap(providers));
assertEquals("${file:/usr/kerberos:key}", config.originals().get("sasl.kerberos.key"));
}

@Test
public void testConfigProviderConfigurationWithConfigParams() {
// Test Case: Valid Test Case With Multiple ConfigProviders as a separate variable
// should have no effect
System.setProperty(AbstractConfig.AUTOMATIC_CONFIG_PROVIDERS_PROPERTY, MockFileConfigProvider.class.getName());
// Test Case: Specify a config provider not allowed, but passed via the trusted providers argument
Properties providers = new Properties();
providers.put("config.providers", "vault");
providers.put("config.providers.vault.class", MockVaultConfigProvider.class.getName());
Expand All @@ -512,7 +578,7 @@ public void testConfigProviderConfigurationWithConfigParams() {
props.put("sasl.truststore.key", "${vault:/usr/truststore:truststoreKey}");
props.put("sasl.truststore.password", "${vault:/usr/truststore:truststorePassword}");
props.put("sasl.truststore.location", "${vault:/usr/truststore:truststoreLocation}");
TestIndirectConfigResolution config = new TestIndirectConfigResolution(props, convertPropertiesToMap(providers));
TestIndirectConfigResolution config = new TestIndirectConfigResolution(props, Utils.castToStringObjectMap(providers));
assertEquals("/usr/vault", config.originals().get("sasl.truststore.location"));
}

Expand Down
2 changes: 1 addition & 1 deletion connect/mirror-client/src/main/java/org/apache/kafka/connect/mirror/MirrorClientConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ public class MirrorClientConfig extends AbstractConfig {
public static final String PRODUCER_CLIENT_PREFIX = "producer.";

MirrorClientConfig(Map<?, ?> props) {
super(CONFIG_DEF, props, true);
super(CONFIG_DEF, props, Utils.castToStringObjectMap(props), true);
}

public ReplicationPolicy replicationPolicy() {
Expand Down
Loading