fix: Update repartition processor names for KAFKA-9098 (#3802)

* fix: Changes for ksql-engine tests from KAFKA-9098 changes * fix: Changes for group-by QueryTranslationTest from KAFKA-9098 changes * fix: Changes for another group-by QueryTranslationTest from KAFKA-9098 changes * fix: Changes for joins in QueryTranslationTest from KAFKA-9098 changes * fix: Last changes for joins in QueryTranslationTest from KAFKA-9098 changes
confluentinc · Nov 9, 2019 · 2b86cd8 · 2b86cd8
1 parent c4dda94
commit 2b86cd8
Show file tree

Hide file tree

Showing 39 changed files with 284 additions and 284 deletions.
diff --git a/ksql-engine/src/test/java/io/confluent/ksql/planner/plan/AggregateNodeTest.java b/ksql-engine/src/test/java/io/confluent/ksql/planner/plan/AggregateNodeTest.java
@@ -196,7 +196,7 @@ public void shouldHaveSourceNodeForSecondSubtopolgyWithKsqlNameForRepartition()
 
     // Then:
     final TopologyDescription.Source node = (TopologyDescription.Source) getNodeByName(
-        builder.build(), "KSTREAM-SOURCE-0000000008");
+        builder.build(), "Aggregate-groupby-repartition-source");
     final List<String> successors = node.successors().stream().map(TopologyDescription.Node::name).collect(Collectors.toList());
     assertThat(node.predecessors(), equalTo(Collections.emptySet()));
     assertThat(successors, equalTo(Collections.singletonList("KSTREAM-AGGREGATE-0000000005")));
@@ -261,9 +261,9 @@ public void shouldHaveSinkNodeWithSameTopicAsSecondSource() {
 
     // Then:
     final TopologyDescription.Sink sink = (TopologyDescription.Sink) getNodeByName(builder.build(),
-        "KSTREAM-SINK-0000000006");
+        "Aggregate-groupby-repartition-sink");
     final TopologyDescription.Source source = (TopologyDescription.Source) getNodeByName(
-        builder.build(), "KSTREAM-SOURCE-0000000008");
+        builder.build(), "Aggregate-groupby-repartition-source");
     assertThat(sink.successors(), equalTo(Collections.emptySet()));
     assertThat(source.topicSet(), hasItem(sink.topic()));
   }

diff --git a/ksql-engine/src/test/java/io/confluent/ksql/planner/plan/JoinNodeTest.java b/ksql-engine/src/test/java/io/confluent/ksql/planner/plan/JoinNodeTest.java
@@ -278,7 +278,7 @@ public void shouldHaveLeftJoin() {
     final List<String> predecessors = leftJoin.predecessors().stream()
         .map(TopologyDescription.Node::name).collect(Collectors.toList());
     assertThat(leftJoin.stores(), equalTo(Utils.mkSet("KafkaTopic_Right-reduce")));
-    assertThat(predecessors, equalTo(Collections.singletonList("KSTREAM-SOURCE-0000000011")));
+    assertThat(predecessors, equalTo(Collections.singletonList("Join-repartition-source")));
   }
 
   @Test

