Skip to content

KAFKA-6687: restrict DSL to allow only Streams from the same source topics #9609

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ableegoldman merged 4 commits into from
Dec 1, 2020
Merged

KAFKA-6687: restrict DSL to allow only Streams from the same source topics #9609

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0b52cc3
refactor nodes, fix up tests
ableegoldman Nov 18, 2020
d4a3147
review feedback
ableegoldman Nov 24, 2020
93518cd
rename StreamsGraphNode to GraphNode
ableegoldman Nov 24, 2020
b1d995a
fix indentation and Set-ify
ableegoldman Nov 25, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/AbstractStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
import org.apache.kafka.streams.kstream.ValueTransformerSupplier;
import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
import org.apache.kafka.streams.kstream.internals.graph.GraphNode;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
import org.apache.kafka.streams.state.StoreBuilder;
Expand All @@ -49,7 +49,7 @@ public abstract class AbstractStream<K, V> {
protected final Serde<K> keySerde;
protected final Serde<V> valueSerde;
protected final Set<String> subTopologySourceNodes;
protected final StreamsGraphNode streamsGraphNode;
protected final GraphNode graphNode;
protected final InternalStreamsBuilder builder;

// This copy-constructor will allow to extend KStream
Expand All @@ -60,14 +60,14 @@ public AbstractStream(final AbstractStream<K, V> stream) {
this.keySerde = stream.keySerde;
this.valueSerde = stream.valueSerde;
this.subTopologySourceNodes = stream.subTopologySourceNodes;
this.streamsGraphNode = stream.streamsGraphNode;
this.graphNode = stream.graphNode;
}

AbstractStream(final String name,
final Serde<K> keySerde,
final Serde<V> valueSerde,
final Set<String> subTopologySourceNodes,
final StreamsGraphNode streamsGraphNode,
final GraphNode graphNode,
final InternalStreamsBuilder builder) {
if (subTopologySourceNodes == null || subTopologySourceNodes.isEmpty()) {
throw new IllegalArgumentException("parameter <sourceNodes> must not be null or empty");
Expand All @@ -78,7 +78,7 @@ public AbstractStream(final AbstractStream<K, V> stream) {
this.keySerde = keySerde;
this.valueSerde = valueSerde;
this.subTopologySourceNodes = subTopologySourceNodes;
this.streamsGraphNode = streamsGraphNode;
this.graphNode = graphNode;
}

// This method allows to expose the InternalTopologyBuilder instance
Expand Down
12 changes: 6 additions & 6 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedKStreamImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
import org.apache.kafka.streams.kstream.TimeWindowedCogroupedKStream;
import org.apache.kafka.streams.kstream.Window;
import org.apache.kafka.streams.kstream.Windows;
import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
import org.apache.kafka.streams.kstream.internals.graph.GraphNode;
import org.apache.kafka.streams.state.KeyValueStore;

import java.util.LinkedHashMap;
Expand All @@ -48,9 +48,9 @@ public class CogroupedKStreamImpl<K, VOut> extends AbstractStream<K, VOut> imple

CogroupedKStreamImpl(final String name,
final Set<String> subTopologySourceNodes,
final StreamsGraphNode streamsGraphNode,
final GraphNode graphNode,
final InternalStreamsBuilder builder) {
super(name, null, null, subTopologySourceNodes, streamsGraphNode, builder);
super(name, null, null, subTopologySourceNodes, graphNode, builder);
groupPatterns = new LinkedHashMap<>();
aggregateBuilder = new CogroupedStreamAggregateBuilder<>(builder);
}
Expand Down Expand Up @@ -104,7 +104,7 @@ public <W extends Window> TimeWindowedCogroupedKStream<K, VOut> windowedBy(final
subTopologySourceNodes,
name,
aggregateBuilder,
streamsGraphNode,
graphNode,
groupPatterns);
}

Expand All @@ -117,7 +117,7 @@ public TimeWindowedCogroupedKStream<K, VOut> windowedBy(final SlidingWindows sli
subTopologySourceNodes,
name,
aggregateBuilder,
streamsGraphNode,
graphNode,
groupPatterns);
}

Expand All @@ -129,7 +129,7 @@ public SessionWindowedCogroupedKStream<K, VOut> windowedBy(final SessionWindows
subTopologySourceNodes,
name,
aggregateBuilder,
streamsGraphNode,
graphNode,
groupPatterns);
}

Expand Down
20 changes: 10 additions & 10 deletions ...main/java/org/apache/kafka/streams/kstream/internals/CogroupedStreamAggregateBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,13 +37,13 @@
import org.apache.kafka.streams.kstream.internals.graph.ProcessorGraphNode;
import org.apache.kafka.streams.kstream.internals.graph.ProcessorParameters;
import org.apache.kafka.streams.kstream.internals.graph.StatefulProcessorNode;
import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
import org.apache.kafka.streams.kstream.internals.graph.GraphNode;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.state.StoreBuilder;

