kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: KAFKA-5698: Sort processor nodes based on its sub-tree size
Date Thu, 07 Sep 2017 21:36:52 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 3410f02fe -> 674407908


KAFKA-5698: Sort processor nodes based on its sub-tree size

1. Sort processor nodes within a sub-topology by its sub-tree size: nodes with largest sizes
are source nodes and hence printed earlier.

2. Sort sub-topologies by ids; sort global stores by the source topic names.

3. Open for discussion: start newlines for predecessor and successor.

4. Minor: space between processor nodes and stores / topics; maintain `[]` for the topic names.

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Damian Guy <damian.guy@gmail.com>, Matthias J. Sax <matthias@confluent.io>,
Ted Yu <yuzhihong@gmail.com>

Closes #3618 from guozhangwang/K5698-topology-description-sorting


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/67440790
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/67440790
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/67440790

Branch: refs/heads/trunk
Commit: 674407908b898dd18b447447ad3346b5268b1dfc
Parents: 3410f02
Author: Guozhang Wang <wangguoz@gmail.com>
Authored: Thu Sep 7 14:36:49 2017 -0700
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Thu Sep 7 14:36:49 2017 -0700

----------------------------------------------------------------------
 .../internals/InternalTopologyBuilder.java      | 112 ++++++++++++++-----
 .../internals/InternalTopologyBuilderTest.java  |  46 ++++++++
 2 files changed, 132 insertions(+), 26 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/67440790/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
