You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2017/08/29 20:02:01 UTC
kafka git commit: KAFKA-5379: ProcessorContext.appConfigs() should
return parsed values
Repository: kafka
Updated Branches:
refs/heads/trunk 1f659913a -> 9ef2b3ce8
KAFKA-5379: ProcessorContext.appConfigs() should return parsed values
Author: Tommy Becker <to...@tivo.com>
Author: Tommy Becker <tw...@gmail.com>
Reviewers: Matthias J. Sax <ma...@confluent.io>, Guozhang Wang <wa...@gmail.com>
Closes #3455 from twbecker/kafka-5379
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/9ef2b3ce
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/9ef2b3ce
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/9ef2b3ce
Branch: refs/heads/trunk
Commit: 9ef2b3ce8e4553504e52385a6370bf641fc6ea1e
Parents: 1f65991
Author: Tommy Becker <to...@tivo.com>
Authored: Tue Aug 29 13:01:57 2017 -0700
Committer: Guozhang Wang <wa...@gmail.com>
Committed: Tue Aug 29 13:01:57 2017 -0700
----------------------------------------------------------------------
.../streams/processor/ProcessorContext.java | 5 ++
.../internals/AbstractProcessorContext.java | 6 +-
.../streams/state/internals/RocksDBStore.java | 8 +-
.../internals/AbstractProcessorContextTest.java | 28 ++++++-
.../state/internals/RocksDBStoreTest.java | 78 ++++++++++++--------
5 files changed, 84 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/9ef2b3ce/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
index 3468f1c..eef7e20 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
@@ -192,6 +192,11 @@ public interface ProcessorContext {
*
* The config properties are defined in the {@link org.apache.kafka.streams.StreamsConfig}
* object and associated to the ProcessorContext.
+ * <p>
+ * The type of the values is dependent on the {@link org.apache.kafka.common.config.ConfigDef.Type type} of the property
+ * (e.g. the value of {@link org.apache.kafka.streams.StreamsConfig#DEFAULT_KEY_SERDE_CLASS_CONFIG DEFAULT_KEY_SERDE_CLASS_CONFIG}
+ * will be of type {@link Class}, even if it was specified as a String to
+ * {@link org.apache.kafka.streams.StreamsConfig#StreamsConfig(Map) StreamsConfig(Map)}).
*
* @return all the key/values from the StreamsConfig properties
*/
http://git-wip-us.apache.org/repos/asf/kafka/blob/9ef2b3ce/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
index 04af9f2..2c28226 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
@@ -25,6 +25,7 @@ import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.state.internals.ThreadCache;
import java.io.File;
+import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
@@ -156,7 +157,10 @@ public abstract class AbstractProcessorContext implements InternalProcessorConte
@Override
public Map<String, Object> appConfigs() {
- return config.originals();
+ final Map<String, Object> combined = new HashMap<>();
+ combined.putAll(config.originals());
+ combined.putAll(config.values());
+ return combined;
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/9ef2b3ce/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
index f142431..d8a844c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
@@ -17,7 +17,6 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Utils;
@@ -155,11 +154,8 @@ public class RocksDBStore<K, V> implements KeyValueStore<K, V> {
fOptions.setWaitForFlush(true);
final Map<String, Object> configs = context.appConfigs();
- final Object configSetterValue = configs.get(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG);
- final Class<RocksDBConfigSetter> configSetterClass = (Class<RocksDBConfigSetter>) ConfigDef.parseType(
- StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG,
- configSetterValue,
- ConfigDef.Type.CLASS);
+ final Class<RocksDBConfigSetter> configSetterClass =
+ (Class<RocksDBConfigSetter>) configs.get(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG);
if (configSetterClass != null) {
final RocksDBConfigSetter configSetter = Utils.newInstance(configSetterClass);
http://git-wip-us.apache.org/repos/asf/kafka/blob/9ef2b3ce/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContextTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContextTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContextTest.java
index e0eed76..72082da 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContextTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContextTest.java
@@ -23,11 +23,14 @@ import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.state.RocksDBConfigSetter;
import org.apache.kafka.streams.state.internals.ThreadCache;
import org.apache.kafka.test.MockStateStoreSupplier;
import org.junit.Before;
import org.junit.Test;
+import java.util.Properties;
+
import static org.apache.kafka.test.StreamsTestUtils.minimalStreamsConfig;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.nullValue;
@@ -136,10 +139,29 @@ public class AbstractProcessorContextTest {
assertThat(context.timestamp(), equalTo(recordContext.timestamp()));
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void appConfigsShouldReturnParsedValues() {
+ assertThat((Class<RocksDBConfigSetter>) context.appConfigs().get(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG), equalTo(RocksDBConfigSetter.class));
+ }
+
+ @Test
+ public void appConfigsShouldReturnUnrecognizedValues() {
+ assertThat((String) context.appConfigs().get("user.supplied.config"), equalTo("user-suppplied-value"));
+ }
+
private static class TestProcessorContext extends AbstractProcessorContext {
- public TestProcessorContext(final MockStreamsMetrics metrics) {
- super(new TaskId(0, 0), "appId", new StreamsConfig(minimalStreamsConfig()), metrics, new StateManagerStub(), new ThreadCache("name", 0, metrics));
+ static Properties config;
+ static {
+ config = minimalStreamsConfig();
+ // Value must be a string to test className -> class conversion
+ config.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, RocksDBConfigSetter.class.getName());
+ config.put("user.supplied.config", "user-suppplied-value");
+ }
+
+ TestProcessorContext(final MockStreamsMetrics metrics) {
+ super(new TaskId(0, 0), "appId", new StreamsConfig(config), metrics, new StateManagerStub(), new ThreadCache("name", 0, metrics));
}
@Override
@@ -177,4 +199,4 @@ public class AbstractProcessorContextTest {
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/9ef2b3ce/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
index 148762d..8b916cb 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
@@ -17,14 +17,17 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.ProcessorStateException;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.internals.AbstractProcessorContext;
import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
-import org.apache.kafka.streams.processor.internals.RecordCollector;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.RocksDBConfigSetter;
import org.apache.kafka.test.MockProcessorContext;
@@ -97,23 +100,13 @@ public class RocksDBStoreTest {
}
@Test
- public void canSpecifyConfigSetterAsClass() throws Exception {
+ public void verifyRocksDbConfigSetterIsCalled() throws Exception {
final Map<String, Object> configs = new HashMap<>();
+ configs.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-application");
+ configs.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "test-server:9092");
configs.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, MockRocksDbConfigSetter.class);
MockRocksDbConfigSetter.called = false;
- subject.openDB(new ConfigurableProcessorContext(tempDir, Serdes.String(), Serdes.String(),
- null, null, configs));
-
- assertTrue(MockRocksDbConfigSetter.called);
- }
-
- @Test
- public void canSpecifyConfigSetterAsString() throws Exception {
- final Map<String, Object> configs = new HashMap<>();
- configs.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, MockRocksDbConfigSetter.class.getName());
- MockRocksDbConfigSetter.called = false;
- subject.openDB(new ConfigurableProcessorContext(tempDir, Serdes.String(), Serdes.String(),
- null, null, configs));
+ subject.openDB(new ConfigurableProcessorContext(new StreamsConfig(configs), tempDir));
assertTrue(MockRocksDbConfigSetter.called);
}
@@ -241,7 +234,7 @@ public class RocksDBStoreTest {
final List<KeyValue<byte[], byte[]>> entries = getKeyValueEntries();
subject.init(context, subject);
-
+
context.restore(subject.name(), entries);
assertEquals(subject.get("1"), "a");
@@ -314,7 +307,7 @@ public class RocksDBStoreTest {
}
@Test(expected = ProcessorStateException.class)
- public void shouldThrowProcessorStateExeptionOnPutDeletedDir() throws IOException {
+ public void shouldThrowProcessorStateExceptionOnPutDeletedDir() throws IOException {
subject.init(context, subject);
Utils.delete(dir);
subject.put("anyKey", "anyValue");
@@ -338,22 +331,45 @@ public class RocksDBStoreTest {
return entries;
}
- private static class ConfigurableProcessorContext extends MockProcessorContext {
- final Map<String, Object> configs;
-
- ConfigurableProcessorContext(final File stateDir,
- final Serde<?> keySerde,
- final Serde<?> valSerde,
- final RecordCollector collector,
- final ThreadCache cache,
- final Map<String, Object> configs) {
- super(stateDir, keySerde, valSerde, collector, cache);
- this.configs = configs;
+ private static class ConfigurableProcessorContext extends AbstractProcessorContext {
+ private final File stateDir;
+
+ ConfigurableProcessorContext(final StreamsConfig config,
+ final File stateDir) {
+ super(null, null, config, null, null, null);
+ this.stateDir = stateDir;
+ }
+
+ @Override
+ public File stateDir() {
+ return stateDir;
}
@Override
- public Map<String, Object> appConfigs() {
- return configs;
+ public StateStore getStateStore(final String name) {
+ return null;
}
+
+ @Override
+ public Cancellable schedule(final long interval,
+ final PunctuationType type,
+ final Punctuator callback) {
+ return null;
+ }
+
+ @Override
+ public void schedule(long interval) { }
+
+ @Override
+ public <K, V> void forward(K key, V value) { }
+
+ @Override
+ public <K, V> void forward(K key, V value, int childIndex) { }
+
+ @Override
+ public <K, V> void forward(K key, V value, String childName) { }
+
+ @Override
+ public void commit() { }
}
}