class CogroupedStreamAggregateBuilder<K, VOut> {
private final InternalStreamsBuilder builder;
private final Map<KGroupedStreamImpl<K, ?>, StreamsGraphNode> parentNodes = new LinkedHashMap<>();
private final Map<KGroupedStreamImpl<K, ?>, GraphNode> parentNodes = new LinkedHashMap<>();

CogroupedStreamAggregateBuilder(final InternalStreamsBuilder builder) {
this.builder = builder;
Expand All @@ -56,7 +56,7 @@ <KR> KTable<KR, VOut> build(final Map<KGroupedStreamImpl<K, ?>, Aggregator<? sup
final Serde<VOut> valueSerde,
final String queryableName) {
processRepartitions(groupPatterns, storeBuilder);
final Collection<StreamsGraphNode> processors = new ArrayList<>();
final Collection<GraphNode> processors = new ArrayList<>();
boolean stateCreated = false;
int counter = 0;
for (final Entry<KGroupedStreamImpl<K, ?>, Aggregator<? super K, Object, VOut>> kGroupedStream : groupPatterns.entrySet()) {
Expand Down Expand Up @@ -85,7 +85,7 @@ <KR, W extends Window> KTable<KR, VOut> build(final Map<KGroupedStreamImpl<K, ?>
final Windows<W> windows) {
processRepartitions(groupPatterns, storeBuilder);

final Collection<StreamsGraphNode> processors = new ArrayList<>();
final Collection<GraphNode> processors = new ArrayList<>();
boolean stateCreated = false;
int counter = 0;
for (final Entry<KGroupedStreamImpl<K, ?>, Aggregator<? super K, Object, VOut>> kGroupedStream : groupPatterns.entrySet()) {
Expand Down Expand Up @@ -114,7 +114,7 @@ <KR> KTable<KR, VOut> build(final Map<KGroupedStreamImpl<K, ?>, Aggregator<? sup
final SessionWindows sessionWindows,
final Merger<? super K, VOut> sessionMerger) {
processRepartitions(groupPatterns, storeBuilder);
final Collection<StreamsGraphNode> processors = new ArrayList<>();
final Collection<GraphNode> processors = new ArrayList<>();
boolean stateCreated = false;
int counter = 0;
for (final Entry<KGroupedStreamImpl<K, ?>, Aggregator<? super K, Object, VOut>> kGroupedStream : groupPatterns.entrySet()) {
Expand Down Expand Up @@ -142,7 +142,7 @@ <KR> KTable<KR, VOut> build(final Map<KGroupedStreamImpl<K, ?>, Aggregator<? sup
final String queryableName,
final SlidingWindows slidingWindows) {
processRepartitions(groupPatterns, storeBuilder);
final Collection<StreamsGraphNode> processors = new ArrayList<>();
final Collection<GraphNode> processors = new ArrayList<>();
boolean stateCreated = false;
int counter = 0;
for (final Entry<KGroupedStreamImpl<K, ?>, Aggregator<? super K, Object, VOut>> kGroupedStream : groupPatterns.entrySet()) {
Expand Down Expand Up @@ -175,12 +175,12 @@ private void processRepartitions(final Map<KGroupedStreamImpl<K, ?>, Aggregator<
createRepartitionSource(repartionNamePrefix, repartitionNodeBuilder, repartitionReqs.keySerde, repartitionReqs.valueSerde);

if (!parentNodes.containsKey(repartitionReqs)) {
final StreamsGraphNode repartitionNode = repartitionNodeBuilder.build();
builder.addGraphNode(repartitionReqs.streamsGraphNode, repartitionNode);
final GraphNode repartitionNode = repartitionNodeBuilder.build();
builder.addGraphNode(repartitionReqs.graphNode, repartitionNode);
parentNodes.put(repartitionReqs, repartitionNode);
}
} else {
parentNodes.put(repartitionReqs, repartitionReqs.streamsGraphNode);
parentNodes.put(repartitionReqs, repartitionReqs.graphNode);
}
}

Expand All @@ -191,7 +191,7 @@ private void processRepartitions(final Map<KGroupedStreamImpl<K, ?>, Aggregator<

}

<KR, VIn> KTable<KR, VOut> createTable(final Collection<StreamsGraphNode> processors,
<KR, VIn> KTable<KR, VOut> createTable(final Collection<GraphNode> processors,
final NamedInternal named,
final Serde<KR> keySerde,
final Serde<VOut> valueSerde,
Expand Down
12 changes: 6 additions & 6 deletions ...c/main/java/org/apache/kafka/streams/kstream/internals/GroupedStreamAggregateBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.internals.graph.ProcessorParameters;
import org.apache.kafka.streams.kstream.internals.graph.StatefulProcessorNode;
import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
import org.apache.kafka.streams.kstream.internals.graph.GraphNode;
import org.apache.kafka.streams.state.StoreBuilder;

import java.util.Collections;
Expand All @@ -40,8 +40,8 @@ class GroupedStreamAggregateBuilder<K, V> {
private final String userProvidedRepartitionTopicName;
private final Set<String> subTopologySourceNodes;
private final String name;
private final StreamsGraphNode streamsGraphNode;
private StreamsGraphNode repartitionNode;
private final GraphNode graphNode;
private GraphNode repartitionNode;

final Initializer<Long> countInitializer = () -> 0L;

Expand All @@ -54,15 +54,15 @@ class GroupedStreamAggregateBuilder<K, V> {
final boolean repartitionRequired,
final Set<String> subTopologySourceNodes,
final String name,
final StreamsGraphNode streamsGraphNode) {
final GraphNode graphNode) {

this.builder = builder;
this.keySerde = groupedInternal.keySerde();
this.valueSerde = groupedInternal.valueSerde();
this.repartitionRequired = repartitionRequired;
this.subTopologySourceNodes = subTopologySourceNodes;
this.name = name;
this.streamsGraphNode = streamsGraphNode;
this.graphNode = graphNode;
this.userProvidedRepartitionTopicName = groupedInternal.name();
}

Expand All @@ -77,7 +77,7 @@ <KR, VR> KTable<KR, VR> build(final NamedInternal functionName,
final String aggFunctionName = functionName.name();

String sourceName = this.name;
StreamsGraphNode parentNode = streamsGraphNode;
GraphNode parentNode = graphNode;

if (repartitionRequired) {
final OptimizableRepartitionNodeBuilder<K, V> repartitionNodeBuilder = optimizableRepartitionNodeBuilder();
Expand Down
Loading