[
https://issues.apache.org/jira/browse/KAFKA-6849?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16481322#comment-16481322
]
ASF GitHub Bot commented on KAFKA-6849:
---------------------------------------
guozhangwang closed pull request #4959: KAFKA-6849: add transformValues methods
to KTable.
URL: https://github.com/apache/kafka/pull/4959
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/docs/streams/developer-guide/dsl-api.html
b/docs/streams/developer-guide/dsl-api.html
index 38955224278..4c76c5e84b6 100644
--- a/docs/streams/developer-guide/dsl-api.html
+++ b/docs/streams/developer-guide/dsl-api.html
@@ -2906,6 +2906,7 @@ <h2><a class="toc-backref" href="#id7">Overview</a><a
class="headerlink" href="#
<tr class="row-even"><td><p
class="first"><strong>Transform (values only)</strong></p>
<ul class="last simple">
<li>KStream -> KStream</li>
+ <li>KTable -> KTable</li>
</ul>
</td>
<td><p class="first">Applies a <code class="docutils
literal"><span class="pre">ValueTransformer</span></code> to each record, while
retaining the key of the original record.
diff --git a/docs/streams/upgrade-guide.html b/docs/streams/upgrade-guide.html
index 7788638c195..0e3725ebe32 100644
--- a/docs/streams/upgrade-guide.html
+++ b/docs/streams/upgrade-guide.html
@@ -192,6 +192,13 @@ <h3><a id="streams_api_changes_110"
href="#streams_api_changes_110">Streams API
to distinguish them from configurations of other clients that share
the same config names.
</p>
+ <p>
+ New method in <code>KTable</code>
+ </p>
+ <ul>
+ <li> <code>transformValues</code> methods have been added to
<code>KTable</code>. Similar to those on <code>KStream</code>, these methods
allow for richer, stateful, value transformation similar to the Processor
API.</li>
+ </ul>
+
<p>
New method in <code>GlobalKTable</code>
</p>
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
index 1d42f11c0e5..e75bb3aa227 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
@@ -544,12 +544,12 @@ void to(final String topic,
/**
* Transform the value of each input record into a new value (with
possible new type) of the output record.
- * A {@link ValueTransformer} (provided by the given {@link
ValueTransformerSupplier}) is applies to each input
+ * A {@link ValueTransformer} (provided by the given {@link
ValueTransformerSupplier}) is applied to each input
* record value and computes a new value for it.
* Thus, an input record {@code <K,V>} can be transformed into an output
record {@code <K:V'>}.
* This is a stateful record-by-record operation (cf. {@link
#mapValues(ValueMapper)}).
* Furthermore, via {@link
org.apache.kafka.streams.processor.Punctuator#punctuate(long)} the processing
progress can be observed and additional
- * periodic actions get be performed.
+ * periodic actions can be performed.
* <p>
* In order to assign a state, the state must be created and registered
beforehand:
* <pre>{@code
@@ -613,12 +613,12 @@ void to(final String topic,
/**
* Transform the value of each input record into a new value (with
possible new type) of the output record.
- * A {@link ValueTransformerWithKey} (provided by the given {@link
ValueTransformerWithKeySupplier}) is applies to each input
+ * A {@link ValueTransformerWithKey} (provided by the given {@link
ValueTransformerWithKeySupplier}) is applied to each input
* record value and computes a new value for it.
* Thus, an input record {@code <K,V>} can be transformed into an output
record {@code <K:V'>}.
* This is a stateful record-by-record operation (cf. {@link
#mapValues(ValueMapperWithKey)}).
* Furthermore, via {@link
org.apache.kafka.streams.processor.Punctuator#punctuate(long)} the processing
progress can be observed and additional
- * periodic actions get be performed.
+ * periodic actions can be performed.
* <p>
* In order to assign a state, the state must be created and registered
beforehand:
* <pre>{@code
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
index dbead79c1f6..da540ba627f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
@@ -23,6 +23,7 @@
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
@@ -382,6 +383,157 @@
*/
<KR> KStream<KR, V> toStream(final KeyValueMapper<? super K, ? super V, ?
extends KR> mapper);
+ /**
+ * Create a new {@code KTable} by transforming the value of each record in
this {@code KTable} into a new value,
+ * (with possibly new type).
+ * A {@link ValueTransformerWithKey} (provided by the given {@link
ValueTransformerWithKeySupplier}) is applied to each input
+ * record value and computes a new value for it.
+ * Thus, an input record {@code <K,V>} can be transformed into an output
record {@code <K:V'>}.
+ * This is similar to {@link #mapValues(ValueMapperWithKey)}, but more
flexible, allowing access to additional state-stores,
+ * and access to the {@link ProcessorContext}.
+ * Furthermore, via {@link
org.apache.kafka.streams.processor.Punctuator#punctuate(long)} the processing
progress can be observed and additional
+ * periodic actions can be performed.
+ * <p>
+ * If the downstream topology uses aggregation functions, (e.g. {@link
KGroupedTable#reduce}, {@link KGroupedTable#aggregate}, etc),
+ * care must be taken when dealing with state, (either held in
state-stores or transformer instances), to ensure correct aggregate results.
+ * In contrast, if the resulting KTable is materialized, (cf. {@link
#transformValues(ValueTransformerWithKeySupplier, Materialized, String...)}),
+ * such concerns are handled for you.
+ * <p>
+ * In order to assign a state, the state must be created and registered
beforehand:
+ * <pre>{@code
+ * // create store
+ * StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
+ *
Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myValueTransformState"),
+ * Serdes.String(),
+ * Serdes.String());
+ * // register store
+ * builder.addStateStore(keyValueStoreBuilder);
+ *
+ * KTable outputTable = inputTable.transformValues(new
ValueTransformerWithKeySupplier() { ... }, "myValueTransformState");
+ * }</pre>
+ * <p>
+ * Within the {@link ValueTransformerWithKey}, the state is obtained via
the
+ * {@link ProcessorContext}.
+ * To trigger periodic actions via {@link
org.apache.kafka.streams.processor.Punctuator#punctuate(long) punctuate()},
+ * a schedule must be registered.
+ * <pre>{@code
+ * new ValueTransformerWithKeySupplier() {
+ * ValueTransformerWithKey get() {
+ * return new ValueTransformerWithKey() {
+ * private KeyValueStore<String, String> state;
+ *
+ * void init(ProcessorContext context) {
+ * this.state = (KeyValueStore<String,
String>)context.getStateStore("myValueTransformState");
+ * context.schedule(1000, PunctuationType.WALL_CLOCK_TIME,
new Punctuator(..)); // punctuate each 1000ms, can access this.state
+ * }
+ *
+ * NewValueType transform(K readOnlyKey, V value) {
+ * // can access this.state and use read-only key
+ * return new NewValueType(readOnlyKey); // or null
+ * }
+ *
+ * void close() {
+ * // can access this.state
+ * }
+ * }
+ * }
+ * }
+ * }</pre>
+ * <p>
+ * Note that the key is read-only and should not be modified, as this can
lead to corrupt partitioning.
+ * Setting a new value preserves data co-location with respect to the key.
+ *
+ * @param transformerSupplier a instance of {@link
ValueTransformerWithKeySupplier} that generates a
+ * {@link ValueTransformerWithKey}.
+ * At least one transformer instance will be
created per streaming task.
+ * Transformers do not need to be thread-safe.
+ * @param stateStoreNames the names of the state stores used by the
processor
+ * @param <VR> the value type of the result table
+ * @return a {@code KTable} that contains records with unmodified key and
new values (possibly of different type)
+ * @see #mapValues(ValueMapper)
+ * @see #mapValues(ValueMapperWithKey)
+ */
+ <VR> KTable<K, VR> transformValues(final ValueTransformerWithKeySupplier<?
super K, ? super V, ? extends VR> transformerSupplier,
+ final String... stateStoreNames);
+
+ /**
+ * Create a new {@code KTable} by transforming the value of each record in
this {@code KTable} into a new value,
+ * (with possibly new type).
+ * A {@link ValueTransformerWithKey} (provided by the given {@link
ValueTransformerWithKeySupplier}) is applied to each input
+ * record value and computes a new value for it.
+ * This is similar to {@link #mapValues(ValueMapperWithKey)}, but more
flexible, allowing stateful, rather than stateless,
+ * record-by-record operation, access to additional state-stores, and
access to the {@link ProcessorContext}.
+ * Furthermore, via {@link
org.apache.kafka.streams.processor.Punctuator#punctuate(long)} the processing
progress can be observed and additional
+ * periodic actions can be performed.
+ * The resulting {@code KTable} is materialized into another state store
(additional to the provided state store names)
+ * as specified by the user via {@link Materialized} parameter, and is
queryable through its given name.
+ * <p>
+ * In order to assign a state, the state must be created and registered
beforehand:
+ * <pre>{@code
+ * // create store
+ * StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
+ *
Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myValueTransformState"),
+ * Serdes.String(),
+ * Serdes.String());
+ * // register store
+ * builder.addStateStore(keyValueStoreBuilder);
+ *
+ * KTable outputTable = inputTable.transformValues(
+ * new ValueTransformerWithKeySupplier() { ... },
+ * Materialized.<String, String, KeyValueStore<Bytes,
byte[]>>as("outputTable")
+ * .withKeySerde(Serdes.String())
+ * .withValueSerde(Serdes.String()),
+ * "myValueTransformState");
+ * }</pre>
+ * <p>
+ * Within the {@link ValueTransformerWithKey}, the state is obtained via
the
+ * {@link ProcessorContext}.
+ * To trigger periodic actions via {@link
org.apache.kafka.streams.processor.Punctuator#punctuate(long) punctuate()},
+ * a schedule must be registered.
+ * <pre>{@code
+ * new ValueTransformerWithKeySupplier() {
+ * ValueTransformerWithKey get() {
+ * return new ValueTransformerWithKey() {
+ * private KeyValueStore<String, String> state;
+ *
+ * void init(ProcessorContext context) {
+ * this.state = (KeyValueStore<String,
String>)context.getStateStore("myValueTransformState");
+ * context.schedule(1000, PunctuationType.WALL_CLOCK_TIME,
new Punctuator(..)); // punctuate each 1000ms, can access this.state
+ * }
+ *
+ * NewValueType transform(K readOnlyKey, V value) {
+ * // can access this.state and use read-only key
+ * return new NewValueType(readOnlyKey); // or null
+ * }
+ *
+ * void close() {
+ * // can access this.state
+ * }
+ * }
+ * }
+ * }
+ * }</pre>
+ * <p>
+ * Note that the key is read-only and should not be modified, as this can
lead to corrupt partitioning.
+ * Setting a new value preserves data co-location with respect to the key.
+ *
+ * @param transformerSupplier a instance of {@link
ValueTransformerWithKeySupplier} that generates a
+ * {@link ValueTransformerWithKey}.
+ * At least one transformer instance will be
created per streaming task.
+ * Transformers do not need to be thread-safe.
+ * @param materialized an instance of {@link Materialized} used to
describe how the state store of the
+ * resulting table should be materialized.
+ * Cannot be {@code null}
+ * @param stateStoreNames the names of the state stores used by the
processor
+ * @param <VR> the value type of the result table
+ * @return a {@code KTable} that contains records with unmodified key and
new values (possibly of different type)
+ * @see #mapValues(ValueMapper)
+ * @see #mapValues(ValueMapperWithKey)
+ */
+ <VR> KTable<K, VR> transformValues(final ValueTransformerWithKeySupplier<?
super K, ? super V, ? extends VR> transformerSupplier,
+ final Materialized<K, VR,
KeyValueStore<Bytes, byte[]>> materialized,
+ final String... stateStoreNames);
+
/**
* Re-groups the records of this {@code KTable} using the provided {@link
KeyValueMapper} and default serializers
* and deserializers.
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java
index 1170ff636c4..1b3a8f44e31 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java
@@ -126,5 +126,9 @@ public void init(final ProcessorContext context) {
public T get(final K key) {
return store.get(key);
}
+
+ @Override
+ public void close() {
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java
index 157c91b50e6..70b9bade2dc 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java
@@ -74,4 +74,9 @@ public void process(final K1 key, final V1 value) {
}
}
}
+
+ @Override
+ public void close() {
+ valueGetter.close();
+ }
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java
index fff93483d38..9f404ea55f6 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java
@@ -123,6 +123,10 @@ public void init(final ProcessorContext context) {
public V get(final K key) {
return store.get(key);
}
+
+ @Override
+ public void close() {
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
index 114c6856156..b91431aa75b 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
@@ -175,6 +175,10 @@ public T get(final Windowed<K> key) {
return value;
}
}
+
+ @Override
+ public void close() {
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValues.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValues.java
index d45b7cfab78..b645960e5fe 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValues.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValues.java
@@ -16,24 +16,12 @@
*/
package org.apache.kafka.streams.kstream.internals;
-import org.apache.kafka.common.serialization.Serde;
-import org.apache.kafka.streams.StreamsMetrics;
-import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
-import org.apache.kafka.streams.processor.Cancellable;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.ProcessorSupplier;
-import org.apache.kafka.streams.processor.PunctuationType;
-import org.apache.kafka.streams.processor.Punctuator;
-import org.apache.kafka.streams.processor.StateRestoreCallback;
-import org.apache.kafka.streams.processor.StateStore;
-import org.apache.kafka.streams.processor.TaskId;
-import org.apache.kafka.streams.processor.To;
-
-import java.io.File;
-import java.util.Map;
+import
org.apache.kafka.streams.processor.internals.ForwardingDisabledProcessorContext;
public class KStreamTransformValues<K, V, R> implements ProcessorSupplier<K,
V> {
@@ -59,111 +47,7 @@
@Override
public void init(final ProcessorContext context) {
- valueTransformer.init(
- new ProcessorContext() {
- @Override
- public String applicationId() {
- return context.applicationId();
- }
-
- @Override
- public TaskId taskId() {
- return context.taskId();
- }
-
- @Override
- public Serde<?> keySerde() {
- return context.keySerde();
- }
-
- @Override
- public Serde<?> valueSerde() {
- return context.valueSerde();
- }
-
- @Override
- public File stateDir() {
- return context.stateDir();
- }
-
- @Override
- public StreamsMetrics metrics() {
- return context.metrics();
- }
-
- @Override
- public void register(final StateStore store,
- final StateRestoreCallback
stateRestoreCallback) {
- context.register(store, stateRestoreCallback);
- }
-
- @Override
- public StateStore getStateStore(final String name) {
- return context.getStateStore(name);
- }
-
- @Override
- public Cancellable schedule(final long interval, final
PunctuationType type, final Punctuator callback) {
- return context.schedule(interval, type, callback);
- }
-
- @Override
- public <K, V> void forward(final K key, final V value) {
- throw new StreamsException("ProcessorContext#forward()
must not be called within TransformValues.");
- }
-
- @Override
- public <K, V> void forward(final K key, final V value,
final To to) {
- throw new StreamsException("ProcessorContext#forward()
must not be called within TransformValues.");
- }
-
- @SuppressWarnings("deprecation")
- @Override
- public <K, V> void forward(final K key, final V value,
final int childIndex) {
- throw new StreamsException("ProcessorContext#forward()
must not be called within TransformValues.");
- }
-
- @SuppressWarnings("deprecation")
- @Override
- public <K, V> void forward(final K key, final V value,
final String childName) {
- throw new StreamsException("ProcessorContext#forward()
must not be called within TransformValues.");
- }
-
- @Override
- public void commit() {
- context.commit();
- }
-
- @Override
- public String topic() {
- return context.topic();
- }
-
- @Override
- public int partition() {
- return context.partition();
- }
-
- @Override
- public long offset() {
- return context.offset();
- }
-
- @Override
- public long timestamp() {
- return context.timestamp();
- }
-
- @Override
- public Map<String, Object> appConfigs() {
- return context.appConfigs();
- }
-
- @Override
- public Map<String, Object> appConfigsWithPrefix(String
prefix) {
- return context.appConfigsWithPrefix(prefix);
- }
- });
+ valueTransformer.init(new
ForwardingDisabledProcessorContext(context));
this.context = context;
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java
index f4bd099c3d9..a7747629948 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java
@@ -145,5 +145,9 @@ public T get(final Windowed<K> windowedKey) {
return windowStore.fetch(key, window.start());
}
+
+ @Override
+ public void close() {
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowReduce.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowReduce.java
index 32dfba39de5..9db861de5c9 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowReduce.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowReduce.java
@@ -141,5 +141,9 @@ public V get(final Windowed<K> windowedKey) {
return windowStore.fetch(key, window.start());
}
+
+ @Override
+ public void close() {
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java
index ee8982bbe69..d020e4b1276 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableFilter.java
@@ -131,6 +131,11 @@ public void init(final ProcessorContext context) {
public V get(final K key) {
return computeValue(key, parentGetter.get(key));
}
+
+ @Override
+ public void close() {
+ parentGetter.close();
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
index c1f0f7ac3fe..bcd31bba16e 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
@@ -29,6 +29,7 @@
import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.kstream.ValueMapperWithKey;
+import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
@@ -63,6 +64,8 @@
private static final String TOSTREAM_NAME = "KTABLE-TOSTREAM-";
+ private static final String TRANSFORMVALUES_NAME =
"KTABLE-TRANSFORMVALUES-";
+
private final ProcessorSupplier<?, ?> processorSupplier;
private final String queryableStoreName;
@@ -219,6 +222,56 @@ public String queryableStoreName() {
return doMapValues(mapper, new MaterializedInternal<>(materialized,
builder, MAPVALUES_NAME));
}
+ @Override
+ public <VR> KTable<K, VR> transformValues(final
ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR>
transformerSupplier,
+ final String... stateStoreNames)
{
+ return doTransformValues(transformerSupplier, null, stateStoreNames);
+ }
+
+ @Override
+ public <VR> KTable<K, VR> transformValues(final
ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR>
transformerSupplier,
+ final Materialized<K, VR,
KeyValueStore<Bytes, byte[]>> materialized,
+ final String... stateStoreNames)
{
+ Objects.requireNonNull(materialized, "materialized can't be null");
+ return doTransformValues(transformerSupplier,
+ new MaterializedInternal<>(materialized, builder,
TRANSFORMVALUES_NAME), stateStoreNames);
+ }
+
+ private <VR> KTable<K, VR> doTransformValues(final
ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR>
transformerSupplier,
+ final MaterializedInternal<K,
VR, KeyValueStore<Bytes, byte[]>> materialized,
+ final String...
stateStoreNames) {
+ Objects.requireNonNull(stateStoreNames, "stateStoreNames");
+
+ final String name = builder.newProcessorName(TRANSFORMVALUES_NAME);
+
+ final boolean shouldMaterialize = materialized != null &&
materialized.isQueryable();
+
+ final KTableProcessorSupplier<K, V, VR> processorSupplier = new
KTableTransformValues<>(
+ this,
+ transformerSupplier,
+ shouldMaterialize ? materialized.storeName() : null);
+
+ builder.internalTopologyBuilder.addProcessor(name, processorSupplier,
this.name);
+
+ if (stateStoreNames.length > 0) {
+
builder.internalTopologyBuilder.connectProcessorAndStateStores(name,
stateStoreNames);
+ }
+
+ if (shouldMaterialize) {
+ builder.internalTopologyBuilder.addStateStore(
+ new KeyValueStoreMaterializer<>(materialized).materialize(),
+ name);
+ }
+
+ return new KTableImpl<>(
+ builder,
+ name,
+ processorSupplier,
+ sourceNodes,
+ shouldMaterialize ? materialized.storeName() :
this.queryableStoreName,
+ shouldMaterialize);
+ }
+
@Override
public KStream<K, V> toStream() {
String name = builder.newProcessorName(TOSTREAM_NAME);
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java
index 5a159ff5168..a7bbb56f360 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java
@@ -109,6 +109,11 @@ public void process(final K key, final Change<V1> change) {
context().forward(key, new Change<>(newValue, oldValue));
}
+
+ @Override
+ public void close() {
+ valueGetter.close();
+ }
}
private class KTableKTableInnerJoinValueGetter implements
KTableValueGetter<K, R> {
@@ -144,5 +149,11 @@ public R get(final K key) {
return null;
}
}
+
+ @Override
+ public void close() {
+ valueGetter1.close();
+ valueGetter2.close();
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java
index a40ac5df2b5..cbb63c6ba46 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java
@@ -102,6 +102,11 @@ public void process(final K key, final Change<V1> change) {
context().forward(key, new Change<>(newValue, oldValue));
}
+
+ @Override
+ public void close() {
+ valueGetter.close();
+ }
}
private class KTableKTableLeftJoinValueGetter implements
KTableValueGetter<K, R> {
@@ -133,6 +138,11 @@ public R get(final K key) {
}
}
+ @Override
+ public void close() {
+ valueGetter1.close();
+ valueGetter2.close();
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java
index 377e2d980d6..27eb6987782 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java
@@ -101,6 +101,11 @@ public void process(final K key, final Change<V1> change) {
context().forward(key, new Change<>(newValue, oldValue));
}
+
+ @Override
+ public void close() {
+ valueGetter.close();
+ }
}
private class KTableKTableOuterJoinValueGetter implements
KTableValueGetter<K, R> {
@@ -133,6 +138,11 @@ public R get(final K key) {
return newValue;
}
+ @Override
+ public void close() {
+ valueGetter1.close();
+ valueGetter2.close();
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java
index d3916dd04db..1e634d00448 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java
@@ -100,6 +100,10 @@ public void process(final K key, final Change<V1> change) {
context().forward(key, new Change<>(newValue, oldValue));
}
+ @Override
+ public void close() {
+ valueGetter.close();
+ }
}
private class KTableKTableRightJoinValueGetter implements
KTableValueGetter<K, R> {
@@ -131,6 +135,11 @@ public R get(final K key) {
}
}
+ @Override
+ public void close() {
+ valueGetter1.close();
+ valueGetter2.close();
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMapValues.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMapValues.java
index 0970b4276ce..1106a94f35b 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMapValues.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMapValues.java
@@ -125,6 +125,12 @@ public void init(final ProcessorContext context) {
public V1 get(final K key) {
return computeValue(key, parentGetter.get(key));
}
+
+
+ @Override
+ public void close() {
+ parentGetter.close();
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMaterializedValueGetterSupplier.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMaterializedValueGetterSupplier.java
index 4cecccefecd..0c17d5908f8 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMaterializedValueGetterSupplier.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableMaterializedValueGetterSupplier.java
@@ -50,5 +50,9 @@ public void init(final ProcessorContext context) {
public V get(final K key) {
return store.get(key);
}
+
+ @Override
+ public void close() {
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
index 20d943ea9f8..e86445a3113 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableRepartitionMap.java
@@ -115,6 +115,11 @@ public void init(final ProcessorContext context) {
public KeyValue<K1, V1> get(final K key) {
return mapper.apply(key, parentGetter.get(key));
}
+
+ @Override
+ public void close() {
+ parentGetter.close();
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSourceValueGetterSupplier.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSourceValueGetterSupplier.java
index 714b8c50dfa..ebf16d17bef 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSourceValueGetterSupplier.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSourceValueGetterSupplier.java
@@ -49,6 +49,9 @@ public V get(K key) {
return store.get(key);
}
+ @Override
+ public void close() {
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableTransformValues.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableTransformValues.java
new file mode 100644
index 00000000000..b3e84d79119
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableTransformValues.java
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.kafka.streams.kstream.internals;
+
+import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
+import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
+import org.apache.kafka.streams.processor.AbstractProcessor;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import
org.apache.kafka.streams.processor.internals.ForwardingDisabledProcessorContext;
+import org.apache.kafka.streams.state.KeyValueStore;
+
+import java.util.Objects;
+
+class KTableTransformValues<K, V, V1> implements KTableProcessorSupplier<K, V,
V1> {
+
+ private final KTableImpl<K, ?, V> parent;
+ private final ValueTransformerWithKeySupplier<? super K, ? super V, ?
extends V1> transformerSupplier;
+ private final String queryableName;
+ private boolean sendOldValues = false;
+
+ KTableTransformValues(final KTableImpl<K, ?, V> parent,
+ final ValueTransformerWithKeySupplier<? super K, ?
super V, ? extends V1> transformerSupplier,
+ final String queryableName) {
+ this.parent = Objects.requireNonNull(parent, "parent");
+ this.transformerSupplier = Objects.requireNonNull(transformerSupplier,
"transformerSupplier");
+ this.queryableName = queryableName;
+ }
+
+ @Override
+ public Processor<K, Change<V>> get() {
+ return new KTableTransformValuesProcessor(transformerSupplier.get());
+ }
+
+ @Override
+ public KTableValueGetterSupplier<K, V1> view() {
+ if (queryableName != null) {
+ return new KTableMaterializedValueGetterSupplier<>(queryableName);
+ }
+
+ return new KTableValueGetterSupplier<K, V1>() {
+ final KTableValueGetterSupplier<K, V> parentValueGetterSupplier =
parent.valueGetterSupplier();
+
+ public KTableValueGetter<K, V1> get() {
+ return new KTableTransformValuesGetter(
+ parentValueGetterSupplier.get(),
+ transformerSupplier.get());
+ }
+
+ @Override
+ public String[] storeNames() {
+ return parentValueGetterSupplier.storeNames();
+ }
+ };
+ }
+
+ @Override
+ public void enableSendingOldValues() {
+ parent.enableSendingOldValues();
+ sendOldValues = true;
+ }
+
+ private class KTableTransformValuesProcessor extends AbstractProcessor<K,
Change<V>> {
+ private final ValueTransformerWithKey<? super K, ? super V, ? extends
V1> valueTransformer;
+ private KeyValueStore<K, V1> store;
+ private TupleForwarder<K, V1> tupleForwarder;
+
+ private KTableTransformValuesProcessor(final ValueTransformerWithKey<?
super K, ? super V, ? extends V1> valueTransformer) {
+ this.valueTransformer = Objects.requireNonNull(valueTransformer,
"valueTransformer");
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void init(final ProcessorContext context) {
+ super.init(context);
+
+ valueTransformer.init(new
ForwardingDisabledProcessorContext(context));
+
+ if (queryableName != null) {
+ final ForwardingCacheFlushListener<K, V1> flushListener = new
ForwardingCacheFlushListener<>(context, sendOldValues);
+ store = (KeyValueStore<K, V1>)
context.getStateStore(queryableName);
+ tupleForwarder = new TupleForwarder<>(store, context,
flushListener, sendOldValues);
+ }
+ }
+
+ @Override
+ public void process(final K key, final Change<V> change) {
+ final V1 newValue = valueTransformer.transform(key,
change.newValue);
+
+ if (queryableName == null) {
+ final V1 oldValue = sendOldValues ?
valueTransformer.transform(key, change.oldValue) : null;
+ context().forward(key, new Change<>(newValue, oldValue));
+ } else {
+ final V1 oldValue = sendOldValues ? store.get(key) : null;
+ store.put(key, newValue);
+ tupleForwarder.maybeForward(key, newValue, oldValue);
+ }
+ }
+
+ @Override
+ public void close() {
+ valueTransformer.close();
+ }
+ }
+
+ private class KTableTransformValuesGetter implements KTableValueGetter<K,
V1> {
+
+ private final KTableValueGetter<K, V> parentGetter;
+ private final ValueTransformerWithKey<? super K, ? super V, ? extends
V1> valueTransformer;
+
+ KTableTransformValuesGetter(final KTableValueGetter<K, V> parentGetter,
+ final ValueTransformerWithKey<? super K, ?
super V, ? extends V1> valueTransformer) {
+ this.parentGetter = Objects.requireNonNull(parentGetter,
"parentGetter");
+ this.valueTransformer = Objects.requireNonNull(valueTransformer,
"valueTransformer");
+ }
+
+ @Override
+ public void init(final ProcessorContext context) {
+ parentGetter.init(context);
+
+ valueTransformer.init(new
ForwardingDisabledProcessorContext(context));
+ }
+
+ @Override
+ public V1 get(final K key) {
+ return valueTransformer.transform(key, parentGetter.get(key));
+ }
+
+ @Override
+ public void close() {
+ parentGetter.close();
+ valueTransformer.close();
+ }
+ }
+}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableValueGetter.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableValueGetter.java
index bb69236f0b4..edd964474e2 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableValueGetter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableValueGetter.java
@@ -24,4 +24,5 @@
V get(K key);
+ void close();
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ForwardingDisabledProcessorContext.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ForwardingDisabledProcessorContext.java
new file mode 100644
index 00000000000..35a0a7ebe1b
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ForwardingDisabledProcessorContext.java
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.Cancellable;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.PunctuationType;
+import org.apache.kafka.streams.processor.Punctuator;
+import org.apache.kafka.streams.processor.StateRestoreCallback;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.To;
+
+import java.io.File;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * {@code ProcessorContext} implementation that will throw on any forward call.
+ */
+public final class ForwardingDisabledProcessorContext implements
ProcessorContext {
+ private final ProcessorContext delegate;
+
+ public ForwardingDisabledProcessorContext(final ProcessorContext delegate)
{
+ this.delegate = Objects.requireNonNull(delegate, "delegate");
+ }
+
+ @Override
+ public String applicationId() {
+ return delegate.applicationId();
+ }
+
+ @Override
+ public TaskId taskId() {
+ return delegate.taskId();
+ }
+
+ @Override
+ public Serde<?> keySerde() {
+ return delegate.keySerde();
+ }
+
+ @Override
+ public Serde<?> valueSerde() {
+ return delegate.valueSerde();
+ }
+
+ @Override
+ public File stateDir() {
+ return delegate.stateDir();
+ }
+
+ @Override
+ public StreamsMetrics metrics() {
+ return delegate.metrics();
+ }
+
+ @Override
+ public void register(final StateStore store,
+ final StateRestoreCallback stateRestoreCallback) {
+ delegate.register(store, stateRestoreCallback);
+ }
+
+ @Override
+ public StateStore getStateStore(final String name) {
+ return delegate.getStateStore(name);
+ }
+
+ @Override
+ public Cancellable schedule(final long intervalMs,
+ final PunctuationType type,
+ final Punctuator callback) {
+ return delegate.schedule(intervalMs, type, callback);
+ }
+
+ @Override
+ public <K, V> void forward(final K key, final V value) {
+ throw new StreamsException("ProcessorContext#forward() not
supported.");
+ }
+
+ @Override
+ public <K, V> void forward(final K key, final V value, final To to) {
+ throw new StreamsException("ProcessorContext#forward() not
supported.");
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public <K, V> void forward(final K key, final V value, final int
childIndex) {
+ throw new StreamsException("ProcessorContext#forward() not
supported.");
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public <K, V> void forward(final K key, final V value, final String
childName) {
+ throw new StreamsException("ProcessorContext#forward() not
supported.");
+ }
+
+ @Override
+ public void commit() {
+ delegate.commit();
+ }
+
+ @Override
+ public String topic() {
+ return delegate.topic();
+ }
+
+ @Override
+ public int partition() {
+ return delegate.partition();
+ }
+
+ @Override
+ public long offset() {
+ return delegate.offset();
+ }
+
+ @Override
+ public long timestamp() {
+ return delegate.timestamp();
+ }
+
+ @Override
+ public Map<String, Object> appConfigs() {
+ return delegate.appConfigs();
+ }
+
+ @Override
+ public Map<String, Object> appConfigsWithPrefix(final String prefix) {
+ return delegate.appConfigsWithPrefix(prefix);
+ }
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java
index fb8a7bc6204..eac2585a1d6 100644
---
a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamTransformValuesTest.java
@@ -21,7 +21,6 @@
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.TopologyTestDriver;
-import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.ValueTransformer;
import org.apache.kafka.streams.kstream.ValueTransformerSupplier;
@@ -29,23 +28,32 @@
import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
-import org.apache.kafka.streams.processor.To;
+import
org.apache.kafka.streams.processor.internals.ForwardingDisabledProcessorContext;
import org.apache.kafka.streams.test.ConsumerRecordFactory;
import org.apache.kafka.test.MockProcessorSupplier;
+import org.apache.kafka.test.SingletonNoOpValueTransformer;
import org.apache.kafka.test.StreamsTestUtils;
+import org.easymock.EasyMockRunner;
+import org.easymock.Mock;
+import org.easymock.MockType;
import org.junit.Test;
+import org.junit.runner.RunWith;
import java.util.Properties;
+import static org.hamcrest.CoreMatchers.isA;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.fail;
+@RunWith(EasyMockRunner.class)
public class KStreamTransformValuesTest {
private String topicName = "topic";
private final MockProcessorSupplier<Integer, Integer> supplier = new
MockProcessorSupplier<>();
private final ConsumerRecordFactory<Integer, Integer> recordFactory = new
ConsumerRecordFactory<>(new IntegerSerializer(), new IntegerSerializer());
private final Properties props =
StreamsTestUtils.topologyTestConfig(Serdes.Integer(), Serdes.Integer());
+ @Mock(MockType.NICE)
+ private ProcessorContext context;
@Test
public void testTransform() {
@@ -134,90 +142,14 @@ public void close() {
assertArrayEquals(expected,
supplier.theCapturedProcessor().processed.toArray());
}
-
@Test
- public void
shouldNotAllowValueTransformerToCallInternalProcessorContextMethods() {
- final BadValueTransformer badValueTransformer = new
BadValueTransformer();
- final KStreamTransformValues<Integer, Integer, Integer> transformValue
= new KStreamTransformValues<>(new ValueTransformerWithKeySupplier<Integer,
Integer, Integer>() {
- @Override
- public ValueTransformerWithKey<Integer, Integer, Integer> get() {
- return new ValueTransformerWithKey<Integer, Integer,
Integer>() {
- @Override
- public void init(final ProcessorContext context) {
- badValueTransformer.init(context);
- }
-
- @Override
- public Integer transform(final Integer readOnlyKey, final
Integer value) {
- return badValueTransformer.transform(readOnlyKey,
value);
- }
-
- @Override
- public void close() {
- badValueTransformer.close();
- }
- };
- }
- });
-
- final Processor transformValueProcessor = transformValue.get();
- transformValueProcessor.init(null);
-
- try {
- transformValueProcessor.process(null, 0);
- fail("should not allow call to context.forward() within
ValueTransformer");
- } catch (final StreamsException e) {
- // expected
- }
+ public void
shouldInitializeTransformerWithForwardDisabledProcessorContext() {
+ final SingletonNoOpValueTransformer<String, String> transformer = new
SingletonNoOpValueTransformer<>();
+ final KStreamTransformValues<String, String, String> transformValues =
new KStreamTransformValues<>(transformer);
+ final Processor<String, String> processor = transformValues.get();
- try {
- transformValueProcessor.process(null, 1);
- fail("should not allow call to context.forward() within
ValueTransformer");
- } catch (final StreamsException e) {
- // expected
- }
-
- try {
- transformValueProcessor.process(null, 2);
- fail("should not allow call to context.forward() within
ValueTransformer");
- } catch (final StreamsException e) {
- // expected
- }
-
- try {
- transformValueProcessor.process(null, 3);
- fail("should not allow call to context.forward() within
ValueTransformer");
- } catch (final StreamsException e) {
- // expected
- }
- }
-
- private static final class BadValueTransformer implements
ValueTransformerWithKey<Integer, Integer, Integer> {
- private ProcessorContext context;
-
- @Override
- public void init(final ProcessorContext context) {
- this.context = context;
- }
-
- @Override
- public Integer transform(final Integer key, final Integer value) {
- if (value == 0) {
- context.forward(null, null);
- }
- if (value == 1) {
- context.forward(null, null, (String) null);
- }
- if (value == 2) {
- context.forward(null, null, 0);
- }
- if (value == 3) {
- context.forward(null, null, To.all());
- }
- throw new RuntimeException("Should never happen in this test");
- }
+ processor.init(context);
- @Override
- public void close() { }
+ assertThat(transformer.context, isA((Class)
ForwardingDisabledProcessorContext.class));
}
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
index a265c62a003..399e519bb4c 100644
---
a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableImplTest.java
@@ -29,6 +29,7 @@
import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.kstream.ValueMapperWithKey;
+import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
import org.apache.kafka.streams.processor.internals.SinkNode;
import org.apache.kafka.streams.processor.internals.SourceNode;
import org.apache.kafka.streams.state.KeyValueStore;
@@ -49,6 +50,7 @@
import java.lang.reflect.Field;
import java.util.List;
+import static org.easymock.EasyMock.mock;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
@@ -472,4 +474,23 @@ public void
shouldThrowNullPointerOnLeftJoinWhenMaterializedIsNull() {
public void shouldThrowNullPointerOnOuterJoinWhenMaterializedIsNull() {
table.leftJoin(table, MockValueJoiner.TOSTRING_JOINER, (Materialized)
null);
}
+
+ @Test(expected = NullPointerException.class)
+ public void
shouldThrowNullPointerOnTransformValuesWithKeyWhenTransformerSupplierIsNull() {
+ table.transformValues((ValueTransformerWithKeySupplier) null);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = NullPointerException.class)
+ public void
shouldThrowNullPointerOnTransformValuesWithKeyWhenMaterializedIsNull() {
+ final ValueTransformerWithKeySupplier<String, String, ?>
valueTransformerSupplier = mock(ValueTransformerWithKeySupplier.class);
+ table.transformValues(valueTransformerSupplier, (Materialized) null);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(expected = NullPointerException.class)
+ public void
shouldThrowNullPointerOnTransformValuesWithKeyWhenStoreNamesNull() {
+ final ValueTransformerWithKeySupplier<String, String, ?>
valueTransformerSupplier = mock(ValueTransformerWithKeySupplier.class);
+ table.transformValues(valueTransformerSupplier, (String[]) null);
+ }
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableTransformValuesTest.java
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableTransformValuesTest.java
new file mode 100644
index 00000000000..7c12dad2e64
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableTransformValuesTest.java
@@ -0,0 +1,547 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.kafka.streams.kstream.internals;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.TopologyTestDriver;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.kstream.Serialized;
+import org.apache.kafka.streams.kstream.ValueMapper;
+import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
+import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import
org.apache.kafka.streams.processor.internals.ForwardingDisabledProcessorContext;
+import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.StoreBuilder;
+import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.streams.test.ConsumerRecordFactory;
+import org.apache.kafka.test.MockProcessorSupplier;
+import org.apache.kafka.test.MockReducer;
+import org.apache.kafka.test.SingletonNoOpValueTransformer;
+import org.apache.kafka.test.TestUtils;
+import org.easymock.EasyMockRunner;
+import org.easymock.Mock;
+import org.easymock.MockType;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.isA;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
+
+@RunWith(EasyMockRunner.class)
+public class KTableTransformValuesTest {
+ private static final String QUERYABLE_NAME = "queryable-store";
+ private static final String INPUT_TOPIC = "inputTopic";
+ private static final String STORE_NAME = "someStore";
+ private static final String OTHER_STORE_NAME = "otherStore";
+
+ private static final Consumed<String, String> CONSUMED =
Consumed.with(Serdes.String(), Serdes.String());
+
+ private final ConsumerRecordFactory<String, String> recordFactory
+ = new ConsumerRecordFactory<>(new StringSerializer(), new
StringSerializer());
+
+ private TopologyTestDriver driver;
+ private MockProcessorSupplier<String, String> capture;
+ private StreamsBuilder builder;
+ @Mock(MockType.NICE)
+ private KTableImpl<String, String, String> parent;
+ @Mock(MockType.NICE)
+ private InternalProcessorContext context;
+ @Mock(MockType.NICE)
+ private KTableValueGetterSupplier<String, String> parentGetterSupplier;
+ @Mock(MockType.NICE)
+ private KTableValueGetter<String, String> parentGetter;
+ @Mock(MockType.NICE)
+ private KeyValueStore<String, String> stateStore;
+ @Mock(MockType.NICE)
+ private ValueTransformerWithKeySupplier<String, String, String>
mockSupplier;
+ @Mock(MockType.NICE)
+ private ValueTransformerWithKey<String, String, String> transformer;
+
+ @After
+ public void cleanup() {
+ if (driver != null) {
+ driver.close();
+ driver = null;
+ }
+ }
+
+ @Before
+ public void setUp() {
+ capture = new MockProcessorSupplier<>();
+ builder = new StreamsBuilder();
+ }
+
+ @Test
+ public void shouldThrowOnGetIfSupplierReturnsNull() {
+ final KTableTransformValues<String, String, String> transformer =
+ new KTableTransformValues<>(parent, new NullSupplier(),
QUERYABLE_NAME);
+
+ try {
+ transformer.get();
+ fail("NPE expected");
+ } catch (final NullPointerException expected) {
+ // expected
+ }
+ }
+
+ @Test
+ public void shouldThrowOnViewGetIfSupplierReturnsNull() {
+ final KTableValueGetterSupplier<String, String> view =
+ new KTableTransformValues<>(parent, new NullSupplier(),
null).view();
+
+ try {
+ view.get();
+ fail("NPE expected");
+ } catch (final NullPointerException expected) {
+ // expected
+ }
+ }
+
+ @Test
+ public void
shouldInitializeTransformerWithForwardDisabledProcessorContext() {
+ final SingletonNoOpValueTransformer<String, String> transformer = new
SingletonNoOpValueTransformer<>();
+ final KTableTransformValues<String, String, String> transformValues =
new KTableTransformValues<>(parent, transformer, null);
+ final Processor<String, Change<String>> processor =
transformValues.get();
+
+ processor.init(context);
+
+ assertThat(transformer.context, isA((Class)
ForwardingDisabledProcessorContext.class));
+ }
+
+ @Test
+ public void shouldNotSendOldValuesByDefault() {
+ final KTableTransformValues<String, String, String> transformValues =
+ new KTableTransformValues<>(parent, new
ExclamationValueTransformerSupplier(), null);
+
+ final Processor<String, Change<String>> processor =
transformValues.get();
+ processor.init(context);
+
+ context.forward("Key", new Change<>("Key->newValue!", null));
+ expectLastCall();
+ replay(context);
+
+ processor.process("Key", new Change<>("newValue", "oldValue"));
+
+ verify(context);
+ }
+
+ @Test
+ public void shouldSendOldValuesIfConfigured() {
+ final KTableTransformValues<String, String, String> transformValues =
+ new KTableTransformValues<>(parent, new
ExclamationValueTransformerSupplier(), null);
+
+ transformValues.enableSendingOldValues();
+ final Processor<String, Change<String>> processor =
transformValues.get();
+ processor.init(context);
+
+ context.forward("Key", new Change<>("Key->newValue!",
"Key->oldValue!"));
+ expectLastCall();
+ replay(context);
+
+ processor.process("Key", new Change<>("newValue", "oldValue"));
+
+ verify(context);
+ }
+
+ @Test
+ public void shouldSetSendOldValuesOnParent() {
+ parent.enableSendingOldValues();
+ expectLastCall();
+ replay(parent);
+
+ new KTableTransformValues<>(parent, new
SingletonNoOpValueTransformer<>(), QUERYABLE_NAME).enableSendingOldValues();
+
+ verify(parent);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldTransformOnGetIfNotMaterialized() {
+ final KTableTransformValues<String, String, String> transformValues =
+ new KTableTransformValues<>(parent, new
ExclamationValueTransformerSupplier(), null);
+
+ expect(parent.valueGetterSupplier()).andReturn(parentGetterSupplier);
+ expect(parentGetterSupplier.get()).andReturn(parentGetter);
+ expect(parentGetter.get("Key")).andReturn("Value");
+ replay(parent, parentGetterSupplier, parentGetter);
+
+ final KTableValueGetter<String, String> getter =
transformValues.view().get();
+ getter.init(context);
+
+ final String result = getter.get("Key");
+
+ assertThat(result, is("Key->Value!"));
+ }
+
+ @Test
+ public void shouldGetFromStateStoreIfMaterialized() {
+ final KTableTransformValues<String, String, String> transformValues =
+ new KTableTransformValues<>(parent, new
ExclamationValueTransformerSupplier(), QUERYABLE_NAME);
+
+ expect(context.getStateStore(QUERYABLE_NAME)).andReturn(stateStore);
+ expect(stateStore.get("Key")).andReturn("something");
+ replay(context, stateStore);
+
+ final KTableValueGetter<String, String> getter =
transformValues.view().get();
+ getter.init(context);
+
+ final String result = getter.get("Key");
+
+ assertThat(result, is("something"));
+ }
+
+ @Test
+ public void shouldGetStoreNamesFromParentIfNotMaterialized() {
+ final KTableTransformValues<String, String, String> transformValues =
+ new KTableTransformValues<>(parent, new
ExclamationValueTransformerSupplier(), null);
+
+ expect(parent.valueGetterSupplier()).andReturn(parentGetterSupplier);
+ expect(parentGetterSupplier.storeNames()).andReturn(new
String[]{"store1", "store2"});
+ replay(parent, parentGetterSupplier);
+
+ final String[] storeNames = transformValues.view().storeNames();
+
+ assertThat(storeNames, is(new String[]{"store1", "store2"}));
+ }
+
+ @Test
+ public void shouldGetQueryableStoreNameIfMaterialized() {
+ final KTableTransformValues<String, String, String> transformValues =
+ new KTableTransformValues<>(parent, new
ExclamationValueTransformerSupplier(), QUERYABLE_NAME);
+
+ final String[] storeNames = transformValues.view().storeNames();
+
+ assertThat(storeNames, is(new String[]{QUERYABLE_NAME}));
+ }
+
+ @Test
+ public void shouldCloseTransformerOnProcessorClose() {
+ final KTableTransformValues<String, String, String> transformValues =
+ new KTableTransformValues<>(parent, mockSupplier, null);
+
+ expect(mockSupplier.get()).andReturn(transformer);
+ transformer.close();
+ expectLastCall();
+ replay(mockSupplier, transformer);
+
+ final Processor<String, Change<String>> processor =
transformValues.get();
+ processor.close();
+
+ verify(transformer);
+ }
+
+ @Test
+ public void shouldCloseTransformerOnGetterClose() {
+ final KTableTransformValues<String, String, String> transformValues =
+ new KTableTransformValues<>(parent, mockSupplier, null);
+
+ expect(mockSupplier.get()).andReturn(transformer);
+ expect(parentGetterSupplier.get()).andReturn(parentGetter);
+ expect(parent.valueGetterSupplier()).andReturn(parentGetterSupplier);
+
+ transformer.close();
+ expectLastCall();
+
+ replay(mockSupplier, transformer, parent, parentGetterSupplier);
+
+ final KTableValueGetter<String, String> getter =
transformValues.view().get();
+ getter.close();
+
+ verify(transformer);
+ }
+
+ @Test
+ public void shouldCloseParentGetterClose() {
+ final KTableTransformValues<String, String, String> transformValues =
+ new KTableTransformValues<>(parent, mockSupplier, null);
+
+ expect(parent.valueGetterSupplier()).andReturn(parentGetterSupplier);
+ expect(mockSupplier.get()).andReturn(transformer);
+ expect(parentGetterSupplier.get()).andReturn(parentGetter);
+
+ parentGetter.close();
+ expectLastCall();
+
+ replay(mockSupplier, parent, parentGetterSupplier, parentGetter);
+
+ final KTableValueGetter<String, String> getter =
transformValues.view().get();
+ getter.close();
+
+ verify(parentGetter);
+ }
+
+ @Test
+ public void shouldTransformValuesWithKey() {
+ builder
+ .addStateStore(storeBuilder(STORE_NAME))
+ .addStateStore(storeBuilder(OTHER_STORE_NAME))
+ .table(INPUT_TOPIC, CONSUMED)
+ .transformValues(
+ new ExclamationValueTransformerSupplier(STORE_NAME,
OTHER_STORE_NAME),
+ STORE_NAME, OTHER_STORE_NAME)
+ .toStream()
+ .process(capture);
+
+ driver = new TopologyTestDriver(builder.build(), props());
+
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "A", "a", 0L));
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "B", "b", 0L));
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "D", null, 0L));
+
+ assertThat(output(), hasItems("A:A->a!", "B:B->b!", "D:D->null!"));
+ assertThat("Store should not be materialized",
driver.getKeyValueStore(QUERYABLE_NAME), is(nullValue()));
+ }
+
+ @Test
+ public void shouldTransformValuesWithKeyAndMaterialize() {
+ builder
+ .addStateStore(storeBuilder(STORE_NAME))
+ .table(INPUT_TOPIC, CONSUMED)
+ .transformValues(
+ new ExclamationValueTransformerSupplier(STORE_NAME,
QUERYABLE_NAME),
+ Materialized.<String, String, KeyValueStore<Bytes,
byte[]>>as(QUERYABLE_NAME)
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.String()),
+ STORE_NAME)
+ .toStream()
+ .process(capture);
+
+ driver = new TopologyTestDriver(builder.build(), props());
+
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "A", "a", 0L));
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "B", "b", 0L));
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "C", null, 0L));
+
+ assertThat(output(), hasItems("A:A->a!", "B:B->b!", "C:C->null!"));
+
+ final KeyValueStore<String, String> keyValueStore =
driver.getKeyValueStore(QUERYABLE_NAME);
+ assertThat(keyValueStore.get("A"), is("A->a!"));
+ assertThat(keyValueStore.get("B"), is("B->b!"));
+ assertThat(keyValueStore.get("C"), is("C->null!"));
+ }
+
+ @Test
+ public void shouldCalculateCorrectOldValuesIfMaterializedEvenIfStateful() {
+ builder
+ .table(INPUT_TOPIC, CONSUMED)
+ .transformValues(
+ new StatefulTransformerSupplier(),
+ Materialized.<String, Integer, KeyValueStore<Bytes,
byte[]>>as(QUERYABLE_NAME)
+ .withKeySerde(Serdes.String())
+ .withValueSerde(Serdes.Integer()))
+ .groupBy(toForceSendingOfOldValues(),
Serialized.with(Serdes.String(), Serdes.Integer()))
+ .reduce(MockReducer.INTEGER_ADDER, MockReducer.INTEGER_SUBTRACTOR)
+ .mapValues(mapBackToStrings())
+ .toStream()
+ .process(capture);
+
+ driver = new TopologyTestDriver(builder.build(), props());
+
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "A", "ignore", 0L));
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "A", "ignored",
0L));
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "A", "ignored",
0L));
+
+ assertThat(output(), hasItems("A:1", "A:0", "A:2", "A:0", "A:3"));
+
+ final KeyValueStore<String, Integer> keyValueStore =
driver.getKeyValueStore(QUERYABLE_NAME);
+ assertThat(keyValueStore.get("A"), is(3));
+ }
+
+ @Test
+ public void
shouldCalculateCorrectOldValuesIfNotStatefulEvenIfNotMaterialized() {
+ builder
+ .table(INPUT_TOPIC, CONSUMED)
+ .transformValues(new StatelessTransformerSupplier())
+ .groupBy(toForceSendingOfOldValues(),
Serialized.with(Serdes.String(), Serdes.Integer()))
+ .reduce(MockReducer.INTEGER_ADDER, MockReducer.INTEGER_SUBTRACTOR)
+ .mapValues(mapBackToStrings())
+ .toStream()
+ .process(capture);
+
+ driver = new TopologyTestDriver(builder.build(), props());
+
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "A", "a", 0L));
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "A", "aa", 0L));
+ driver.pipeInput(recordFactory.create(INPUT_TOPIC, "A", "aaa", 0L));
+
+ assertThat(output(), hasItems("A:1", "A:0", "A:2", "A:0", "A:3"));
+ }
+
+ private ArrayList<String> output() {
+ return capture.capturedProcessors(1).get(0).processed;
+ }
+
+ private static KeyValueMapper<String, Integer, KeyValue<String, Integer>>
toForceSendingOfOldValues() {
+ return new KeyValueMapper<String, Integer, KeyValue<String,
Integer>>() {
+ @Override
+ public KeyValue<String, Integer> apply(String key, Integer value) {
+ return new KeyValue<>(key, value);
+ }
+ };
+ }
+
+ private static ValueMapper<Integer, String> mapBackToStrings() {
+ return new ValueMapper<Integer, String>() {
+ @Override
+ public String apply(Integer value) {
+ return value.toString();
+ }
+ };
+ }
+
+ private static StoreBuilder<KeyValueStore<Long, Long>> storeBuilder(final
String storeName) {
+ return
Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore(storeName),
Serdes.Long(), Serdes.Long());
+ }
+
+ public static Properties props() {
+ final Properties props = new Properties();
+ props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG,
"kstream-transform-values-test");
+ props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9091");
+ props.setProperty(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getAbsolutePath());
+ props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.Integer().getClass().getName());
+ props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.Integer().getClass().getName());
+ return props;
+ }
+
+ private static void throwIfStoresNotAvailable(final ProcessorContext
context,
+ final List<String>
expectedStoredNames) {
+ final List<String> missing = new ArrayList<>();
+
+ for (final String storedName : expectedStoredNames) {
+ if (context.getStateStore(storedName) == null) {
+ missing.add(storedName);
+ }
+ }
+
+ if (!missing.isEmpty()) {
+ throw new AssertionError("State stores are not accessible: " +
missing);
+ }
+ }
+
+ public static class ExclamationValueTransformerSupplier implements
ValueTransformerWithKeySupplier<Object, String, String> {
+ private final List<String> expectedStoredNames;
+
+ ExclamationValueTransformerSupplier(final String...
expectedStoreNames) {
+ this.expectedStoredNames = Arrays.asList(expectedStoreNames);
+ }
+
+ @Override
+ public ExclamationValueTransformer get() {
+ return new ExclamationValueTransformer(expectedStoredNames);
+ }
+ }
+
+ public static class ExclamationValueTransformer implements
ValueTransformerWithKey<Object, String, String> {
+ private final List<String> expectedStoredNames;
+
+ ExclamationValueTransformer(final List<String> expectedStoredNames) {
+ this.expectedStoredNames = expectedStoredNames;
+ }
+
+ @Override
+ public void init(final ProcessorContext context) {
+ throwIfStoresNotAvailable(context, expectedStoredNames);
+ }
+
+ @Override
+ public String transform(final Object readOnlyKey, final String value) {
+ return readOnlyKey.toString() + "->" + value + "!";
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ private static class NullSupplier implements
ValueTransformerWithKeySupplier<String, String, String> {
+ @Override
+ public ValueTransformerWithKey<String, String, String> get() {
+ return null;
+ }
+ }
+
+ private static class StatefulTransformerSupplier implements
ValueTransformerWithKeySupplier<String, String, Integer> {
+ @Override
+ public ValueTransformerWithKey<String, String, Integer> get() {
+ return new StatefulTransformer();
+ }
+ }
+
+ private static class StatefulTransformer implements
ValueTransformerWithKey<String, String, Integer> {
+ private int counter;
+
+ @Override
+ public void init(final ProcessorContext context) {
+ }
+
+ @Override
+ public Integer transform(final String readOnlyKey, final String value)
{
+ return ++counter;
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ private static class StatelessTransformerSupplier implements
ValueTransformerWithKeySupplier<String, String, Integer> {
+ @Override
+ public ValueTransformerWithKey<String, String, Integer> get() {
+ return new StatelessTransformer();
+ }
+ }
+
+ private static class StatelessTransformer implements
ValueTransformerWithKey<String, String, Integer> {
+ @Override
+ public void init(final ProcessorContext context) {
+ }
+
+ @Override
+ public Integer transform(final String readOnlyKey, final String value)
{
+ return value == null ? null : value.length();
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+}
\ No newline at end of file
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ForwardingDisabledProcessorContextTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ForwardingDisabledProcessorContextTest.java
new file mode 100644
index 00000000000..03e79b77c89
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ForwardingDisabledProcessorContextTest.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.To;
+import org.easymock.EasyMockRunner;
+import org.easymock.Mock;
+import org.easymock.MockType;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(EasyMockRunner.class)
+public class ForwardingDisabledProcessorContextTest {
+ @Mock(MockType.NICE)
+ private ProcessorContext delegate;
+ private ForwardingDisabledProcessorContext context;
+
+ @Before
+ public void setUp() {
+ context = new ForwardingDisabledProcessorContext(delegate);
+ }
+
+ @Test(expected = StreamsException.class)
+ public void shouldThrowOnForward() {
+ context.forward("key", "value");
+ }
+
+ @Test(expected = StreamsException.class)
+ public void shouldThrowOnForwardWithTo() {
+ context.forward("key", "value", To.all());
+ }
+
+ @Test(expected = StreamsException.class)
+ public void shouldThrowOnForwardWithChildIndex() {
+ context.forward("key", "value", 1);
+ }
+
+ @Test(expected = StreamsException.class)
+ public void shouldThrowOnForwardWithChildName() {
+ context.forward("key", "value", "child1");
+ }
+}
\ No newline at end of file
diff --git
a/streams/src/test/java/org/apache/kafka/test/KTableValueGetterStub.java
b/streams/src/test/java/org/apache/kafka/test/KTableValueGetterStub.java
index 0efb8d92585..f73975512e0 100644
--- a/streams/src/test/java/org/apache/kafka/test/KTableValueGetterStub.java
+++ b/streams/src/test/java/org/apache/kafka/test/KTableValueGetterStub.java
@@ -43,4 +43,8 @@ public void put(final K key, V value) {
public void remove(final K key) {
data.remove(key);
}
+
+ @Override
+ public void close() {
+ }
}
diff --git
a/streams/src/test/java/org/apache/kafka/test/SingletonNoOpValueTransformer.java
b/streams/src/test/java/org/apache/kafka/test/SingletonNoOpValueTransformer.java
new file mode 100644
index 00000000000..1a99e0092fa
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/test/SingletonNoOpValueTransformer.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.kafka.test;
+
+import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
+import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
+import org.apache.kafka.streams.processor.ProcessorContext;
+
+public class SingletonNoOpValueTransformer<K, V> implements
ValueTransformerWithKeySupplier<K, V, V> {
+ public ProcessorContext context;
+ private final ValueTransformerWithKey<K, V, V> transformer = new
ValueTransformerWithKey<K, V, V>() {
+
+ @Override
+ public void init(final ProcessorContext context) {
+ SingletonNoOpValueTransformer.this.context = context;
+ }
+
+ @Override
+ public V transform(final K readOnlyKey, final V value) {
+ return value;
+ }
+
+ @Override
+ public void close() {
+ }
+ };
+
+ @Override
+ public ValueTransformerWithKey<K, V, V> get() {
+ return transformer;
+ }
+}
\ No newline at end of file
diff --git
a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KStream.scala
b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KStream.scala
index e3e84701413..4b0dc2b89ab 100644
---
a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KStream.scala
+++
b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KStream.scala
@@ -320,9 +320,9 @@ class KStream[K, V](val inner: KStreamJ[K, V]) {
* @return a [[KStream]] that contains records with unmodified key and new
values (possibly of different type)
* @see `org.apache.kafka.streams.kstream.KStream#transformValues`
*/
- def transformValues[VR](valueTransformerWithKeySupplier:
ValueTransformerWithKeySupplier[K, V, VR],
+ def transformValues[VR](valueTransformerSupplier:
ValueTransformerWithKeySupplier[K, V, VR],
stateStoreNames: String*): KStream[K, VR] = {
- inner.transformValues[VR](valueTransformerWithKeySupplier,
stateStoreNames: _*)
+ inner.transformValues[VR](valueTransformerSupplier, stateStoreNames: _*)
}
/**
diff --git
a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KTable.scala
b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KTable.scala
index 218063e3d4c..65cf895750f 100644
---
a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KTable.scala
+++
b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KTable.scala
@@ -20,9 +20,11 @@
package org.apache.kafka.streams.scala
package kstream
+import org.apache.kafka.common.utils.Bytes
import org.apache.kafka.streams.kstream.{KTable => KTableJ, _}
import org.apache.kafka.streams.scala.ImplicitConversions._
import org.apache.kafka.streams.scala.FunctionConversions._
+import org.apache.kafka.streams.state.KeyValueStore
/**
* Wraps the Java class [[org.apache.kafka.streams.kstream.KTable]] and
delegates method calls to the underlying Java object.
@@ -162,6 +164,59 @@ class KTable[K, V](val inner: KTableJ[K, V]) {
def toStream[KR](mapper: (K, V) => KR): KStream[KR, V] =
inner.toStream[KR](mapper.asKeyValueMapper)
+ /**
+ * Create a new `KTable` by transforming the value of each record in this
`KTable` into a new value, (with possibly new type).
+ * Transform the value of each input record into a new value (with possible
new type) of the output record.
+ * A `ValueTransformerWithKey` (provided by the given
`ValueTransformerWithKeySupplier`) is applied to each input
+ * record value and computes a new value for it.
+ * This is similar to `#mapValues(ValueMapperWithKey)`, but more flexible,
allowing access to additional state-stores,
+ * and to the `ProcessorContext`.
+ * If the downstream topology uses aggregation functions, (e.g.
`KGroupedTable#reduce`, `KGroupedTable#aggregate`, etc),
+ * care must be taken when dealing with state, (either held in state-stores
or transformer instances), to ensure correct
+ * aggregate results.
+ * In contrast, if the resulting KTable is materialized, (cf.
`#transformValues(ValueTransformerWithKeySupplier, Materialized, String...)`),
+ * such concerns are handled for you.
+ * In order to assign a state, the state must be created and registered
+ * beforehand via stores added via `addStateStore` or `addGlobalStore`
before they can be connected to the `Transformer`
+ *
+ * @param valueTransformerWithKeySupplier a instance of
`ValueTransformerWithKeySupplier` that generates a `ValueTransformerWithKey`.
+ * At least one transformer instance will
be created per streaming task.
+ * Transformer implementations doe not need
to be thread-safe.
+ * @param stateStoreNames the names of the state stores used by
the processor
+ * @return a [[KStream]] that contains records with unmodified key and new
values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KStream#transformValues`
+ */
+ def transformValues[VR](valueTransformerWithKeySupplier:
ValueTransformerWithKeySupplier[K, V, VR],
+ stateStoreNames: String*): KTable[K, VR] = {
+ inner.transformValues[VR](valueTransformerWithKeySupplier,
stateStoreNames: _*)
+ }
+
+ /**
+ * Create a new `KTable` by transforming the value of each record in this
`KTable` into a new value, (with possibly new type).
+ * A `ValueTransformer` (provided by the given `ValueTransformerSupplier`)
is applied to each input
+ * record value and computes a new value for it.
+ * This is similar to `#mapValues(ValueMapperWithKey)`, but more flexible,
allowing stateful, rather than stateless,
+ * record-by-record operation, access to additional state-stores, and
access to the `ProcessorContext`.
+ * In order to assign a state, the state must be created and registered
+ * beforehand via stores added via `addStateStore` or `addGlobalStore`
before they can be connected to the `Transformer`
+ * The resulting `KTable` is materialized into another state store
(additional to the provided state store names)
+ * as specified by the user via `Materialized` parameter, and is queryable
through its given name.
+ *
+ * @param valueTransformerWithKeySupplier a instance of
`ValueTransformerWithKeySupplier` that generates a `ValueTransformerWithKey`
+ * At least one transformer instance will
be created per streaming task.
+ * Transformer implementations doe not need
to be thread-safe.
+ * @param materialized an instance of `Materialized` used to
describe how the state store of the
+ * resulting table should be materialized.
+ * @param stateStoreNames the names of the state stores used by
the processor
+ * @return a [[KStream]] that contains records with unmodified key and new
values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KStream#transformValues`
+ */
+ def transformValues[VR](valueTransformerWithKeySupplier:
ValueTransformerWithKeySupplier[K, V, VR],
+ materialized: Materialized[K, VR,
KeyValueStore[Bytes, Array[Byte]]],
+ stateStoreNames: String*): KTable[K, VR] = {
+ inner.transformValues[VR](valueTransformerWithKeySupplier, materialized,
stateStoreNames: _*)
+ }
+
/**
* Re-groups the records of this [[KTable]] using the provided key/value
mapper
* and `Serde`s as specified by `Serialized`.
@@ -268,3 +323,4 @@ class KTable[K, V](val inner: KTableJ[K, V]) {
*/
def queryableStoreName: String = inner.queryableStoreName
}
+
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Add transformValues() method to KTable
> ---------------------------------------
>
> Key: KAFKA-6849
> URL: https://issues.apache.org/jira/browse/KAFKA-6849
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Reporter: Andy Coates
> Assignee: Andy Coates
> Priority: Major
> Labels: kip
>
> Add {{transformValues()}} methods to the {{KTable}} interface with the same
> semantics as the functions of the same name on the {{KStream}} interface.
>
> More details in
> [KIP-292|https://cwiki.apache.org/confluence/display/KAFKA/KIP-292%3A+Add+transformValues%28%29+method+to+KTable].
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
