You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/08/10 21:54:04 UTC
[GitHub] [kafka] vvcephei commented on a change in pull request #9141: MINOR: Improve checks for CogroupedStreamAggregateBuilder
vvcephei commented on a change in pull request #9141:
URL: https://github.com/apache/kafka/pull/9141#discussion_r468206463
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedStreamAggregateBuilder.java
##########
@@ -85,68 +160,38 @@
groupedStreams.remove(kGrouped);
kGrouped.ensureCopartitionWith(groupedStreams);
- final Collection<StreamsGraphNode> processors = new ArrayList<>();
- boolean stateCreated = false;
- int counter = 0;
- for (final Entry<KGroupedStreamImpl<K, ?>, Aggregator<? super K, Object, VOut>> kGroupedStream : groupPatterns.entrySet()) {
- final StatefulProcessorNode<K, ?> statefulProcessorNode = getStatefulProcessorNode(
- kGroupedStream.getValue(),
- initializer,
- named.suffixWithOrElseGet(
- "-cogroup-agg-" + counter++,
- builder,
- CogroupedKStreamImpl.AGGREGATE_NAME),
- stateCreated,
- storeBuilder,
- windows,
- sessionWindows,
- sessionMerger);
- stateCreated = true;
- processors.add(statefulProcessorNode);
- builder.addGraphNode(parentNodes.get(kGroupedStream.getKey()), statefulProcessorNode);
- }
+ }
+
+ <KR, VIn> KTable<KR, VOut> createTable(final Collection<StreamsGraphNode> processors,
+ final NamedInternal named,
+ final Serde<KR> keySerde,
+ final Serde<VOut> valueSerde,
+ final String queryableName) {
final String mergeProcessorName = named.suffixWithOrElseGet(
- "-cogroup-merge",
- builder,
- CogroupedKStreamImpl.MERGE_NAME);
+ "-cogroup-merge",
+ builder,
+ CogroupedKStreamImpl.MERGE_NAME);
Review comment:
It seems like your indentation is set to 8 spaces instead of 4.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org