index da5fe38..437e9e5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
@@ -35,10 +35,12 @@ import org.apache.kafka.streams.state.internals.WindowStoreSupplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -47,6 +49,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.regex.Pattern;
 
 
@@ -59,7 +62,7 @@ public class InternalTopologyBuilder {
     private static final String[] NO_PREDECESSORS = {};
 
     // node factories in a topological order
-    private final LinkedHashMap<String, NodeFactory> nodeFactories = new LinkedHashMap<>();
+    private final Map<String, NodeFactory> nodeFactories = new LinkedHashMap<>();
 
     // state factories
     private final Map<String, StateStoreFactory> stateFactories = new HashMap<>();
@@ -338,7 +341,6 @@ public class InternalTopologyBuilder {
 
             if (pattern == null) {
                 sourceTopics = topics.toString();
-                sourceTopics = sourceTopics.substring(1, sourceTopics.length() - 1); // trim
first and last, ie. []
             } else {
                 sourceTopics = pattern.toString();
             }
@@ -1263,6 +1265,34 @@ public class InternalTopologyBuilder {
         return false;
     }
 
+    private static class NodeComparator implements Comparator<TopologyDescription.Node>,
Serializable {
+
+        @Override
+        public int compare(final TopologyDescription.Node node1,
+                           final TopologyDescription.Node node2) {
+            final int size1 = ((AbstractNode) node1).size;
+            final int size2 = ((AbstractNode) node2).size;
+
+            // it is possible that two nodes have the same sub-tree size (think two nodes
connected via state stores)
+            // in this case default to processor name string
+            if (size1 != size2) {
+                return size2 - size1;
+            } else {
+                return node1.name().compareTo(node2.name());
+            }
+        }
+    }
+
+    private final static NodeComparator NODE_COMPARATOR = new NodeComparator();
+
+    private static void updateSize(final AbstractNode node, final int delta) {
+        node.size += delta;
+
+        for (final TopologyDescription.Node predecessor : node.predecessors()) {
+            updateSize((AbstractNode) predecessor, delta);
+        }
+    }
+
     private void describeSubtopology(final TopologyDescription description,
                                      final Integer subtopologyId,
                                      final Set<String> nodeNames) {
@@ -1280,12 +1310,13 @@ public class InternalTopologyBuilder {
                 final AbstractNode predecessor = nodesByName.get(predecessorName);
                 node.addPredecessor(predecessor);
                 predecessor.addSuccessor(node);
+                updateSize(predecessor, node.size);
             }
         }
 
         description.addSubtopology(new Subtopology(
-            subtopologyId,
-            new HashSet<TopologyDescription.Node>(nodesByName.values())));
+                subtopologyId,
+                new HashSet<TopologyDescription.Node>(nodesByName.values())));
     }
 
     private void describeGlobalStores(final TopologyDescription description) {
@@ -1339,8 +1370,8 @@ public class InternalTopologyBuilder {
 
         @Override
         public String toString() {
-            return "GlobalStore: " + source.name + "(topic: " + source.topics + ") ->
"
-                + processor.name + "(store: " + processor.stores.iterator().next() + ")\n";
+            return "GlobalStore: " + source.name + " (topic: " + source.topics + ")\n   
  --> "
+                + processor.name + " (store: " + processor.stores.iterator().next() + ")\n";
         }
 
         @Override
@@ -1365,11 +1396,15 @@ public class InternalTopologyBuilder {
 
     public abstract static class AbstractNode implements TopologyDescription.Node {
         final String name;
-        final Set<TopologyDescription.Node> predecessors = new HashSet<>();
-        final Set<TopologyDescription.Node> successors = new HashSet<>();
+        final Set<TopologyDescription.Node> predecessors = new TreeSet<>(NODE_COMPARATOR);
+        final Set<TopologyDescription.Node> successors = new TreeSet<>(NODE_COMPARATOR);
+
+        // size of the sub-topology rooted at this node, including the node itself
+        int size;
 
         AbstractNode(final String name) {
             this.name = name;
+            this.size = 1;
         }
 
         @Override
@@ -1417,7 +1452,7 @@ public class InternalTopologyBuilder {
 
         @Override
         public String toString() {
-            return "Source: " + name + "(topics: " + topics + ") --> " + nodeNames(successors);
+            return "Source: " + name + " (topics: " + topics + ")\n      --> " + nodeNames(successors);
         }
 
         @Override
@@ -1458,7 +1493,7 @@ public class InternalTopologyBuilder {
 
         @Override
         public String toString() {
-            return "Processor: " + name + "(stores: " + stores + ") --> " + nodeNames(successors)
+ " <-- " + nodeNames(predecessors);
+            return "Processor: " + name + " (stores: " + stores + ")\n      --> " + nodeNames(successors)
+ "\n      <-- " + nodeNames(predecessors);
         }
 
         @Override
@@ -1505,7 +1540,7 @@ public class InternalTopologyBuilder {
 
         @Override
         public String toString() {
-            return "Sink: " + name + "(topic: " + topic + ") <-- " + nodeNames(predecessors);
+            return "Sink: " + name + " (topic: " + topic + ")\n      <-- " + nodeNames(predecessors);
         }
 
         @Override
@@ -1532,12 +1567,12 @@ public class InternalTopologyBuilder {
 
     public final static class Subtopology implements org.apache.kafka.streams.TopologyDescription.Subtopology
{
         private final int id;
-        private final Set<org.apache.kafka.streams.TopologyDescription.Node> nodes;
+        private final Set<TopologyDescription.Node> nodes;
 
-        public Subtopology(final int id,
-                    final Set<org.apache.kafka.streams.TopologyDescription.Node> nodes)
{
+        public Subtopology(final int id, final Set<TopologyDescription.Node> nodes)
{
             this.id = id;
-            this.nodes = nodes;
+            this.nodes = new TreeSet<>(NODE_COMPARATOR);
+            this.nodes.addAll(nodes);
         }
 
         @Override
@@ -1546,10 +1581,15 @@ public class InternalTopologyBuilder {
         }
 
         @Override
-        public Set<org.apache.kafka.streams.TopologyDescription.Node> nodes() {
+        public Set<TopologyDescription.Node> nodes() {
             return Collections.unmodifiableSet(nodes);
         }
 
+        // only for testing
+        public Iterator<TopologyDescription.Node> nodesInOrder() {
+            return nodes.iterator();
+        }
+
         @Override
         public String toString() {
             return "Sub-topology: " + id + "\n" + nodesAsString();
@@ -1557,7 +1597,7 @@ public class InternalTopologyBuilder {
 
         private String nodesAsString() {
             final StringBuilder sb = new StringBuilder();
-            for (final org.apache.kafka.streams.TopologyDescription.Node node : nodes) {
+            for (final TopologyDescription.Node node : nodes) {
                 sb.append("    ");
                 sb.append(node);
                 sb.append('\n');
@@ -1628,31 +1668,51 @@ public class InternalTopologyBuilder {
         }
     }
 
+    private static class GlobalStoreComparator implements Comparator<TopologyDescription.GlobalStore>,
Serializable {
+        @Override
+        public int compare(final TopologyDescription.GlobalStore globalStore1,
+                           final TopologyDescription.GlobalStore globalStore2) {
+            return globalStore1.source().name().compareTo(globalStore2.source().name());
+        }
+    }
+
+    private final static GlobalStoreComparator GLOBALSTORE_COMPARATOR = new GlobalStoreComparator();
+
+    private static class SubtopologyComparator implements Comparator<TopologyDescription.Subtopology>,
Serializable {
+        @Override
+        public int compare(final TopologyDescription.Subtopology subtopology1,
+                           final TopologyDescription.Subtopology subtopology2) {
+            return subtopology1.id() - subtopology2.id();
+        }
+    }
+
+    private final static SubtopologyComparator SUBTOPOLOGY_COMPARATOR = new SubtopologyComparator();
+
     public final static class TopologyDescription implements org.apache.kafka.streams.TopologyDescription
{
-        private final Set<org.apache.kafka.streams.TopologyDescription.Subtopology>
subtopologies = new HashSet<>();
-        private final Set<org.apache.kafka.streams.TopologyDescription.GlobalStore>
globalStores = new HashSet<>();
+        private final Set<TopologyDescription.Subtopology> subtopologies = new TreeSet<>(SUBTOPOLOGY_COMPARATOR);
+        private final Set<TopologyDescription.GlobalStore> globalStores = new TreeSet<>(GLOBALSTORE_COMPARATOR);
 
-        public void addSubtopology(final org.apache.kafka.streams.TopologyDescription.Subtopology
subtopology) {
+        public void addSubtopology(final TopologyDescription.Subtopology subtopology) {
             subtopologies.add(subtopology);
         }
 
-        public void addGlobalStore(final org.apache.kafka.streams.TopologyDescription.GlobalStore
globalStore) {
+        public void addGlobalStore(final TopologyDescription.GlobalStore globalStore) {
             globalStores.add(globalStore);
         }
 
         @Override
-        public Set<org.apache.kafka.streams.TopologyDescription.Subtopology> subtopologies()
{
+        public Set<TopologyDescription.Subtopology> subtopologies() {
             return Collections.unmodifiableSet(subtopologies);
         }
 
         @Override
-        public Set<org.apache.kafka.streams.TopologyDescription.GlobalStore> globalStores()
{
+        public Set<TopologyDescription.GlobalStore> globalStores() {
             return Collections.unmodifiableSet(globalStores);
         }
 
         @Override
         public String toString() {
-            return subtopologiesAsString() + globalStoresAsString();
+            return subtopologiesAsString() + "\n" + globalStoresAsString();
         }
 
         private String subtopologiesAsString() {
@@ -1661,7 +1721,7 @@ public class InternalTopologyBuilder {
             if (subtopologies.isEmpty()) {
                 sb.append("  none\n");
             } else {
-                for (final org.apache.kafka.streams.TopologyDescription.Subtopology st :
subtopologies) {
+                for (final TopologyDescription.Subtopology st : subtopologies) {
                     sb.append("  ");
                     sb.append(st);
                 }
@@ -1675,7 +1735,7 @@ public class InternalTopologyBuilder {
             if (globalStores.isEmpty()) {
                 sb.append("  none\n");
             } else {
-                for (final org.apache.kafka.streams.TopologyDescription.GlobalStore gs :
globalStores) {
+                for (final TopologyDescription.GlobalStore gs : globalStores) {
                     sb.append("  ");
                     sb.append(gs);
                 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/67440790/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
index 91edac5..95636ec 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
@@ -20,6 +20,7 @@ import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyDescription;
 import org.apache.kafka.streams.errors.StreamsException;
 import org.apache.kafka.streams.errors.TopologyBuilderException;
 import org.apache.kafka.streams.errors.TopologyException;
@@ -43,6 +44,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -665,6 +667,50 @@ public class InternalTopologyBuilderTest {
     }
 
     @Test
+    public void shouldSortProcessorNodesCorrectly() throws Exception {
+        builder.addSource(null, "source1", null, null, null, "topic1");
+        builder.addSource(null, "source2", null, null, null, "topic2");
+        builder.addProcessor("processor1", new MockProcessorSupplier(), "source1");
+        builder.addProcessor("processor2", new MockProcessorSupplier(), "source1", "source2");
+        builder.addProcessor("processor3", new MockProcessorSupplier(), "processor2");
+        builder.addSink("sink1", "topic2", null, null, null, "processor1", "processor3");
+
+        assertEquals(1, builder.describe().subtopologies().size());
+
+        final Iterator<TopologyDescription.Node> iterator = ((InternalTopologyBuilder.Subtopology)
builder.describe().subtopologies().iterator().next()).nodesInOrder();
+
+        assertTrue(iterator.hasNext());
+        InternalTopologyBuilder.AbstractNode node = (InternalTopologyBuilder.AbstractNode)
iterator.next();
+        assertTrue(node.name.equals("source1"));
+        assertEquals(6, node.size);
+
+        assertTrue(iterator.hasNext());
+        node = (InternalTopologyBuilder.AbstractNode) iterator.next();
+        assertTrue(node.name.equals("source2"));
+        assertEquals(4, node.size);
+
+        assertTrue(iterator.hasNext());
+        node = (InternalTopologyBuilder.AbstractNode) iterator.next();
+        assertTrue(node.name.equals("processor2"));
+        assertEquals(3, node.size);
+
+        assertTrue(iterator.hasNext());
+        node = (InternalTopologyBuilder.AbstractNode) iterator.next();
+        assertTrue(node.name.equals("processor1"));
+        assertEquals(2, node.size);
+
+        assertTrue(iterator.hasNext());
+        node = (InternalTopologyBuilder.AbstractNode) iterator.next();
+        assertTrue(node.name.equals("processor3"));
+        assertEquals(2, node.size);
+
+        assertTrue(iterator.hasNext());
+        node = (InternalTopologyBuilder.AbstractNode) iterator.next();
+        assertTrue(node.name.equals("sink1"));
+        assertEquals(1, node.size);
+    }
+
+    @Test
     public void shouldConnectRegexMatchedTopicsToStateStore() throws Exception {
         builder.addSource(null, "ingest", null, null, null, Pattern.compile("topic-\\d+"));
         builder.addProcessor("my-processor", new MockProcessorSupplier(), "ingest");


Mime
View raw message