From 20fd108f36231de06fa0d94889d20c35e6efdb0c Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Fri, 31 Jan 2020 14:32:17 -0800 Subject: [PATCH 1/3] KAFKA-7658: Follow up to original PR - add more unit tests - fix bug for auto-repartitioning - some code cleanup --- .../apache/kafka/streams/StreamsBuilder.java | 5 +- .../apache/kafka/streams/kstream/KStream.java | 56 +++++++- .../kstream/internals/AbstractStream.java | 14 +- .../internals/CogroupedKStreamImpl.java | 29 ++-- .../GroupedStreamAggregateBuilder.java | 15 +- .../kstream/internals/KGroupedStreamImpl.java | 17 +-- .../kstream/internals/KGroupedTableImpl.java | 4 +- .../kstream/internals/KStreamImpl.java | 121 +++++++--------- .../kstream/internals/KStreamImplJoin.java | 10 +- .../streams/kstream/internals/KTableImpl.java | 39 ++--- .../SessionWindowedCogroupedKStreamImpl.java | 13 +- .../internals/SessionWindowedKStreamImpl.java | 4 +- .../TimeWindowedCogroupedKStreamImpl.java | 13 +- .../internals/TimeWindowedKStreamImpl.java | 4 +- .../kstream/internals/AbstractStreamTest.java | 6 +- .../kstream/internals/KStreamImplTest.java | 136 +++++++++--------- 16 files changed, 265 insertions(+), 221 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java b/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java index 8d650d2dcefa5..9cda5113985cf 100644 --- a/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java +++ b/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java @@ -289,8 +289,9 @@ public synchronized KTable table(final String topic, final MaterializedInternal> materializedInternal = new MaterializedInternal<>( - Materialized.with(consumedInternal.keySerde(), consumedInternal.valueSerde()), - internalStreamsBuilder, topic + "-"); + Materialized.with(consumedInternal.keySerde(), consumedInternal.valueSerde()), + internalStreamsBuilder, + topic + "-"); return internalStreamsBuilder.table(topic, consumedInternal, materializedInternal); } 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 782d99a2a6c69..a0017f4b7d20d 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 @@ -879,11 +879,23 @@ void to(final TopicNameExtractor topicExtractor, /** * Convert this stream to a {@link KTable}. *

- * an internal repartitioning topic may need to be created in Kafka if a key changed + * If a key changing operator was used before this operation (e.g., {@link #selectKey(KeyValueMapper)}, + * {@link #map(KeyValueMapper)}, {@link #flatMap(KeyValueMapper)}, or + * {@link #transform(TransformerSupplier, String...)}), and no data redistribution happened afterwards (e.g., via + * {@link #through(String)}) an internal repartitioning topic will be created in Kafka. * This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, * "<name>" is an internally generated name, and "-repartition" is a fixed suffix. *

+ * You can retrieve all generated internal topic names via {@link Topology#describe()}. + *

+ * For this case, all data of this stream will be redistributed through the repartitioning topic by writing all + * records to it, and rereading all records from it, such that the resulting {@link KTable} is partitioned + * correctly on its key. + * Note that you cannot enable {@link StreamsConfig#TOPOLOGY_OPTIMIZATION} config for this case, because + * repartition topics are considered transient and don't allow to recover the result {@link KTable} in cause of + * a failure; hence, a dedicated changelog topic is required to guarantee fault-tolerance. + *

* Note that this is a logical operation and only changes the "interpretation" of the stream, i.e., each record of * it was a "fact/event" and is re-interpreted as update now (cf. {@link KStream} vs {@code KTable}). * @@ -894,11 +906,23 @@ void to(final TopicNameExtractor topicExtractor, /** * Convert this stream to a {@link KTable}. *

- * an internal repartitioning topic may need to be created in Kafka if a key changed + * If a key changing operator was used before this operation (e.g., {@link #selectKey(KeyValueMapper)}, + * {@link #map(KeyValueMapper)}, {@link #flatMap(KeyValueMapper)}, or + * {@link #transform(TransformerSupplier, String...)}), and no data redistribution happened afterwards (e.g., via + * {@link #through(String)}) an internal repartitioning topic will be created in Kafka. * This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, * "<name>" is an internally generated name, and "-repartition" is a fixed suffix. *

+ * You can retrieve all generated internal topic names via {@link Topology#describe()}. + *

+ * For this case, all data of this stream will be redistributed through the repartitioning topic by writing all + * records to it, and rereading all records from it, such that the resulting {@link KTable} is partitioned + * correctly on its key. + * Note that you cannot enable {@link StreamsConfig#TOPOLOGY_OPTIMIZATION} config for this case, because + * repartition topics are considered transient and don't allow to recover the result {@link KTable} in cause of + * a failure; hence, a dedicated changelog topic is required to guarantee fault-tolerance. + *

* Note that this is a logical operation and only changes the "interpretation" of the stream, i.e., each record of * it was a "fact/event" and is re-interpreted as update now (cf. {@link KStream} vs {@code KTable}). * @@ -910,11 +934,23 @@ void to(final TopicNameExtractor topicExtractor, /** * Convert this stream to a {@link KTable}. *

- * an internal repartitioning topic may need to be created in Kafka if a key changed + * If a key changing operator was used before this operation (e.g., {@link #selectKey(KeyValueMapper)}, + * {@link #map(KeyValueMapper)}, {@link #flatMap(KeyValueMapper)}, or + * {@link #transform(TransformerSupplier, String...)}), and no data redistribution happened afterwards (e.g., via + * {@link #through(String)}) an internal repartitioning topic will be created in Kafka. * This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, * "<name>" is an internally generated name, and "-repartition" is a fixed suffix. *

+ * You can retrieve all generated internal topic names via {@link Topology#describe()}. + *

+ * For this case, all data of this stream will be redistributed through the repartitioning topic by writing all + * records to it, and rereading all records from it, such that the resulting {@link KTable} is partitioned + * correctly on its key. + * Note that you cannot enable {@link StreamsConfig#TOPOLOGY_OPTIMIZATION} config for this case, because + * repartition topics are considered transient and don't allow to recover the result {@link KTable} in cause of + * a failure; hence, a dedicated changelog topic is required to guarantee fault-tolerance. + *

* Note that this is a logical operation and only changes the "interpretation" of the stream, i.e., each record of * it was a "fact/event" and is re-interpreted as update now (cf. {@link KStream} vs {@code KTable}). * @@ -927,11 +963,23 @@ void to(final TopicNameExtractor topicExtractor, /** * Convert this stream to a {@link KTable}. *

- * an internal repartitioning topic may need to be created in Kafka if a key changed + * If a key changing operator was used before this operation (e.g., {@link #selectKey(KeyValueMapper)}, + * {@link #map(KeyValueMapper)}, {@link #flatMap(KeyValueMapper)}, or + * {@link #transform(TransformerSupplier, String...)}), and no data redistribution happened afterwards (e.g., via + * {@link #through(String)}) an internal repartitioning topic will be created in Kafka. * This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, * "<name>" is an internally generated name, and "-repartition" is a fixed suffix. *

+ * You can retrieve all generated internal topic names via {@link Topology#describe()}. + *

+ * For this case, all data of this stream will be redistributed through the repartitioning topic by writing all + * records to it, and rereading all records from it, such that the resulting {@link KTable} is partitioned + * correctly on its key. + * Note that you cannot enable {@link StreamsConfig#TOPOLOGY_OPTIMIZATION} config for this case, because + * repartition topics are considered transient and don't allow to recover the result {@link KTable} in cause of + * a failure; hence, a dedicated changelog topic is required to guarantee fault-tolerance. + *

* Note that this is a logical operation and only changes the "interpretation" of the stream, i.e., each record of * it was a "fact/event" and is re-interpreted as update now (cf. {@link KStream} vs {@code KTable}). * diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java index 0670da4920a67..3c6b591e1927d 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java @@ -46,7 +46,7 @@ public abstract class AbstractStream { protected final String name; protected final Serde keySerde; protected final Serde valSerde; - protected final Set sourceNodes; + protected final Set subTopologySourceNodes; protected final StreamsGraphNode streamsGraphNode; protected final InternalStreamsBuilder builder; @@ -57,17 +57,17 @@ public AbstractStream(final AbstractStream stream) { this.builder = stream.builder; this.keySerde = stream.keySerde; this.valSerde = stream.valSerde; - this.sourceNodes = stream.sourceNodes; + this.subTopologySourceNodes = stream.subTopologySourceNodes; this.streamsGraphNode = stream.streamsGraphNode; } AbstractStream(final String name, final Serde keySerde, final Serde valSerde, - final Set sourceNodes, + final Set subTopologySourceNodes, final StreamsGraphNode streamsGraphNode, final InternalStreamsBuilder builder) { - if (sourceNodes == null || sourceNodes.isEmpty()) { + if (subTopologySourceNodes == null || subTopologySourceNodes.isEmpty()) { throw new IllegalArgumentException("parameter must not be null or empty"); } @@ -75,7 +75,7 @@ public AbstractStream(final AbstractStream stream) { this.builder = builder; this.keySerde = keySerde; this.valSerde = valSerde; - this.sourceNodes = sourceNodes; + this.subTopologySourceNodes = subTopologySourceNodes; this.streamsGraphNode = streamsGraphNode; } @@ -86,9 +86,9 @@ protected InternalTopologyBuilder internalTopologyBuilder() { } Set ensureCopartitionWith(final Collection> otherStreams) { - final Set allSourceNodes = new HashSet<>(sourceNodes); + final Set allSourceNodes = new HashSet<>(subTopologySourceNodes); for (final AbstractStream other: otherStreams) { - allSourceNodes.addAll(other.sourceNodes); + allSourceNodes.addAll(other.subTopologySourceNodes); } builder.internalTopologyBuilder.copartitionSources(allSourceNodes); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImpl.java index 2eabcbf4a428f..41ddcddb037b8 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImpl.java @@ -16,10 +16,6 @@ */ package org.apache.kafka.streams.kstream.internals; -import java.util.LinkedHashMap; -import java.util.Map; -import java.util.Objects; -import java.util.Set; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.kstream.Aggregator; import org.apache.kafka.streams.kstream.CogroupedKStream; @@ -36,6 +32,11 @@ import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode; import org.apache.kafka.streams.state.KeyValueStore; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + public class CogroupedKStreamImpl extends AbstractStream implements CogroupedKStream { static final String AGGREGATE_NAME = "COGROUPKSTREAM-AGGREGATE-"; @@ -45,10 +46,10 @@ public class CogroupedKStreamImpl extends AbstractStream imple final private CogroupedStreamAggregateBuilder aggregateBuilder; CogroupedKStreamImpl(final String name, - final Set sourceNodes, + final Set subTopologySourceNodes, final StreamsGraphNode streamsGraphNode, final InternalStreamsBuilder builder) { - super(name, null, null, sourceNodes, streamsGraphNode, builder); + super(name, null, null, subTopologySourceNodes, streamsGraphNode, builder); groupPatterns = new LinkedHashMap<>(); aggregateBuilder = new CogroupedStreamAggregateBuilder<>(builder); } @@ -97,13 +98,13 @@ public KTable aggregate(final Initializer initializer) { public TimeWindowedCogroupedKStream windowedBy(final Windows windows) { Objects.requireNonNull(windows, "windows can't be null"); return new TimeWindowedCogroupedKStreamImpl<>( - windows, - builder, - sourceNodes, - name, - aggregateBuilder, - streamsGraphNode, - groupPatterns); + windows, + builder, + subTopologySourceNodes, + name, + aggregateBuilder, + streamsGraphNode, + groupPatterns); } @Override @@ -111,7 +112,7 @@ public SessionWindowedCogroupedKStream windowedBy(final SessionWindows Objects.requireNonNull(sessionWindows, "sessionWindows can't be null"); return new SessionWindowedCogroupedKStreamImpl<>(sessionWindows, builder, - sourceNodes, + subTopologySourceNodes, name, aggregateBuilder, streamsGraphNode, diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/GroupedStreamAggregateBuilder.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/GroupedStreamAggregateBuilder.java index b77a2303d9d29..13e7218275a62 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/GroupedStreamAggregateBuilder.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/GroupedStreamAggregateBuilder.java @@ -26,13 +26,12 @@ import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.state.StoreBuilder; -import static org.apache.kafka.streams.kstream.internals.graph.OptimizableRepartitionNode.OptimizableRepartitionNodeBuilder; -import static org.apache.kafka.streams.kstream.internals.graph.OptimizableRepartitionNode.optimizableRepartitionNodeBuilder; - - import java.util.Collections; import java.util.Set; +import static org.apache.kafka.streams.kstream.internals.graph.OptimizableRepartitionNode.OptimizableRepartitionNodeBuilder; +import static org.apache.kafka.streams.kstream.internals.graph.OptimizableRepartitionNode.optimizableRepartitionNodeBuilder; + class GroupedStreamAggregateBuilder { private final InternalStreamsBuilder builder; @@ -40,7 +39,7 @@ class GroupedStreamAggregateBuilder { private final Serde valueSerde; private final boolean repartitionRequired; private final String userProvidedRepartitionTopicName; - private final Set sourceNodes; + private final Set subTopologySourceNodes; private final String name; private final StreamsGraphNode streamsGraphNode; private StreamsGraphNode repartitionNode; @@ -54,7 +53,7 @@ class GroupedStreamAggregateBuilder { GroupedStreamAggregateBuilder(final InternalStreamsBuilder builder, final GroupedInternal groupedInternal, final boolean repartitionRequired, - final Set sourceNodes, + final Set subTopologySourceNodes, final String name, final StreamsGraphNode streamsGraphNode) { @@ -62,7 +61,7 @@ class GroupedStreamAggregateBuilder { this.keySerde = groupedInternal.keySerde(); this.valueSerde = groupedInternal.valueSerde(); this.repartitionRequired = repartitionRequired; - this.sourceNodes = sourceNodes; + this.subTopologySourceNodes = subTopologySourceNodes; this.name = name; this.streamsGraphNode = streamsGraphNode; this.userProvidedRepartitionTopicName = groupedInternal.name(); @@ -110,7 +109,7 @@ KTable build(final NamedInternal functionName, return new KTableImpl<>(aggFunctionName, keySerde, valSerde, - sourceName.equals(this.name) ? sourceNodes : Collections.singleton(sourceName), + sourceName.equals(this.name) ? subTopologySourceNodes : Collections.singleton(sourceName), queryableStoreName, aggregateSupplier, statefulProcessorNode, diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImpl.java index 66c44ac884b4b..bd8586aabeab2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImpl.java @@ -16,8 +16,6 @@ */ package org.apache.kafka.streams.kstream.internals; -import java.util.Objects; -import java.util.Set; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.kstream.Aggregator; @@ -36,6 +34,9 @@ import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode; import org.apache.kafka.streams.state.KeyValueStore; +import java.util.Objects; +import java.util.Set; + class KGroupedStreamImpl extends AbstractStream implements KGroupedStream { static final String REDUCE_NAME = "KSTREAM-REDUCE-"; @@ -46,19 +47,19 @@ class KGroupedStreamImpl extends AbstractStream implements KGroupedS final String userProvidedRepartitionTopicName; KGroupedStreamImpl(final String name, - final Set sourceNodes, + final Set subTopologySourceNodes, final GroupedInternal groupedInternal, final boolean repartitionRequired, final StreamsGraphNode streamsGraphNode, final InternalStreamsBuilder builder) { - super(name, groupedInternal.keySerde(), groupedInternal.valueSerde(), sourceNodes, streamsGraphNode, builder); + super(name, groupedInternal.keySerde(), groupedInternal.valueSerde(), subTopologySourceNodes, streamsGraphNode, builder); this.repartitionRequired = repartitionRequired; this.userProvidedRepartitionTopicName = groupedInternal.name(); this.aggregateBuilder = new GroupedStreamAggregateBuilder<>( builder, groupedInternal, repartitionRequired, - sourceNodes, + subTopologySourceNodes, name, streamsGraphNode ); @@ -192,7 +193,7 @@ public TimeWindowedKStream windowedBy(final Windows return new TimeWindowedKStreamImpl<>( windows, builder, - sourceNodes, + subTopologySourceNodes, name, keySerde, valSerde, @@ -207,7 +208,7 @@ public SessionWindowedKStream windowedBy(final SessionWindows windows) { return new SessionWindowedKStreamImpl<>( windows, builder, - sourceNodes, + subTopologySourceNodes, name, keySerde, valSerde, @@ -231,7 +232,7 @@ private KTable doAggregate(final KStreamAggProcessorSupplier CogroupedKStream cogroup(final Aggregator aggregator) { Objects.requireNonNull(aggregator, "aggregator can't be null"); - return new CogroupedKStreamImpl(name, sourceNodes, streamsGraphNode, builder) + return new CogroupedKStreamImpl(name, subTopologySourceNodes, streamsGraphNode, builder) .cogroup(this, aggregator); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java index e3d03a5bc2e59..e710b4b4bea3e 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedTableImpl.java @@ -60,10 +60,10 @@ public class KGroupedTableImpl extends AbstractStream implements KGr KGroupedTableImpl(final InternalStreamsBuilder builder, final String name, - final Set sourceNodes, + final Set subTopologySourceNodes, final GroupedInternal groupedInternal, final StreamsGraphNode streamsGraphNode) { - super(name, groupedInternal.keySerde(), groupedInternal.valueSerde(), sourceNodes, streamsGraphNode, builder); + super(name, groupedInternal.keySerde(), groupedInternal.valueSerde(), subTopologySourceNodes, streamsGraphNode, builder); this.userProvidedRepartitionTopicName = groupedInternal.name(); } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java index 33f017436f5c0..308aafed21948 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java @@ -19,7 +19,6 @@ import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.KeyValue; -import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.ForeachAction; import org.apache.kafka.streams.kstream.GlobalKTable; import org.apache.kafka.streams.kstream.Grouped; @@ -124,11 +123,11 @@ public class KStreamImpl extends AbstractStream implements KStream keySerde, final Serde valueSerde, - final Set sourceNodes, + final Set subTopologySourceNodes, final boolean repartitionRequired, final StreamsGraphNode streamsGraphNode, final InternalStreamsBuilder builder) { - super(name, keySerde, valueSerde, sourceNodes, streamsGraphNode, builder); + super(name, keySerde, valueSerde, subTopologySourceNodes, streamsGraphNode, builder); this.repartitionRequired = repartitionRequired; } @@ -152,13 +151,13 @@ public KStream filter(final Predicate predicate, builder.addGraphNode(streamsGraphNode, filterProcessorNode); return new KStreamImpl<>( - name, - keySerde, - valSerde, - sourceNodes, - repartitionRequired, - filterProcessorNode, - builder); + name, + keySerde, + valSerde, + subTopologySourceNodes, + repartitionRequired, + filterProcessorNode, + builder); } @Override @@ -181,13 +180,13 @@ public KStream filterNot(final Predicate predicate, builder.addGraphNode(streamsGraphNode, filterNotProcessorNode); return new KStreamImpl<>( - name, - keySerde, - valSerde, - sourceNodes, - repartitionRequired, - filterNotProcessorNode, - builder); + name, + keySerde, + valSerde, + subTopologySourceNodes, + repartitionRequired, + filterNotProcessorNode, + builder); } @Override @@ -211,7 +210,7 @@ public KStream selectKey(final KeyValueMapper KStream map(final KeyValueMapper( - name, - null, - null, - sourceNodes, - true, - mapProcessorNode, - builder); + name, + null, + null, + subTopologySourceNodes, + true, + mapProcessorNode, + builder); } @Override @@ -291,13 +290,13 @@ public KStream mapValues(final ValueMapperWithKey( - name, - keySerde, - null, - sourceNodes, - repartitionRequired, - mapValuesProcessorNode, - builder); + name, + keySerde, + null, + subTopologySourceNodes, + repartitionRequired, + mapValuesProcessorNode, + builder); } @Override @@ -321,7 +320,7 @@ public KStream flatMap(final KeyValueMapper(name, null, null, sourceNodes, true, flatMapNode, builder); + return new KStreamImpl<>(name, null, null, subTopologySourceNodes, true, flatMapNode, builder); } @Override @@ -360,7 +359,7 @@ public KStream flatMapValues(final ValueMapperWithKey peek(final ForeachAction action, name, keySerde, valSerde, - sourceNodes, + subTopologySourceNodes, repartitionRequired, peekNode, builder); @@ -475,7 +474,7 @@ private KStream[] doBranch(final NamedInternal named, new ProcessorGraphNode<>(childNames[i], innerProcessorParameters); builder.addGraphNode(branchNode, branchChildNode); - branchChildren[i] = new KStreamImpl<>(childNames[i], keySerde, valSerde, sourceNodes, repartitionRequired, branchChildNode, builder); + branchChildren[i] = new KStreamImpl<>(childNames[i], keySerde, valSerde, subTopologySourceNodes, repartitionRequired, branchChildNode, builder); } return branchChildren; @@ -501,9 +500,9 @@ private KStream merge(final InternalStreamsBuilder builder, final KStreamImpl streamImpl = (KStreamImpl) stream; final boolean requireRepartitioning = streamImpl.repartitionRequired || repartitionRequired; final String name = named.orElseGenerateWithPrefix(builder, MERGE_NAME); - final Set allSourceNodes = new HashSet<>(); - allSourceNodes.addAll(sourceNodes); - allSourceNodes.addAll(streamImpl.sourceNodes); + final Set allSubTopologySourceNodes = new HashSet<>(); + allSubTopologySourceNodes.addAll(subTopologySourceNodes); + allSubTopologySourceNodes.addAll(streamImpl.subTopologySourceNodes); final ProcessorParameters processorParameters = new ProcessorParameters<>(new PassThrough<>(), name); @@ -518,7 +517,7 @@ private KStream merge(final InternalStreamsBuilder builder, name, null, null, - allSourceNodes, + allSubTopologySourceNodes, requireRepartitioning, mergeNode, builder); @@ -611,20 +610,12 @@ private void to(final TopicNameExtractor topicExtractor, @Override public KTable toTable() { - return toTable(NamedInternal.empty()); + return toTable(NamedInternal.empty(), Materialized.with(keySerde, valSerde)); } @Override public KTable toTable(final Named named) { - final ConsumedInternal consumedInternal = new ConsumedInternal<>(Consumed.with(keySerde, valSerde)); - - final MaterializedInternal> materializedInternal = - new MaterializedInternal<>( - Materialized.with(consumedInternal.keySerde(), consumedInternal.valueSerde()), - builder, - TO_KTABLE_NAME); - - return doToTable(named, materializedInternal); + return toTable(named, Materialized.with(keySerde, valSerde)); } @Override @@ -635,18 +626,14 @@ public KTable toTable(final Materialized toTable(final Named named, final Materialized> materialized) { + Objects.requireNonNull(named, "named can't be null"); Objects.requireNonNull(materialized, "materialized can't be null"); - final MaterializedInternal> materializedInternal = - new MaterializedInternal<>(materialized); - return doToTable( - named, - materializedInternal); - } + final NamedInternal namedInternal = new NamedInternal(named); + final String name = namedInternal.orElseGenerateWithPrefix(builder, TO_KTABLE_NAME); - private KTable doToTable(final Named named, - final MaterializedInternal> materializedInternal) { - Objects.requireNonNull(named, "named can't be null"); + final MaterializedInternal> materializedInternal = + new MaterializedInternal<>(materialized, builder, TO_KTABLE_NAME); final Serde keySerdeOverride = materializedInternal.keySerde() == null ? keySerde @@ -655,8 +642,6 @@ private KTable doToTable(final Named named, ? valSerde : materializedInternal.valueSerde(); - final NamedInternal namedInternal = new NamedInternal(named); - final String name = namedInternal.orElseGenerateWithPrefix(builder, TO_KTABLE_NAME); final Set subTopologySourceNodes; final StreamsGraphNode tableParentNode; @@ -675,7 +660,7 @@ private KTable doToTable(final Named named, subTopologySourceNodes = Collections.singleton(sourceName); } else { tableParentNode = streamsGraphNode; - subTopologySourceNodes = sourceNodes; + subTopologySourceNodes = this.subTopologySourceNodes; } final KTableSource tableSource = new KTableSource<>( @@ -734,7 +719,7 @@ public KGroupedStream groupBy(final KeyValueMapper( selectKeyMapNode.nodeName(), - sourceNodes, + subTopologySourceNodes, groupedInternal, true, selectKeyMapNode, @@ -763,7 +748,7 @@ public KGroupedStream groupByKey(final Grouped grouped) { return new KGroupedStreamImpl<>( name, - sourceNodes, + subTopologySourceNodes, groupedInternal, repartitionRequired, streamsGraphNode, @@ -1118,7 +1103,7 @@ private KStream globalTableJoin(final GlobalKTable g name, keySerde, null, - sourceNodes, + subTopologySourceNodes, repartitionRequired, streamTableJoinNode, builder); @@ -1213,7 +1198,7 @@ public KStream flatTransform(final TransformerSupplier KStream doTransformValues(final ValueTransformerWithKeySuppl name, keySerde, null, - sourceNodes, + subTopologySourceNodes, repartitionRequired, transformNode, builder); @@ -1343,7 +1328,7 @@ private KStream doFlatTransformValues(final ValueTransformerWithKeyS name, keySerde, null, - sourceNodes, + subTopologySourceNodes, repartitionRequired, transformNode, builder); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImplJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImplJoin.java index 9affbe9a143eb..923ca04341476 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImplJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImplJoin.java @@ -77,8 +77,8 @@ public KStream join(final KStream lhs, final String joinMergeName = renamed.suffixWithOrElseGet( "-merge", builder, KStreamImpl.MERGE_NAME); - final StreamsGraphNode thisStreamsGraphNode = ((AbstractStream) lhs).streamsGraphNode; - final StreamsGraphNode otherStreamsGraphNode = ((AbstractStream) other).streamsGraphNode; + final StreamsGraphNode thisStreamsGraphNode = ((AbstractStream) lhs).streamsGraphNode; + final StreamsGraphNode otherStreamsGraphNode = ((AbstractStream) other).streamsGraphNode; final StoreBuilder> thisWindowStore; final StoreBuilder> otherWindowStore; @@ -155,8 +155,8 @@ public KStream join(final KStream lhs, builder.addGraphNode(Arrays.asList(thisStreamsGraphNode, otherStreamsGraphNode), joinGraphNode); - final Set allSourceNodes = new HashSet<>(((KStreamImpl) lhs).sourceNodes); - allSourceNodes.addAll(((KStreamImpl) other).sourceNodes); + final Set allSourceNodes = new HashSet<>(((KStreamImpl) lhs).subTopologySourceNodes); + allSourceNodes.addAll(((KStreamImpl) other).subTopologySourceNodes); // do not have serde for joined result; // also for key serde we do not inherit from either since we cannot tell if these two serdes are different @@ -184,7 +184,6 @@ private void assertUniqueStoreNames(final WindowBytesStoreSupplier supplier, } } - @SuppressWarnings("deprecation") // continuing to support Windows#maintainMs/segmentInterval in fallback mode private static StoreBuilder> joinWindowStoreBuilder(final String storeName, final JoinWindows windows, final Serde keySerde, @@ -201,7 +200,6 @@ private static StoreBuilder> joinWindowStoreBuilder(fin ); } - @SuppressWarnings("deprecation") // continuing to support Windows#maintainMs/segmentInterval in fallback mode private static StoreBuilder> joinWindowStoreBuilderFromSupplier(final WindowBytesStoreSupplier storeSupplier, final Serde keySerde, final Serde valueSerde) { 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 052a5f8445c4b..38e5c49198fd4 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 @@ -130,12 +130,12 @@ public class KTableImpl extends AbstractStream implements KTable< public KTableImpl(final String name, final Serde keySerde, final Serde valSerde, - final Set sourceNodes, + final Set subTopologySourceNodes, final String queryableStoreName, final ProcessorSupplier processorSupplier, final StreamsGraphNode streamsGraphNode, final InternalStreamsBuilder builder) { - super(name, keySerde, valSerde, sourceNodes, streamsGraphNode, builder); + super(name, keySerde, valSerde, subTopologySourceNodes, streamsGraphNode, builder); this.processorSupplier = processorSupplier; this.queryableStoreName = queryableStoreName; } @@ -191,14 +191,15 @@ private KTable doFilter(final Predicate predicate, builder.addGraphNode(this.streamsGraphNode, tableNode); - return new KTableImpl<>(name, - keySerde, - valueSerde, - sourceNodes, - queryableStoreName, - processorSupplier, - tableNode, - builder); + return new KTableImpl<>( + name, + keySerde, + valueSerde, + subTopologySourceNodes, + queryableStoreName, + processorSupplier, + tableNode, + builder); } @Override @@ -310,7 +311,7 @@ private KTable doMapValues(final ValueMapperWithKey KTable doTransformValues(final ValueTransformerWithKeySuppli name, keySerde, valueSerde, - sourceNodes, + subTopologySourceNodes, queryableStoreName, processorSupplier, tableNode, @@ -493,7 +494,7 @@ public KStream toStream(final Named named) { builder.addGraphNode(this.streamsGraphNode, toStreamNode); // we can inherit parent key and value serde - return new KStreamImpl<>(name, keySerde, valSerde, sourceNodes, false, toStreamNode, builder); + return new KStreamImpl<>(name, keySerde, valSerde, subTopologySourceNodes, false, toStreamNode, builder); } @Override @@ -683,7 +684,7 @@ private KTable doJoin(final KTable other, enableSendingOldValues(); } if (rightOuter) { - ((KTableImpl) other).enableSendingOldValues(); + ((KTableImpl) other).enableSendingOldValues(); } final KTableKTableAbstractJoin joinThis; @@ -732,9 +733,9 @@ private KTable doJoin(final KTable other, .withJoinThisProcessorParameters(joinThisProcessorParameters) .withJoinOtherProcessorParameters(joinOtherProcessorParameters) .withThisJoinSideNodeName(name) - .withOtherJoinSideNodeName(((KTableImpl) other).name) + .withOtherJoinSideNodeName(((KTableImpl) other).name) .withJoinThisStoreNames(valueGetterSupplier().storeNames()) - .withJoinOtherStoreNames(((KTableImpl) other).valueGetterSupplier().storeNames()) + .withJoinOtherStoreNames(((KTableImpl) other).valueGetterSupplier().storeNames()) .withKeySerde(keySerde) .withValueSerde(valueSerde) .withQueryableStoreName(queryableStoreName) @@ -790,7 +791,7 @@ public KGroupedTable groupBy(final KeyValueMapper( builder, selectName, - sourceNodes, + subTopologySourceNodes, groupedInternal, groupByMapNode ); @@ -995,7 +996,7 @@ private KTable doJoinOnForeignKey(final KTable forei // The subscription source is the source node on the *receiving* end *after* the repartition. // This topic needs to be copartitioned with the Foreign Key table. final Set copartitionedRepartitionSources = - new HashSet<>(((KTableImpl) foreignKeyTable).sourceNodes); + new HashSet<>(((KTableImpl) foreignKeyTable).subTopologySourceNodes); copartitionedRepartitionSources.add(subscriptionSource.nodeName()); builder.internalTopologyBuilder.copartitionSources(copartitionedRepartitionSources); @@ -1065,7 +1066,7 @@ private KTable doJoinOnForeignKey(final KTable forei builder.addGraphNode(foreignResponseSink, foreignResponseSource); // the response topic has to be copartitioned with the left (primary) side of the join - final Set resultSourceNodes = new HashSet<>(this.sourceNodes); + final Set resultSourceNodes = new HashSet<>(this.subTopologySourceNodes); resultSourceNodes.add(foreignResponseSource.nodeName()); builder.internalTopologyBuilder.copartitionSources(resultSourceNodes); diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImpl.java index 5b05e38ee058c..f755a743f9a37 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImpl.java @@ -17,10 +17,6 @@ package org.apache.kafka.streams.kstream.internals; -import java.time.Duration; -import java.util.Map; -import java.util.Objects; -import java.util.Set; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.kstream.Aggregator; import org.apache.kafka.streams.kstream.Initializer; @@ -38,6 +34,11 @@ import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; +import java.time.Duration; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + public class SessionWindowedCogroupedKStreamImpl extends AbstractStream implements SessionWindowedCogroupedKStream { @@ -47,12 +48,12 @@ public class SessionWindowedCogroupedKStreamImpl extends SessionWindowedCogroupedKStreamImpl(final SessionWindows sessionWindows, final InternalStreamsBuilder builder, - final Set sourceNodes, + final Set subTopologySourceNodes, final String name, final CogroupedStreamAggregateBuilder aggregateBuilder, final StreamsGraphNode streamsGraphNode, final Map, Aggregator> groupPatterns) { - super(name, null, null, sourceNodes, streamsGraphNode, builder); + super(name, null, null, subTopologySourceNodes, streamsGraphNode, builder); //keySerde and valueSerde are null because there are many different groupStreams that they could be from this.sessionWindows = sessionWindows; this.aggregateBuilder = aggregateBuilder; diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImpl.java index ac49f9187cf76..a0ddefcb3c9cc 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImpl.java @@ -50,13 +50,13 @@ public class SessionWindowedKStreamImpl extends AbstractStream imple SessionWindowedKStreamImpl(final SessionWindows windows, final InternalStreamsBuilder builder, - final Set sourceNodes, + final Set subTopologySourceNodes, final String name, final Serde keySerde, final Serde valSerde, final GroupedStreamAggregateBuilder aggregateBuilder, final StreamsGraphNode streamsGraphNode) { - super(name, keySerde, valSerde, sourceNodes, streamsGraphNode, builder); + super(name, keySerde, valSerde, subTopologySourceNodes, streamsGraphNode, builder); Objects.requireNonNull(windows, "windows can't be null"); this.windows = windows; this.aggregateBuilder = aggregateBuilder; diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedCogroupedKStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedCogroupedKStreamImpl.java index 04a0b498b6018..5cb76783a48f9 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedCogroupedKStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedCogroupedKStreamImpl.java @@ -17,10 +17,6 @@ package org.apache.kafka.streams.kstream.internals; -import java.time.Duration; -import java.util.Map; -import java.util.Objects; -import java.util.Set; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.kstream.Aggregator; import org.apache.kafka.streams.kstream.Initializer; @@ -39,6 +35,11 @@ import org.apache.kafka.streams.state.WindowStore; import org.apache.kafka.streams.state.internals.RocksDbWindowBytesStoreSupplier; +import java.time.Duration; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + public class TimeWindowedCogroupedKStreamImpl extends AbstractStream implements TimeWindowedCogroupedKStream { @@ -48,12 +49,12 @@ public class TimeWindowedCogroupedKStreamImpl extends Ab TimeWindowedCogroupedKStreamImpl(final Windows windows, final InternalStreamsBuilder builder, - final Set sourceNodes, + final Set subTopologySourceNodes, final String name, final CogroupedStreamAggregateBuilder aggregateBuilder, final StreamsGraphNode streamsGraphNode, final Map, Aggregator> groupPatterns) { - super(name, null, null, sourceNodes, streamsGraphNode, builder); + super(name, null, null, subTopologySourceNodes, streamsGraphNode, builder); //keySerde and valueSerde are null because there are many different groupStreams that they could be from this.windows = windows; this.aggregateBuilder = aggregateBuilder; diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java index 84316cbe9ccf4..67b4ab98789a4 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java @@ -51,13 +51,13 @@ public class TimeWindowedKStreamImpl extends AbstractStr TimeWindowedKStreamImpl(final Windows windows, final InternalStreamsBuilder builder, - final Set sourceNodes, + final Set subTopologySourceNodes, final String name, final Serde keySerde, final Serde valSerde, final GroupedStreamAggregateBuilder aggregateBuilder, final StreamsGraphNode streamsGraphNode) { - super(name, keySerde, valSerde, sourceNodes, streamsGraphNode, builder); + super(name, keySerde, valSerde, subTopologySourceNodes, streamsGraphNode, builder); this.windows = Objects.requireNonNull(windows, "windows can't be null"); this.aggregateBuilder = aggregateBuilder; } diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/AbstractStreamTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/AbstractStreamTest.java index f78eebbe7a231..f421ed8ffff7c 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/AbstractStreamTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/AbstractStreamTest.java @@ -97,7 +97,7 @@ public void testShouldBeExtensible() { assertTrue(supplier.theCapturedProcessor().processed.size() <= expectedKeys.length); } - private class ExtendedKStream extends AbstractStream { + private static class ExtendedKStream extends AbstractStream { ExtendedKStream(final KStream stream) { super((KStreamImpl) stream); @@ -109,11 +109,11 @@ KStream randomFilter() { name, new ProcessorParameters<>(new ExtendedKStreamDummy<>(), name)); builder.addGraphNode(this.streamsGraphNode, processorNode); - return new KStreamImpl<>(name, null, null, sourceNodes, false, processorNode, builder); + return new KStreamImpl<>(name, null, null, subTopologySourceNodes, false, processorNode, builder); } } - private class ExtendedKStreamDummy implements ProcessorSupplier { + private static class ExtendedKStreamDummy implements ProcessorSupplier { private final Random rand; diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java index 7cfe510814124..40aba07f1f767 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java @@ -24,7 +24,6 @@ import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.KeyValueTimestamp; import org.apache.kafka.streams.StreamsBuilder; -import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.TestInputTopic; import org.apache.kafka.streams.TestOutputTopic; import org.apache.kafka.streams.Topology; @@ -65,7 +64,6 @@ import org.apache.kafka.test.MockProcessorSupplier; import org.apache.kafka.test.MockValueJoiner; import org.apache.kafka.test.StreamsTestUtils; -import org.apache.kafka.test.TestUtils; import org.junit.Before; import org.junit.Test; @@ -73,13 +71,11 @@ import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; -import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.function.Function; import java.util.regex.Matcher; @@ -90,7 +86,6 @@ import static java.util.Collections.emptyMap; import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; -import static org.apache.kafka.common.utils.Utils.mkProperties; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; @@ -692,6 +687,38 @@ public void shouldNotAllowNullGroupedOnGroupByKey() { assertThat(exception.getMessage(), equalTo("grouped can't be null")); } + @Test + public void shouldNotAllowNullNamedOnToTable() { + final NullPointerException exception = assertThrows( + NullPointerException.class, + () -> testStream.toTable((Named) null)); + assertThat(exception.getMessage(), equalTo("named can't be null")); + } + + @Test + public void shouldNotAllowNullMaterializedOnToTable() { + final NullPointerException exception = assertThrows( + NullPointerException.class, + () -> testStream.toTable((Materialized>) null)); + assertThat(exception.getMessage(), equalTo("materialized can't be null")); + } + + @Test + public void shouldNotAllowNullNamedOnToTableWithMaterialized() { + final NullPointerException exception = assertThrows( + NullPointerException.class, + () -> testStream.toTable(null, Materialized.with(null, null))); + assertThat(exception.getMessage(), equalTo("named can't be null")); + } + + @Test + public void shouldNotAllowNullMaterializedOnToTableWithNamed() { + final NullPointerException exception = assertThrows( + NullPointerException.class, + () -> testStream.toTable(Named.as("name"), null)); + assertThat(exception.getMessage(), equalTo("materialized can't be null")); + } + @Test public void shouldNotAllowNullOtherStreamOnJoin() { final NullPointerException exception = assertThrows( @@ -1550,8 +1577,6 @@ public void shouldPropagateRepartitionFlagAfterGlobalKTableJoin() { .groupByKey() .count(); - final String topologyDescription = builder.build().describe().toString(); - final Pattern repartitionTopicPattern = Pattern.compile("Sink: .*-repartition"); final String topology = builder.build().describe().toString(); final Matcher matcher = repartitionTopicPattern.matcher(topology); @@ -1559,7 +1584,6 @@ public void shouldPropagateRepartitionFlagAfterGlobalKTableJoin() { final String match = matcher.group(); assertThat(match, notNullValue()); assertTrue(match.endsWith("repartition")); - } @Test @@ -2420,8 +2444,9 @@ public void shouldMaterializeKTableFromKStream() { builder.stream(input, consumed) .toTable(Materialized.as(Stores.inMemoryKeyValueStore(storeName))); - final String topologyDescription = builder.build().describe().toString(); + final Topology topology = builder.build(); + final String topologyDescription = topology.describe().toString(); assertThat( topologyDescription, equalTo("Topologies:\n" + @@ -2433,13 +2458,15 @@ public void shouldMaterializeKTableFromKStream() { " <-- KSTREAM-SOURCE-0000000000\n\n") ); - try (final TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) { + try (final TopologyTestDriver driver = new TopologyTestDriver(topology, props)) { final TestInputTopic inputTopic = driver.createInputTopic(input, Serdes.String().serializer(), Serdes.String().serializer()); final KeyValueStore store = driver.getKeyValueStore(storeName); inputTopic.pipeInput("A", "01"); - final Map expectedStore = mkMap(mkEntry("A", "01")); + inputTopic.pipeInput("B", "02"); + inputTopic.pipeInput("A", "03"); + final Map expectedStore = mkMap(mkEntry("A", "03"), mkEntry("B", "02")); assertThat(asMap(store), is(expectedStore)); } @@ -2455,14 +2482,13 @@ public void shouldSupportKeyChangeKTableFromKStream() { final String output = "output"; builder.stream(input, consumed) - .map((key, value) -> new KeyValue<>(key + "-", value)) - .toTable() + .map((key, value) -> new KeyValue<>(key.charAt(0) - 'A', value)) + .toTable(Materialized.with(Serdes.Integer(), null)) .toStream().to(output); final Topology topology = builder.build(); final String topologyDescription = topology.describe().toString(); - assertThat( topologyDescription, equalTo("Topologies:\n" + @@ -2475,18 +2501,18 @@ public void shouldSupportKeyChangeKTableFromKStream() { " Processor: KSTREAM-FILTER-0000000005 (stores: [])\n" + " --> KSTREAM-SINK-0000000004\n" + " <-- KSTREAM-MAP-0000000001\n" + - " Sink: KSTREAM-SINK-0000000004 (topic: KSTREAM-TOTABLE-0000000003-repartition)\n" + + " Sink: KSTREAM-SINK-0000000004 (topic: KSTREAM-TOTABLE-0000000002-repartition)\n" + " <-- KSTREAM-FILTER-0000000005\n" + "\n" + " Sub-topology: 1\n" + - " Source: KSTREAM-SOURCE-0000000006 (topics: [KSTREAM-TOTABLE-0000000003-repartition])\n" + - " --> KSTREAM-TOTABLE-0000000003\n" + - " Processor: KSTREAM-TOTABLE-0000000003 (stores: [])\n" + + " Source: KSTREAM-SOURCE-0000000006 (topics: [KSTREAM-TOTABLE-0000000002-repartition])\n" + + " --> KSTREAM-TOTABLE-0000000002\n" + + " Processor: KSTREAM-TOTABLE-0000000002 (stores: [])\n" + " --> KTABLE-TOSTREAM-0000000007\n" + " <-- KSTREAM-SOURCE-0000000006\n" + " Processor: KTABLE-TOSTREAM-0000000007 (stores: [])\n" + " --> KSTREAM-SINK-0000000008\n" + - " <-- KSTREAM-TOTABLE-0000000003\n" + + " <-- KSTREAM-TOTABLE-0000000002\n" + " Sink: KSTREAM-SINK-0000000008 (topic: output)\n" + " <-- KTABLE-TOSTREAM-0000000007\n\n") ); @@ -2494,8 +2520,8 @@ public void shouldSupportKeyChangeKTableFromKStream() { try (final TopologyTestDriver driver = new TopologyTestDriver(topology, props)) { final TestInputTopic inputTopic = driver.createInputTopic(input, Serdes.String().serializer(), Serdes.String().serializer()); - final TestOutputTopic outputTopic = - driver.createOutputTopic(output, Serdes.String().deserializer(), Serdes.String().deserializer()); + final TestOutputTopic outputTopic = + driver.createOutputTopic(output, Serdes.Integer().deserializer(), Serdes.String().deserializer()); inputTopic.pipeInput("A", "01", 5L); inputTopic.pipeInput("B", "02", 100L); @@ -2504,19 +2530,18 @@ public void shouldSupportKeyChangeKTableFromKStream() { inputTopic.pipeInput("A", "05", 10L); inputTopic.pipeInput("A", "06", 8L); - final List> outputExpectRecords = new ArrayList<>(); - outputExpectRecords.add(new TestRecord<>("A-", "01", Instant.ofEpochMilli(5L))); - outputExpectRecords.add(new TestRecord<>("B-", "02", Instant.ofEpochMilli(100L))); - outputExpectRecords.add(new TestRecord<>("C-", "03", Instant.ofEpochMilli(0L))); - outputExpectRecords.add(new TestRecord<>("D-", "04", Instant.ofEpochMilli(0L))); - outputExpectRecords.add(new TestRecord<>("A-", "05", Instant.ofEpochMilli(10L))); - outputExpectRecords.add(new TestRecord<>("A-", "06", Instant.ofEpochMilli(8L))); + final List> outputExpectRecords = new ArrayList<>(); + outputExpectRecords.add(new TestRecord<>(0, "01", Instant.ofEpochMilli(5L))); + outputExpectRecords.add(new TestRecord<>(1, "02", Instant.ofEpochMilli(100L))); + outputExpectRecords.add(new TestRecord<>(2, "03", Instant.ofEpochMilli(0L))); + outputExpectRecords.add(new TestRecord<>(3, "04", Instant.ofEpochMilli(0L))); + outputExpectRecords.add(new TestRecord<>(0, "05", Instant.ofEpochMilli(10L))); + outputExpectRecords.add(new TestRecord<>(0, "06", Instant.ofEpochMilli(8L))); assertEquals(outputTopic.readRecordsToList(), outputExpectRecords); } } - @Test public void shouldSupportForeignKeyTableTableJoinWithKTableFromKStream() { final Consumed consumed = Consumed.with(Serdes.String(), Serdes.String()); @@ -2644,8 +2669,8 @@ public void shouldSupportTableTableJoinWithKStreamToKTable() { table1.join(table2, MockValueJoiner.TOSTRING_JOINER).toStream().to(outputTopic); final Topology topology = builder.build(props); - final String topologyDescription = topology.describe().toString(); + final String topologyDescription = topology.describe().toString(); assertThat( topologyDescription, equalTo("Topologies:\n" + @@ -2675,11 +2700,6 @@ public void shouldSupportTableTableJoinWithKStreamToKTable() { " Sink: KSTREAM-SINK-0000000010 (topic: output)\n" + " <-- KTABLE-TOSTREAM-0000000009\n\n")); - final Collection> copartitionGroups = - TopologyWrapper.getInternalTopologyBuilder(topology).copartitionGroups(); - - assertEquals(1, copartitionGroups.size()); - try (final TopologyTestDriver driver = new TopologyTestDriver(topology, props)) { final TestInputTopic left = driver.createInputTopic(leftTopic, new StringSerializer(), new StringSerializer()); final TestInputTopic right = driver.createInputTopic(rightTopic, new StringSerializer(), new StringSerializer()); @@ -2719,17 +2739,17 @@ public void shouldSupportTableTableJoinWithKStreamToKTable() { mkEntry("lhs1", "lhsValue4+rhsValue1") )) ); - } } @Test public void shouldSupportStreamTableJoinWithKStreamToKTable() { + final StreamsBuilder builder = new StreamsBuilder(); + final Consumed consumed = Consumed.with(Serdes.String(), Serdes.String()); + final String streamTopic = "streamTopic"; final String tableTopic = "tableTopic"; final String outputTopic = "output"; - final StreamsBuilder builder = new StreamsBuilder(); - final Consumed consumed = Consumed.with(Serdes.String(), Serdes.String()); final KStream stream = builder.stream(streamTopic, consumed); final KTable table = builder.stream(tableTopic, consumed).toTable(); @@ -2737,8 +2757,8 @@ public void shouldSupportStreamTableJoinWithKStreamToKTable() { stream.join(table, MockValueJoiner.TOSTRING_JOINER).to(outputTopic); final Topology topology = builder.build(props); - final String topologyDescription = topology.describe().toString(); + final String topologyDescription = topology.describe().toString(); assertThat( topologyDescription, equalTo("Topologies:\n" + @@ -2756,11 +2776,6 @@ public void shouldSupportStreamTableJoinWithKStreamToKTable() { " --> none\n" + " <-- KSTREAM-SOURCE-0000000001\n\n")); - final Collection> copartitionGroups = - TopologyWrapper.getInternalTopologyBuilder(topology).copartitionGroups(); - - assertEquals(1, copartitionGroups.size()); - try (final TopologyTestDriver driver = new TopologyTestDriver(topology, props)) { final TestInputTopic left = driver.createInputTopic(streamTopic, new StringSerializer(), new StringSerializer()); final TestInputTopic right = driver.createInputTopic(tableTopic, new StringSerializer(), new StringSerializer()); @@ -2800,7 +2815,6 @@ public void shouldSupportStreamTableJoinWithKStreamToKTable() { mkEntry("lhs1", "lhsValue4+rhsValue1") )) ); - } } @@ -2808,6 +2822,7 @@ public void shouldSupportStreamTableJoinWithKStreamToKTable() { public void shouldSupportGroupByCountWithKStreamToKTable() { final Consumed consumed = Consumed.with(Serdes.String(), Serdes.String()); final StreamsBuilder builder = new StreamsBuilder(); + final String input = "input"; final String output = "output"; @@ -2820,8 +2835,8 @@ public void shouldSupportGroupByCountWithKStreamToKTable() { .to(output); final Topology topology = builder.build(props); - final String topologyDescription = topology.describe().toString(); + final String topologyDescription = topology.describe().toString(); assertThat( topologyDescription, equalTo("Topologies:\n" + @@ -2856,7 +2871,6 @@ public void shouldSupportGroupByCountWithKStreamToKTable() { final TestOutputTopic outputTopic = driver.createOutputTopic(output, Serdes.String().deserializer(), Serdes.Long().deserializer()); - inputTopic.pipeInput("A", "green", 10L); inputTopic.pipeInput("B", "green", 9L); inputTopic.pipeInput("A", "blue", 12L); @@ -2879,6 +2893,7 @@ public void shouldSupportGroupByCountWithKStreamToKTable() { public void shouldSupportTriggerMaterializedWithKTableFromKStream() { final Consumed consumed = Consumed.with(Serdes.String(), Serdes.String()); final StreamsBuilder builder = new StreamsBuilder(); + final String input = "input"; final String output = "output"; final String storeName = "store"; @@ -2894,20 +2909,20 @@ public void shouldSupportTriggerMaterializedWithKTableFromKStream() { .to(output); final Topology topology = builder.build(props); - final String topologyDescription = topology.describe().toString(); + final String topologyDescription = topology.describe().toString(); assertThat( topologyDescription, equalTo("Topologies:\n" + " Sub-topology: 0\n" + " Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" + - " --> KSTREAM-TOTABLE-0000000002\n" + - " Processor: KSTREAM-TOTABLE-0000000002 (stores: [])\n" + + " --> KSTREAM-TOTABLE-0000000001\n" + + " Processor: KSTREAM-TOTABLE-0000000001 (stores: [])\n" + " --> KTABLE-MAPVALUES-0000000003\n" + " <-- KSTREAM-SOURCE-0000000000\n" + " Processor: KTABLE-MAPVALUES-0000000003 (stores: [store])\n" + " --> KTABLE-TOSTREAM-0000000004\n" + - " <-- KSTREAM-TOTABLE-0000000002\n" + + " <-- KSTREAM-TOTABLE-0000000001\n" + " Processor: KTABLE-TOSTREAM-0000000004 (stores: [])\n" + " --> KSTREAM-SINK-0000000005\n" + " <-- KTABLE-MAPVALUES-0000000003\n" + @@ -2915,19 +2930,12 @@ public void shouldSupportTriggerMaterializedWithKTableFromKStream() { " <-- KTABLE-TOSTREAM-0000000004\n\n")); try ( - final TopologyTestDriver driver = new TopologyTestDriver( - topology, - mkProperties(mkMap( - mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy"), - mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, "test"), - mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory("kafka-test").getAbsolutePath()) - )), - Instant.ofEpochMilli(0L))) { + final TopologyTestDriver driver = new TopologyTestDriver(topology, props)) { final TestInputTopic inputTopic = driver.createInputTopic(input, new StringSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ZERO); - final TestOutputTopic outputTopic = - driver.createOutputTopic(output, Serdes.String().deserializer(), Serdes.String().deserializer()); - final KeyValueStore store = driver.getKeyValueStore(storeName); + final TestOutputTopic outputTopic = + driver.createOutputTopic(output, Serdes.String().deserializer(), Serdes.Integer().deserializer()); + final KeyValueStore store = driver.getKeyValueStore(storeName); inputTopic.pipeInput("A", "green", 10L); inputTopic.pipeInput("B", "green", 9L); @@ -2955,8 +2963,8 @@ public void shouldSupportTriggerMaterializedWithKTableFromKStream() { } } - private static Map asMap(final KeyValueStore store) { - final HashMap result = new HashMap<>(); + private static Map asMap(final KeyValueStore store) { + final HashMap result = new HashMap<>(); store.all().forEachRemaining(kv -> result.put(kv.key, kv.value)); return result; } From a8b67dca4c103c39815e1da092437059033d8784 Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Sat, 1 Feb 2020 11:22:35 -0800 Subject: [PATCH 2/3] Fix failing tests --- .../kstream/internals/KStreamImplTest.java | 50 +++++++++---------- 1 file changed, 25 insertions(+), 25 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java index 40aba07f1f767..0671276ee76ce 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java @@ -2397,13 +2397,13 @@ public void shouldNotMaterializedKTableFromKStream() { equalTo("Topologies:\n" + " Sub-topology: 0\n" + " Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" + - " --> KSTREAM-TOTABLE-0000000002\n" + - " Processor: KSTREAM-TOTABLE-0000000002 (stores: [])\n" + + " --> KSTREAM-TOTABLE-0000000001\n" + + " Processor: KSTREAM-TOTABLE-0000000001 (stores: [])\n" + " --> KTABLE-TOSTREAM-0000000003\n" + " <-- KSTREAM-SOURCE-0000000000\n" + " Processor: KTABLE-TOSTREAM-0000000003 (stores: [])\n" + " --> KSTREAM-SINK-0000000004\n" + - " <-- KSTREAM-TOTABLE-0000000002\n" + + " <-- KSTREAM-TOTABLE-0000000001\n" + " Sink: KSTREAM-SINK-0000000004 (topic: output)\n" + " <-- KTABLE-TOSTREAM-0000000003\n\n") ); @@ -2570,11 +2570,11 @@ public void shouldSupportForeignKeyTableTableJoinWithKTableFromKStream() { " Source: KTABLE-SOURCE-0000000016 (topics: [KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic])\n" + " --> KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-RESOLVER-PROCESSOR-0000000017\n" + " Source: KSTREAM-SOURCE-0000000000 (topics: [input1])\n" + - " --> KSTREAM-TOTABLE-0000000002\n" + - " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-RESOLVER-PROCESSOR-0000000017 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000001])\n" + + " --> KSTREAM-TOTABLE-0000000001\n" + + " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-RESOLVER-PROCESSOR-0000000017 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + " --> KTABLE-FK-JOIN-OUTPUT-0000000018\n" + " <-- KTABLE-SOURCE-0000000016\n" + - " Processor: KSTREAM-TOTABLE-0000000002 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000001])\n" + + " Processor: KSTREAM-TOTABLE-0000000001 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + " --> KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000007\n" + " <-- KSTREAM-SOURCE-0000000000\n" + " Processor: KTABLE-FK-JOIN-OUTPUT-0000000018 (stores: [])\n" + @@ -2582,7 +2582,7 @@ public void shouldSupportForeignKeyTableTableJoinWithKTableFromKStream() { " <-- KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-RESOLVER-PROCESSOR-0000000017\n" + " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000007 (stores: [])\n" + " --> KTABLE-SINK-0000000008\n" + - " <-- KSTREAM-TOTABLE-0000000002\n" + + " <-- KSTREAM-TOTABLE-0000000001\n" + " Processor: KTABLE-TOSTREAM-0000000020 (stores: [])\n" + " --> KSTREAM-SINK-0000000021\n" + " <-- KTABLE-FK-JOIN-OUTPUT-0000000018\n" + @@ -2593,21 +2593,21 @@ public void shouldSupportForeignKeyTableTableJoinWithKTableFromKStream() { "\n" + " Sub-topology: 1\n" + " Source: KSTREAM-SOURCE-0000000003 (topics: [input2])\n" + - " --> KSTREAM-TOTABLE-0000000005\n" + + " --> KSTREAM-TOTABLE-0000000004\n" + " Source: KTABLE-SOURCE-0000000009 (topics: [KTABLE-FK-JOIN-SUBSCRIPTION-REGISTRATION-0000000006-topic])\n" + " --> KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000011\n" + - " Processor: KSTREAM-TOTABLE-0000000005 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000004])\n" + + " Processor: KSTREAM-TOTABLE-0000000004 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005])\n" + " --> KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000013\n" + " <-- KSTREAM-SOURCE-0000000003\n" + " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000011 (stores: [KTABLE-FK-JOIN-SUBSCRIPTION-STATE-STORE-0000000010])\n" + " --> KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000012\n" + " <-- KTABLE-SOURCE-0000000009\n" + - " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000012 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000004])\n" + + " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000012 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005])\n" + " --> KTABLE-SINK-0000000015\n" + " <-- KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000011\n" + " Processor: KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000013 (stores: [KTABLE-FK-JOIN-SUBSCRIPTION-STATE-STORE-0000000010])\n" + " --> KTABLE-SINK-0000000015\n" + - " <-- KSTREAM-TOTABLE-0000000005\n" + + " <-- KSTREAM-TOTABLE-0000000004\n" + " Sink: KTABLE-SINK-0000000015 (topic: KTABLE-FK-JOIN-SUBSCRIPTION-RESPONSE-0000000014-topic)\n" + " <-- KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000012, KTABLE-FK-JOIN-SUBSCRIPTION-PROCESSOR-0000000013\n\n") ); @@ -2676,21 +2676,21 @@ public void shouldSupportTableTableJoinWithKStreamToKTable() { equalTo("Topologies:\n" + " Sub-topology: 0\n" + " Source: KSTREAM-SOURCE-0000000000 (topics: [left])\n" + - " --> KSTREAM-TOTABLE-0000000002\n" + + " --> KSTREAM-TOTABLE-0000000001\n" + " Source: KSTREAM-SOURCE-0000000003 (topics: [right])\n" + - " --> KSTREAM-TOTABLE-0000000005\n" + - " Processor: KSTREAM-TOTABLE-0000000002 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000001])\n" + + " --> KSTREAM-TOTABLE-0000000004\n" + + " Processor: KSTREAM-TOTABLE-0000000001 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + " --> KTABLE-JOINTHIS-0000000007\n" + " <-- KSTREAM-SOURCE-0000000000\n" + - " Processor: KSTREAM-TOTABLE-0000000005 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000004])\n" + + " Processor: KSTREAM-TOTABLE-0000000004 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005])\n" + " --> KTABLE-JOINOTHER-0000000008\n" + " <-- KSTREAM-SOURCE-0000000003\n" + - " Processor: KTABLE-JOINOTHER-0000000008 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000001])\n" + + " Processor: KTABLE-JOINOTHER-0000000008 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + " --> KTABLE-MERGE-0000000006\n" + - " <-- KSTREAM-TOTABLE-0000000005\n" + - " Processor: KTABLE-JOINTHIS-0000000007 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000004])\n" + + " <-- KSTREAM-TOTABLE-0000000004\n" + + " Processor: KTABLE-JOINTHIS-0000000007 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000005])\n" + " --> KTABLE-MERGE-0000000006\n" + - " <-- KSTREAM-TOTABLE-0000000002\n" + + " <-- KSTREAM-TOTABLE-0000000001\n" + " Processor: KTABLE-MERGE-0000000006 (stores: [])\n" + " --> KTABLE-TOSTREAM-0000000009\n" + " <-- KTABLE-JOINTHIS-0000000007, KTABLE-JOINOTHER-0000000008\n" + @@ -2765,14 +2765,14 @@ public void shouldSupportStreamTableJoinWithKStreamToKTable() { " Sub-topology: 0\n" + " Source: KSTREAM-SOURCE-0000000000 (topics: [streamTopic])\n" + " --> KSTREAM-JOIN-0000000004\n" + - " Processor: KSTREAM-JOIN-0000000004 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + + " Processor: KSTREAM-JOIN-0000000004 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000003])\n" + " --> KSTREAM-SINK-0000000005\n" + " <-- KSTREAM-SOURCE-0000000000\n" + " Source: KSTREAM-SOURCE-0000000001 (topics: [tableTopic])\n" + - " --> KSTREAM-TOTABLE-0000000003\n" + + " --> KSTREAM-TOTABLE-0000000002\n" + " Sink: KSTREAM-SINK-0000000005 (topic: output)\n" + " <-- KSTREAM-JOIN-0000000004\n" + - " Processor: KSTREAM-TOTABLE-0000000003 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + + " Processor: KSTREAM-TOTABLE-0000000002 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000003])\n" + " --> none\n" + " <-- KSTREAM-SOURCE-0000000001\n\n")); @@ -2842,13 +2842,13 @@ public void shouldSupportGroupByCountWithKStreamToKTable() { equalTo("Topologies:\n" + " Sub-topology: 0\n" + " Source: KSTREAM-SOURCE-0000000000 (topics: [input])\n" + - " --> KSTREAM-TOTABLE-0000000002\n" + - " Processor: KSTREAM-TOTABLE-0000000002 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000001])\n" + + " --> KSTREAM-TOTABLE-0000000001\n" + + " Processor: KSTREAM-TOTABLE-0000000001 (stores: [KSTREAM-TOTABLE-STATE-STORE-0000000002])\n" + " --> KTABLE-SELECT-0000000003\n" + " <-- KSTREAM-SOURCE-0000000000\n" + " Processor: KTABLE-SELECT-0000000003 (stores: [])\n" + " --> KSTREAM-SINK-0000000005\n" + - " <-- KSTREAM-TOTABLE-0000000002\n" + + " <-- KSTREAM-TOTABLE-0000000001\n" + " Sink: KSTREAM-SINK-0000000005 (topic: KTABLE-AGGREGATE-STATE-STORE-0000000004-repartition)\n" + " <-- KTABLE-SELECT-0000000003\n" + "\n" + From 295398d006e707a60672b3898068cac2049a0f7f Mon Sep 17 00:00:00 2001 From: "Matthias J. Sax" Date: Fri, 7 Feb 2020 19:57:20 -0800 Subject: [PATCH 3/3] Partly revert HOTFIX (https://github.com/apache/kafka/pull/8063) that seems not to be the correct fix --- .../kstream/internals/KStreamImplTest.java | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java index 0671276ee76ce..06675f455df56 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamImplTest.java @@ -2901,10 +2901,10 @@ public void shouldSupportTriggerMaterializedWithKTableFromKStream() { builder.stream(input, consumed) .toTable() .mapValues( - value -> String.valueOf(value.charAt(0) - (int) 'a'), - Materialized.>as(storeName) + value -> value.charAt(0) - (int) 'a', + Materialized.>as(storeName) .withKeySerde(Serdes.String()) - .withValueSerde(Serdes.String())) + .withValueSerde(Serdes.Integer())) .toStream() .to(output); @@ -2943,21 +2943,21 @@ public void shouldSupportTriggerMaterializedWithKTableFromKStream() { inputTopic.pipeInput("C", "yellow", 15L); inputTopic.pipeInput("D", "green", 11L); - final Map expectedStore = new HashMap<>(); - expectedStore.putIfAbsent("A", "1"); - expectedStore.putIfAbsent("B", "6"); - expectedStore.putIfAbsent("C", "24"); - expectedStore.putIfAbsent("D", "6"); + final Map expectedStore = new HashMap<>(); + expectedStore.putIfAbsent("A", 1); + expectedStore.putIfAbsent("B", 6); + expectedStore.putIfAbsent("C", 24); + expectedStore.putIfAbsent("D", 6); assertEquals(expectedStore, asMap(store)); assertEquals( asList( - new TestRecord<>("A", "6", Instant.ofEpochMilli(10)), - new TestRecord<>("B", "6", Instant.ofEpochMilli(9)), - new TestRecord<>("A", "1", Instant.ofEpochMilli(12)), - new TestRecord<>("C", "24", Instant.ofEpochMilli(15)), - new TestRecord<>("D", "6", Instant.ofEpochMilli(11))), + new TestRecord<>("A", 6, Instant.ofEpochMilli(10)), + new TestRecord<>("B", 6, Instant.ofEpochMilli(9)), + new TestRecord<>("A", 1, Instant.ofEpochMilli(12)), + new TestRecord<>("C", 24, Instant.ofEpochMilli(15)), + new TestRecord<>("D", 6, Instant.ofEpochMilli(11))), outputTopic.readRecordsToList()); }