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 1 commit
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
71 changes: 71 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/SourceGraphNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
/*
* 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.graph;

import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.regex.Pattern;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.streams.kstream.internals.ConsumedInternal;

abstract public class SourceGraphNode<K, V> extends StreamsGraphNode {

@cadonna cadonna Nov 23, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we also rename StreamsGraphNode to GraphNode? The Streams prefix is a bit confusing, IMO, because StreamSourceNode and StreamsGraphNode seem really similar although they are quite different.

@ableegoldman ableegoldman Nov 24, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, but don't come crying when the PR blows up in length 😉 (but yeah that makes sense to me)

@cadonna cadonna Nov 25, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have never said you need to do it in this PR 😉 . Jokes apart, I think in general it would be better to do such things in a separate PR, but when I wrote my comment, I completely forgot about it. Sorry about that!


private Collection<String> topicNames;
Comment thread
ableegoldman marked this conversation as resolved.
Outdated
private Pattern topicPattern;
Comment thread
ableegoldman marked this conversation as resolved.
Outdated
private final ConsumedInternal<K, V> consumedInternal;

public SourceGraphNode(final String nodeName,
final Collection<String> topicNames,
Comment thread
ableegoldman marked this conversation as resolved.
Outdated
final ConsumedInternal<K, V> consumedInternal) {
super(nodeName);

this.topicNames = topicNames;
this.consumedInternal = consumedInternal;
}

public SourceGraphNode(final String nodeName,
final Pattern topicPattern,
final ConsumedInternal<K, V> consumedInternal) {

super(nodeName);

this.topicPattern = topicPattern;
this.consumedInternal = consumedInternal;
}

public Set<String> topicNames() {
return new HashSet<>(topicNames);

@cadonna cadonna Nov 23, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we already demand a set of topics in the constructors of SourceGraphNode() and its children?

@ableegoldman ableegoldman Nov 24, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure I understand exactly what you're asking, but I made a few changes to this topic collection/method. Please lmk if it hasn't addressed your question

@cadonna cadonna Nov 25, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I like your changes. What I meant is that we could change the constructor of SourceGraphNode to:

public SourceGraphNode(final String nodeName,
                       final Set<String> topicNames,
                       final ConsumedInternal<K, V> consumedInternal)

and the one of StreamSourceNode to:

public StreamSourceNode(final String nodeName,
                        final Set<String> topicNames,
                        final ConsumedInternal<K, V> consumedInternal)

In this way, we have a set of topics as soon as possible in the code path from the public API. I think this makes it clearer that it is not possible to have duplicates of topics internally.

To keep this PR small, I would propose to just do the changes for SourceGraphNode, and do the other changes in a separate PR.

@ableegoldman ableegoldman Dec 2, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah ok you meant making it a Set vs a Collection -- I do agree with the principle, and I did push the Set-ification of the topics up one level so that the actual class field is a Set. But I don't think it's really worth it to push it up another layer and Set-ify the constructor argument. For one thing we would just have to do the same conversion to a Set but in more places, and more importantly, the actual callers of the constructor don't care at all whether it's a Set or any other Collection. So I think it actually does make sense to convert to a Set inside the constructor body

@cadonna cadonna Dec 2, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fair enough and I think that this is nothing urgent or absolute necessary. However, I would like to explain my line of thoughts. I think an interface of a class should also describe the constraints on the the object and as far as I see it does not make any sense to pass the same topic name multiple times to a source node.
I do not see an issue with making the same conversion in more places and actually this is even not true because the only place we would do a conversion is in StreamsBuilder#stream(). All other dependent calls create a singleton collection which can be easily replaced with a singleton set. Actually, I do not understand why StreamsBuilder#stream() takes a collection instead of a set.

I am not sure I can follow your other argument

the actual callers of the constructor don't care at all whether it's a Set or any other Collection

Do you refer to the creation of the singleton collection in the callers?

As I said, I do not say we need to follow my proposal. I just wanted to argue in favor of a cleaner and more descriptive interface.

@ableegoldman ableegoldman Dec 2, 2020 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

All other dependent calls create a singleton collection which can be easily replaced with a singleton set

Ah ok, I didn't notice that. I guess I only looked at StreamsBuilder#stream

Actually, I do not understand why StreamsBuilder#stream() takes a collection instead of a set.

This I totally agree with. I suspect the intention was just for convenience, so users don't have to do a list->set conversion themselves, but I personally don't find that to be a very strong argument. It doesn't seem worth doing a KIP over, but maybe if we rewrite some large parts of the DSL in the future, we can fix this as well

By "callers" I meant the method body of StreamsBuilder#stream, which doesn't really care whether there are duplicates in the collection because its only job is to pass the topics straight from the user to this source node.

But I see your point. If I touch on some related code in a future PR I can fix this on the side, or I'd be happy to review a PR if you want to submit one. Thanks for the discussion

}

public Pattern topicPattern() {
return topicPattern;
}

public ConsumedInternal<K, V> consumedInternal() {
return consumedInternal;
}

public Serde<K> keySerde() {
return consumedInternal.keySerde();
}

public Serde<V> valueSerde() {
return consumedInternal.valueSerde();
}

}
70 changes: 18 additions & 52 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamSourceNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,6 @@

package org.apache.kafka.streams.kstream.internals.graph;

import java.util.HashSet;
import java.util.Set;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.streams.Topology.AutoOffsetReset;
import org.apache.kafka.streams.errors.TopologyException;
import org.apache.kafka.streams.kstream.internals.ConsumedInternal;
Expand All @@ -30,56 +27,25 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class StreamSourceNode<K, V> extends StreamsGraphNode {
public class StreamSourceNode<K, V> extends SourceGraphNode<K, V> {

private final Logger log = LoggerFactory.getLogger(StreamSourceNode.class);

private Collection<String> topicNames;
private Pattern topicPattern;
private final ConsumedInternal<K, V> consumedInternal;


public StreamSourceNode(final String nodeName,
final Collection<String> topicNames,
final ConsumedInternal<K, V> consumedInternal) {
super(nodeName);

this.topicNames = topicNames;
this.consumedInternal = consumedInternal;
super(nodeName, topicNames, consumedInternal);
}

public StreamSourceNode(final String nodeName,
final Pattern topicPattern,
final ConsumedInternal<K, V> consumedInternal) {

super(nodeName);

this.topicPattern = topicPattern;
this.consumedInternal = consumedInternal;
}

public Set<String> topicNames() {
return new HashSet<>(topicNames);
}

public Pattern topicPattern() {
return topicPattern;
}

public ConsumedInternal<K, V> consumedInternal() {
return consumedInternal;
}

public Serde<K> keySerde() {
return consumedInternal.keySerde();
}

public Serde<V> valueSerde() {
return consumedInternal.valueSerde();
super(nodeName, topicPattern, consumedInternal);
}

public void merge(final StreamSourceNode<?, ?> other) {
final AutoOffsetReset resetPolicy = consumedInternal.offsetResetPolicy();
final AutoOffsetReset resetPolicy = consumedInternal().offsetResetPolicy();
final AutoOffsetReset otherResetPolicy = other.consumedInternal().offsetResetPolicy();
if (resetPolicy != null && !resetPolicy.equals(otherResetPolicy)
|| otherResetPolicy != null && !otherResetPolicy.equals(resetPolicy)) {
Expand All @@ -96,29 +62,29 @@ public void merge(final StreamSourceNode<?, ?> other) {
@Override
public String toString() {
return "StreamSourceNode{" +
"topicNames=" + topicNames +
", topicPattern=" + topicPattern +
", consumedInternal=" + consumedInternal +
"topicNames=" + topicNames() +
", topicPattern=" + topicPattern() +
", consumedInternal=" + consumedInternal() +
"} " + super.toString();
}

@Override
public void writeToTopology(final InternalTopologyBuilder topologyBuilder) {

if (topicPattern != null) {
topologyBuilder.addSource(consumedInternal.offsetResetPolicy(),
if (topicPattern() != null) {
topologyBuilder.addSource(consumedInternal().offsetResetPolicy(),
nodeName(),
consumedInternal.timestampExtractor(),
consumedInternal.keyDeserializer(),
consumedInternal.valueDeserializer(),
topicPattern);
consumedInternal().timestampExtractor(),
consumedInternal().keyDeserializer(),
consumedInternal().valueDeserializer(),
topicPattern());
} else {
topologyBuilder.addSource(consumedInternal.offsetResetPolicy(),
topologyBuilder.addSource(consumedInternal().offsetResetPolicy(),
nodeName(),
consumedInternal.timestampExtractor(),
consumedInternal.keyDeserializer(),
consumedInternal.valueDeserializer(),
topicNames.toArray(new String[topicNames.size()]));
consumedInternal().timestampExtractor(),
consumedInternal().keyDeserializer(),
consumedInternal().valueDeserializer(),
topicNames().toArray(new String[topicNames().size()]));

}
}
Expand Down
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/TableSourceNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
* Used to represent either a KTable source or a GlobalKTable source. A boolean flag is used to indicate if this represents a GlobalKTable a {@link
* org.apache.kafka.streams.kstream.GlobalKTable}
*/
public class TableSourceNode<K, V> extends StreamSourceNode<K, V> {
public class TableSourceNode<K, V> extends SourceGraphNode<K, V> {

private final MaterializedInternal<K, V, ?> materializedInternal;
private final ProcessorParameters<K, V, ?, ?> processorParameters;
Expand Down
28 changes: 14 additions & 14 deletions streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -906,20 +906,6 @@ public void shouldAllowStreamsFromSameTopic() {
assertBuildDoesNotThrow(builder);
}

@Test
public void shouldAllowTablesFromSameTopic() {
builder.table("topic");
builder.table("topic");
assertBuildDoesNotThrow(builder);
}

@Test
public void shouldAllowStreamAndTableFromSameTopic() {
builder.stream("topic");
builder.table("topic");
assertBuildDoesNotThrow(builder);
}

@Test
public void shouldAllowSubscribingToSamePattern() {
builder.stream(Pattern.compile("some-regex"));
Expand Down Expand Up @@ -976,6 +962,20 @@ public void shouldThrowWhenSubscribedToAPatternWithSetAndUnsetResetPolicies() {
assertThrows(TopologyException.class, builder::build);
}

@Test
public void shouldNotAllowTablesFromSameTopic() {
builder.table("topic");
builder.table("topic");
assertThrows(TopologyException.class, builder::build);
}

@Test
public void shouldNowAllowStreamAndTableFromSameTopic() {
builder.stream("topic");
builder.table("topic");
assertThrows(TopologyException.class, builder::build);
}

private static void assertBuildDoesNotThrow(final StreamsBuilder builder) {
try {
builder.build();
Expand Down