diff --git a/...xpected_topology/0_6_0-pre/elements_-_group-by_removes_ROWTIME_from_output's_value_schema b/...xpected_topology/0_6_0-pre/elements_-_group-by_removes_ROWTIME_from_output's_value_schema
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...l-tests/src/test/resources/expected_topology/0_6_0-pre/group-by_-_ROWTIME_(stream-_table) b/...l-tests/src/test/resources/expected_topology/0_6_0-pre/group-by_-_ROWTIME_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...ted_topology/0_6_0-pre/group-by_-_UDAF_nested_in_UDF_in_select_expression_(stream-_table) b/...ted_topology/0_6_0-pre/group-by_-_UDAF_nested_in_UDF_in_select_expression_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...ted_topology/0_6_0-pre/group-by_-_UDF_nested_in_UDAF_in_select_expression_(stream-_table) b/...ted_topology/0_6_0-pre/group-by_-_UDF_nested_in_UDAF_in_select_expression_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...tic_binary_expression_with_projection_in-order_&_non-commutative_group_by_(stream-_table) b/...tic_binary_expression_with_projection_in-order_&_non-commutative_group_by_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...-tests/src/test/resources/expected_topology/0_6_0-pre/group-by_-_constant_(stream-_table) b/...-tests/src/test/resources/expected_topology/0_6_0-pre/group-by_-_constant_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...rc/test/resources/expected_topology/0_6_0-pre/group-by_-_duplicate_fields_(stream-_table) b/...rc/test/resources/expected_topology/0_6_0-pre/group-by_-_duplicate_fields_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...logy/0_6_0-pre/group-by_-_field_with_field_used_in_function_in_projection_(stream-_table) b/...logy/0_6_0-pre/group-by_-_field_with_field_used_in_function_in_projection_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...c/test/resources/expected_topology/0_6_0-pre/group-by_-_field_with_re-key_(stream-_table) b/...c/test/resources/expected_topology/0_6_0-pre/group-by_-_field_with_re-key_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...sources/expected_topology/0_6_0-pre/group-by_-_field_with_re-key_(stream-_table)_-_format b/...sources/expected_topology/0_6_0-pre/group-by_-_field_with_re-key_(stream-_table)_-_format
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...al-tests/src/test/resources/expected_topology/0_6_0-pre/group-by_-_fields_(stream-_table) b/...al-tests/src/test/resources/expected_topology/0_6_0-pre/group-by_-_fields_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...src/test/resources/expected_topology/0_6_0-pre/group-by_-_fields_(stream-_table)_-_format b/...src/test/resources/expected_topology/0_6_0-pre/group-by_-_fields_(stream-_table)_-_format
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005

diff --git a/...-tests/src/test/resources/expected_topology/0_6_0-pre/group-by_-_function_(stream-_table) b/...-tests/src/test/resources/expected_topology/0_6_0-pre/group-by_-_function_(stream-_table)
@@ -59,20 +59,20 @@ Topologies:
       --> Aggregate-groupby
       <-- KSTREAM-MAPVALUES-0000000002
     Processor: Aggregate-groupby (stores: [])
-      --> KSTREAM-FILTER-0000000007
+      --> Aggregate-groupby-repartition-filter
       <-- KSTREAM-FILTER-0000000003
-    Processor: KSTREAM-FILTER-0000000007 (stores: [])
-      --> KSTREAM-SINK-0000000006
+    Processor: Aggregate-groupby-repartition-filter (stores: [])
+      --> Aggregate-groupby-repartition-sink
       <-- Aggregate-groupby
-    Sink: KSTREAM-SINK-0000000006 (topic: Aggregate-groupby-repartition)
-      <-- KSTREAM-FILTER-0000000007
+    Sink: Aggregate-groupby-repartition-sink (topic: Aggregate-groupby-repartition)
+      <-- Aggregate-groupby-repartition-filter
 
   Sub-topology: 1
-    Source: KSTREAM-SOURCE-0000000008 (topics: [Aggregate-groupby-repartition])
+    Source: Aggregate-groupby-repartition-source (topics: [Aggregate-groupby-repartition])
       --> KSTREAM-AGGREGATE-0000000005
     Processor: KSTREAM-AGGREGATE-0000000005 (stores: [Aggregate-aggregate])
       --> KTABLE-MAPVALUES-0000000009
-      <-- KSTREAM-SOURCE-0000000008
+      <-- Aggregate-groupby-repartition-source
     Processor: KTABLE-MAPVALUES-0000000009 (stores: [])
       --> KTABLE-MAPVALUES-0000000010
       <-- KSTREAM-AGGREGATE-0000000005