You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by rs...@apache.org on 2019/06/03 11:51:15 UTC
[kafka] branch 2.3 updated (a2daa49 -> dd763fb)
This is an automated email from the ASF dual-hosted git repository.
rsivaram pushed a change to branch 2.3
in repository https://gitbox.apache.org/repos/asf/kafka.git.
from a2daa49 KAFKA-8463: Fix redundant reassignment of tasks when leader worker leaves (#6859)
new 2e1137a MINOR: Reordering the props modification with configs construction
new dd763fb KAFKA-8425: Fix for correctly handling immutable maps (KIP-421 bug) (#6795)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/kafka/common/config/AbstractConfig.java | 37 +++++++++++++++++++---
.../kafka/common/config/AbstractConfigTest.java | 14 ++++++++
gradle/spotbugs-exclude.xml | 7 ++++
.../apache/kafka/streams/StreamsConfigTest.java | 15 +++++----
4 files changed, 63 insertions(+), 10 deletions(-)
[kafka] 01/02: MINOR: Reordering the props modification with
configs construction
Posted by rs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rsivaram pushed a commit to branch 2.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 2e1137a0341e1eb9d900cffdd50bcdd2cc295172
Author: Guozhang Wang <wa...@gmail.com>
AuthorDate: Fri May 31 17:02:40 2019 -0700
MINOR: Reordering the props modification with configs construction
Reviewers: Randall Hauch <rh...@gmail.com>, Matthias J. Sax <mj...@apache.org>, Bill Bejeck <bb...@gmail.com>
---
.../java/org/apache/kafka/streams/StreamsConfigTest.java | 15 +++++++++------
1 file changed, 9 insertions(+), 6 deletions(-)
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
index 5f053bc..aa3860e 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
@@ -152,6 +152,7 @@ public class StreamsConfigTest {
public void testGetMainConsumerConfigsWithMainConsumerOverridenPrefix() {
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "5");
props.put(StreamsConfig.mainConsumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "50");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> returnedProps = streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
assertEquals("50", returnedProps.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
}
@@ -215,24 +216,24 @@ public class StreamsConfigTest {
@Test
public void shouldSupportPrefixedPropertiesThatAreNotPartOfConsumerConfig() {
- final StreamsConfig streamsConfig = new StreamsConfig(props);
props.put(consumerPrefix("interceptor.statsd.host"), "host");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> consumerConfigs = streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
assertEquals("host", consumerConfigs.get("interceptor.statsd.host"));
}
@Test
public void shouldSupportPrefixedPropertiesThatAreNotPartOfRestoreConsumerConfig() {
- final StreamsConfig streamsConfig = new StreamsConfig(props);
props.put(consumerPrefix("interceptor.statsd.host"), "host");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> consumerConfigs = streamsConfig.getRestoreConsumerConfigs(clientId);
assertEquals("host", consumerConfigs.get("interceptor.statsd.host"));
}
@Test
public void shouldSupportPrefixedPropertiesThatAreNotPartOfProducerConfig() {
- final StreamsConfig streamsConfig = new StreamsConfig(props);
props.put(producerPrefix("interceptor.statsd.host"), "host");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> producerConfigs = streamsConfig.getProducerConfigs(clientId);
assertEquals("host", producerConfigs.get("interceptor.statsd.host"));
}
@@ -277,8 +278,8 @@ public class StreamsConfigTest {
@Test
public void shouldForwardCustomConfigsWithNoPrefixToAllClients() {
- final StreamsConfig streamsConfig = new StreamsConfig(props);
props.put("custom.property.host", "host");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> consumerConfigs = streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
final Map<String, Object> restoreConsumerConfigs = streamsConfig.getRestoreConsumerConfigs(clientId);
final Map<String, Object> producerConfigs = streamsConfig.getProducerConfigs(clientId);
@@ -291,11 +292,11 @@ public class StreamsConfigTest {
@Test
public void shouldOverrideNonPrefixedCustomConfigsWithPrefixedConfigs() {
- final StreamsConfig streamsConfig = new StreamsConfig(props);
props.put("custom.property.host", "host0");
props.put(consumerPrefix("custom.property.host"), "host1");
props.put(producerPrefix("custom.property.host"), "host2");
props.put(adminClientPrefix("custom.property.host"), "host3");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> consumerConfigs = streamsConfig.getMainConsumerConfigs(groupId, clientId, threadIdx);
final Map<String, Object> restoreConsumerConfigs = streamsConfig.getRestoreConsumerConfigs(clientId);
final Map<String, Object> producerConfigs = streamsConfig.getProducerConfigs(clientId);
@@ -374,6 +375,7 @@ public class StreamsConfigTest {
public void testGetRestoreConsumerConfigsWithRestoreConsumerOverridenPrefix() {
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "5");
props.put(StreamsConfig.restoreConsumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "50");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> returnedProps = streamsConfig.getRestoreConsumerConfigs(clientId);
assertEquals("50", returnedProps.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
}
@@ -395,8 +397,8 @@ public class StreamsConfigTest {
@Test
public void shouldSupportPrefixedPropertiesThatAreNotPartOfGlobalConsumerConfig() {
- final StreamsConfig streamsConfig = new StreamsConfig(props);
props.put(consumerPrefix("interceptor.statsd.host"), "host");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> consumerConfigs = streamsConfig.getGlobalConsumerConfigs(clientId);
assertEquals("host", consumerConfigs.get("interceptor.statsd.host"));
}
@@ -421,6 +423,7 @@ public class StreamsConfigTest {
public void testGetGlobalConsumerConfigsWithGlobalConsumerOverridenPrefix() {
props.put(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "5");
props.put(StreamsConfig.globalConsumerPrefix(ConsumerConfig.MAX_POLL_RECORDS_CONFIG), "50");
+ final StreamsConfig streamsConfig = new StreamsConfig(props);
final Map<String, Object> returnedProps = streamsConfig.getGlobalConsumerConfigs(clientId);
assertEquals("50", returnedProps.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
}
[kafka] 02/02: KAFKA-8425: Fix for correctly handling immutable
maps (KIP-421 bug) (#6795)
Posted by rs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rsivaram pushed a commit to branch 2.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit dd763fbd6a9b1d712b1c3dbd1fbb497943cd28a6
Author: tadsul <43...@users.noreply.github.com>
AuthorDate: Mon Jun 3 04:43:11 2019 -0700
KAFKA-8425: Fix for correctly handling immutable maps (KIP-421 bug) (#6795)
Since the originals map passed to AbstractConfig constructor may be immutable, avoid updating this map while resolving indirect config variables. Instead a new ResolvingMap instance is now used to store resolved configs.
Reviewers: Randall Hauch <rh...@gmail.com>, Boyang Chen <bc...@outlook.com>, Rajini Sivaram <ra...@googlemail.com>
---
.../apache/kafka/common/config/AbstractConfig.java | 37 +++++++++++++++++++---
.../kafka/common/config/AbstractConfigTest.java | 14 ++++++++
gradle/spotbugs-exclude.xml | 7 ++++
3 files changed, 54 insertions(+), 4 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
index 555b634..13865d0 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java
@@ -103,7 +103,6 @@ public class AbstractConfig {
throw new ConfigException(entry.getKey().toString(), entry.getValue(), "Key must be a string.");
this.originals = resolveConfigVariables(configProviderProps, (Map<String, Object>) originals);
-
this.values = definition.parse(this.originals);
Map<String, Object> configUpdates = postProcessParsedConfig(Collections.unmodifiableMap(this.values));
for (Map.Entry<String, Object> update : configUpdates.entrySet()) {
@@ -459,10 +458,11 @@ public class AbstractConfig {
private Map<String, ?> resolveConfigVariables(Map<String, ?> configProviderProps, Map<String, Object> originals) {
Map<String, String> providerConfigString;
Map<String, ?> configProperties;
-
+ 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);
+ resolvedOriginals.putAll(originals);
if (configProviderProps == null || configProviderProps.isEmpty()) {
providerConfigString = indirectVariables;
configProperties = originals;
@@ -475,10 +475,12 @@ public class AbstractConfig {
if (!providers.isEmpty()) {
ConfigTransformer configTransformer = new ConfigTransformer(providers);
ConfigTransformerResult result = configTransformer.transform(indirectVariables);
- originals.putAll(result.data());
+ if (!result.data().isEmpty()) {
+ resolvedOriginals.putAll(result.data());
+ }
}
- return originals;
+ return new ResolvingMap<>(resolvedOriginals, originals);
}
private Map<String, Object> configProviderProperties(String configProviderPrefix, Map<String, ?> providerConfigProperties) {
@@ -585,4 +587,31 @@ public class AbstractConfig {
return super.get(key);
}
}
+
+ /**
+ * ResolvingMap keeps a track of the original map instance and the resolved configs.
+ * The originals are tracked in a separate nested map and may be a `RecordingMap`; thus
+ * any access to a value for a key needs to be recorded on the originals map.
+ * The resolved configs are kept in the inherited map and are therefore mutable, though any
+ * mutations are not applied to the originals.
+ */
+ private static class ResolvingMap<V> extends HashMap<String, V> {
+
+ private final Map<String, ?> originals;
+
+ ResolvingMap(Map<String, ? extends V> resolved, Map<String, ?> originals) {
+ super(resolved);
+ this.originals = Collections.unmodifiableMap(originals);
+ }
+
+ @Override
+ public V get(Object key) {
+ if (key instanceof String && originals.containsKey(key)) {
+ // Intentionally ignore the result; call just to mark the original entry as used
+ originals.get(key);
+ }
+ // But always use the resolved entry
+ return super.get(key);
+ }
+ }
}
diff --git a/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java b/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java
index a007fd3..33aaac5 100644
--- a/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java
@@ -364,6 +364,20 @@ public class AbstractConfigTest {
}
@Test
+ public void testImmutableOriginalsWithConfigProvidersProps() {
+ // Test Case: Valid Test Case for ConfigProviders as a separate variable
+ Properties providers = new Properties();
+ providers.put("config.providers", "file");
+ providers.put("config.providers.file.class", "org.apache.kafka.common.config.provider.MockFileConfigProvider");
+ Properties props = new Properties();
+ props.put("sasl.kerberos.key", "${file:/usr/kerberos:key}");
+ Map<?, ?> immutableMap = Collections.unmodifiableMap(props);
+ Map<String, ?> provMap = convertPropertiesToMap(providers);
+ TestIndirectConfigResolution config = new TestIndirectConfigResolution(immutableMap, provMap);
+ assertEquals(config.originals().get("sasl.kerberos.key"), "testKey");
+ }
+
+ @Test
public void testAutoConfigResolutionWithMultipleConfigProviders() {
// Test Case: Valid Test Case With Multiple ConfigProviders as a separate variable
Properties providers = new Properties();
diff --git a/gradle/spotbugs-exclude.xml b/gradle/spotbugs-exclude.xml
index 721b05e..60e2eb2 100644
--- a/gradle/spotbugs-exclude.xml
+++ b/gradle/spotbugs-exclude.xml
@@ -332,4 +332,11 @@ For a detailed description of spotbugs bug categories, see https://spotbugs.read
<!-- END Suppress warnings for unused members that are undetectably used by Jackson -->
+ <Match>
+ <!-- Suppress a warning about ignoring return value because this is intentional. -->
+ <Class name="org.apache.kafka.common.config.AbstractConfig$ResolvingMap"/>
+ <Method name="get"/>
+ <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT"/>
+ </Match>
+
</FindBugsFilter>