Skip to content

KAFKA-7502: Cleanup KTable materialization logic in a single place #6174

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
guozhangwang merged 9 commits into from
Mar 15, 2019
Merged

KAFKA-7502: Cleanup KTable materialization logic in a single place #6174

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
a5ee071
1. Separate Materialized-related logic into [NonMaterialized|Material…
dongjinleekr Jan 19, 2019
e7a8f76
Remove key/value serde decision logic from KTableImpl#doJoin
dongjinleekr Jan 19, 2019
30a699e
Trivial: change method names following coding convention (i.e., remov…
dongjinleekr Jan 19, 2019
749ab8c
Remove key/value serde decision logic from KTableImpl#doJoin
dongjinleekr Jan 19, 2019
ca57df5
1. Merge MaterializedKTableKTableJoinNode, NonMaterializedKTableKTabl…
dongjinleekr Mar 4, 2019
f8a8d53
Remove unused fields: KTableKTableJoinNode.KTableKTableJoinNodeBuilde…
dongjinleekr Mar 4, 2019
e39d8d7
1. Remove valueJoiner parameter from KTableKTableJoinNode. 2. Remove …
dongjinleekr Mar 12, 2019
c4afbff
Remove InternalStreamsBuilder parameter from KTableKTableJoinNode#kTa…
dongjinleekr Mar 12, 2019
0b278dc
Cleanup KTableKTableJoinNode building routine
dongjinleekr Mar 14, 2019
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
108 changes: 51 additions & 57 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import org.apache.kafka.streams.kstream.internals.suppress.SuppressedInternal;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.internals.InMemoryTimeOrderedKeyValueBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -465,28 +466,9 @@ private <VO, VR> KTable<K, VR> doJoin(final KTable<K, VO> other,
final boolean rightOuter) {
Objects.requireNonNull(other, "other can't be null");
Objects.requireNonNull(joiner, "joiner can't be null");
final String internalQueryableName = materializedInternal == null ? null : materializedInternal.storeName();
final String joinMergeName = builder.newProcessorName(MERGE_NAME);

return buildJoin(
(AbstractStream<K, VO>) other,
joiner,
leftOuter,
rightOuter,
joinMergeName,
internalQueryableName,
materializedInternal
);
}

private <V1, R> KTable<K, R> buildJoin(final AbstractStream<K, V1> other,
final ValueJoiner<? super V, ? super V1, ? extends R> joiner,
final boolean leftOuter,
final boolean rightOuter,
final String joinMergeName,
final String internalQueryableName,
final MaterializedInternal<K, R, KeyValueStore<Bytes, byte[]>> materializedInternal) {
final Set<String> allSourceNodes = ensureJoinableWith(other);
final String joinMergeName = builder.newProcessorName(MERGE_NAME);
final Set<String> allSourceNodes = ensureJoinableWith((AbstractStream<K, VO>) other);

if (leftOuter) {
enableSendingOldValues();
Expand All @@ -495,57 +477,69 @@ private <V1, R> KTable<K, R> buildJoin(final AbstractStream<K, V1> other,
((KTableImpl) other).enableSendingOldValues();
}

final String joinThisName = builder.newProcessorName(JOINTHIS_NAME);
final String joinOtherName = builder.newProcessorName(JOINOTHER_NAME);


final KTableKTableAbstractJoin<K, R, V, V1> joinThis;
final KTableKTableAbstractJoin<K, R, V1, V> joinOther;
final KTableKTableAbstractJoin<K, VR, V, VO> joinThis;
final KTableKTableAbstractJoin<K, VR, VO, V> joinOther;

if (!leftOuter) { // inner
joinThis = new KTableKTableInnerJoin<>(this, (KTableImpl<K, ?, V1>) other, joiner);
joinOther = new KTableKTableInnerJoin<>((KTableImpl<K, ?, V1>) other, this, reverseJoiner(joiner));
joinThis = new KTableKTableInnerJoin<>(this, (KTableImpl<K, ?, VO>) other, joiner);
joinOther = new KTableKTableInnerJoin<>((KTableImpl<K, ?, VO>) other, this, reverseJoiner(joiner));
} else if (!rightOuter) { // left
joinThis = new KTableKTableLeftJoin<>(this, (KTableImpl<K, ?, V1>) other, joiner);
joinOther = new KTableKTableRightJoin<>((KTableImpl<K, ?, V1>) other, this, reverseJoiner(joiner));
joinThis = new KTableKTableLeftJoin<>(this, (KTableImpl<K, ?, VO>) other, joiner);
joinOther = new KTableKTableRightJoin<>((KTableImpl<K, ?, VO>) other, this, reverseJoiner(joiner));
} else { // outer
joinThis = new KTableKTableOuterJoin<>(this, (KTableImpl<K, ?, V1>) other, joiner);
joinOther = new KTableKTableOuterJoin<>((KTableImpl<K, ?, V1>) other, this, reverseJoiner(joiner));
joinThis = new KTableKTableOuterJoin<>(this, (KTableImpl<K, ?, VO>) other, joiner);
joinOther = new KTableKTableOuterJoin<>((KTableImpl<K, ?, VO>) other, this, reverseJoiner(joiner));
}

final KTableKTableJoinMerger<K, R> joinMerge = new KTableKTableJoinMerger<>(joinThis, joinOther, internalQueryableName);
final String joinThisName = builder.newProcessorName(JOINTHIS_NAME);
final String joinOtherName = builder.newProcessorName(JOINOTHER_NAME);

final KTableKTableJoinNode.KTableKTableJoinNodeBuilder<K, V, V1, R> kTableJoinNodeBuilder = KTableKTableJoinNode.kTableKTableJoinNodeBuilder();
final ProcessorParameters<K, Change<V>> joinThisProcessorParameters = new ProcessorParameters<>(joinThis, joinThisName);
final ProcessorParameters<K, Change<VO>> joinOtherProcessorParameters = new ProcessorParameters<>(joinOther, joinOtherName);

final KTableKTableJoinNode.KTableKTableJoinNodeBuilder<K, V, VO, VR> kTableJoinNodeBuilder =
KTableKTableJoinNode.<K, V, VO, VR>kTableKTableJoinNodeBuilder()
.withNodeName(joinMergeName)
.withJoinThisProcessorParameters(joinThisProcessorParameters)
.withJoinOtherProcessorParameters(joinOtherProcessorParameters)
.withThisJoinSideNodeName(name)
.withOtherJoinSideNodeName(((KTableImpl) other).name)
.withJoinThisStoreNames(valueGetterSupplier().storeNames())
.withJoinOtherStoreNames(((KTableImpl) other).valueGetterSupplier().storeNames());

// only materialize if specified in Materialized
if (materializedInternal != null) {
kTableJoinNodeBuilder.withMaterializedInternal(materializedInternal);
final Serde<K> keySerde = materializedInternal.keySerde() != null ? materializedInternal.keySerde() : this.keySerde;
final Serde<VR> valueSerde = materializedInternal.valueSerde();
final String queryableStoreName = materializedInternal.storeName();
final StoreBuilder<KeyValueStore<K, VR>> storeBuilder = new KeyValueStoreMaterializer<>(materializedInternal).materialize();
kTableJoinNodeBuilder
Comment thread
dongjinleekr marked this conversation as resolved.
Outdated
.withKeySerde(keySerde)
.withValueSerde(valueSerde)
.withQueryableStoreName(queryableStoreName)
.withStoreBuilder(storeBuilder);
} else {
final Serde<K> keySerde = this.keySerde;
final Serde<VR> valueSerde = null;
final String queryableStoreName = null;
final StoreBuilder<KeyValueStore<K, VR>> storeBuilder = null;
kTableJoinNodeBuilder
.withKeySerde(keySerde)
.withValueSerde(valueSerde)
.withQueryableStoreName(queryableStoreName)
.withStoreBuilder(storeBuilder);
}
kTableJoinNodeBuilder.withNodeName(joinMergeName);

final ProcessorParameters<K, Change<V>> joinThisProcessorParameters = new ProcessorParameters<>(joinThis, joinThisName);
final ProcessorParameters<K, Change<V1>> joinOtherProcessorParameters = new ProcessorParameters<>(joinOther, joinOtherName);
final ProcessorParameters<K, Change<R>> joinMergeProcessorParameters = new ProcessorParameters<>(joinMerge, joinMergeName);

kTableJoinNodeBuilder.withJoinMergeProcessorParameters(joinMergeProcessorParameters)
.withJoinOtherProcessorParameters(joinOtherProcessorParameters)
.withJoinThisProcessorParameters(joinThisProcessorParameters)
.withJoinThisStoreNames(valueGetterSupplier().storeNames())
.withJoinOtherStoreNames(((KTableImpl) other).valueGetterSupplier().storeNames())
.withOtherJoinSideNodeName(((KTableImpl) other).name)
.withThisJoinSideNodeName(name);

final KTableKTableJoinNode<K, V, V1, R> kTableKTableJoinNode = kTableJoinNodeBuilder.build();
final KTableKTableJoinNode<K, V, VO, VR> kTableKTableJoinNode = kTableJoinNodeBuilder.build();
builder.addGraphNode(this.streamsGraphNode, kTableKTableJoinNode);

// we can inherit parent key serde if user do not provide specific overrides
return new KTableImpl<K, Change<R>, R>(
joinMergeName,
materializedInternal != null && materializedInternal.keySerde() != null ? materializedInternal.keySerde() : keySerde,
materializedInternal != null ? materializedInternal.valueSerde() : null,
return new KTableImpl<K, Change<VR>, VR>(
kTableKTableJoinNode.nodeName(),
kTableKTableJoinNode.keySerde(),
kTableKTableJoinNode.valueSerde(),
allSourceNodes,
internalQueryableName,
joinMerge,
kTableKTableJoinNode.queryableStoreName(),
kTableKTableJoinNode.joinMerger(),
kTableKTableJoinNode,
builder
);
Expand Down
17 changes: 16 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableJoinMerger.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@
import java.util.HashSet;
import java.util.Set;

class KTableKTableJoinMerger<K, V> implements KTableProcessorSupplier<K, V, V> {
public class KTableKTableJoinMerger<K, V> implements KTableProcessorSupplier<K, V, V> {

private final KTableProcessorSupplier<K, ?, V> parent1;
private final KTableProcessorSupplier<K, ?, V> parent2;
Expand All @@ -40,6 +40,10 @@ class KTableKTableJoinMerger<K, V> implements KTableProcessorSupplier<K, V, V> {
this.queryableName = queryableName;
}

public String getQueryableName() {
return queryableName;
}

@Override
public Processor<K, Change<V>> get() {
return new KTableKTableJoinMergeProcessor();
Expand Down Expand Up @@ -78,6 +82,17 @@ public void enableSendingOldValues() {
sendOldValues = true;
}

public static <K, V> KTableKTableJoinMerger<K, V> of(final KTableProcessorSupplier<K, ?, V> parent1,
final KTableProcessorSupplier<K, ?, V> parent2) {
return of(parent1, parent2, null);
}

public static <K, V> KTableKTableJoinMerger<K, V> of(final KTableProcessorSupplier<K, ?, V> parent1,
final KTableProcessorSupplier<K, ?, V> parent2,
final String queryableName) {
return new KTableKTableJoinMerger<>(parent1, parent2, queryableName);
}

private class KTableKTableJoinMergeProcessor extends AbstractProcessor<K, Change<V>> {
private KeyValueStore<K, V> store;
private TupleForwarder<K, V> tupleForwarder;
Expand Down
Loading