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-4645: Improve test coverage of ProcessorTopology
Date Thu, 02 Feb 2017 04:30:07 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk da70316a5 -> 35cd008e5


KAFKA-4645: Improve test coverage of ProcessorTopology

the toString method prints the topology, but had no tests making sure it works and/or doesn't
cause exceptions

Author: Damian Guy <damian.guy@gmail.com>

Reviewers: Matthias J. Sax, Eno Thereska, Guozhang Wang

Closes #2444 from dguy/KAFKA-4645


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

Branch: refs/heads/trunk
Commit: 35cd008e5a1ec89d09f956a6f9952ad834e556d7
Parents: da70316
Author: Damian Guy <damian.guy@gmail.com>
Authored: Wed Feb 1 20:30:04 2017 -0800
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Wed Feb 1 20:30:04 2017 -0800

----------------------------------------------------------------------
 .../internals/ProcessorTopologyTest.java        | 88 +++++++++++++++-----
 1 file changed, 67 insertions(+), 21 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/35cd008e/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
index 71c234e..f35a2b5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
@@ -46,6 +46,8 @@ import org.junit.Test;
 import java.io.File;
 import java.util.Properties;
 
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
@@ -55,12 +57,14 @@ public class ProcessorTopologyTest {
     private static final Serializer<String> STRING_SERIALIZER = new StringSerializer();
     private static final Deserializer<String> STRING_DESERIALIZER = new StringDeserializer();
 
-    protected static final String INPUT_TOPIC_1 = "input-topic-1";
-    protected static final String INPUT_TOPIC_2 = "input-topic-2";
-    protected static final String OUTPUT_TOPIC_1 = "output-topic-1";
-    protected static final String OUTPUT_TOPIC_2 = "output-topic-2";
+    private static final String INPUT_TOPIC_1 = "input-topic-1";
+    private static final String INPUT_TOPIC_2 = "input-topic-2";
+    private static final String OUTPUT_TOPIC_1 = "output-topic-1";
+    private static final String OUTPUT_TOPIC_2 = "output-topic-2";
 
     private static long timestamp = 1000L;
+    private final TopologyBuilder builder = new TopologyBuilder();
+    private final MockProcessorSupplier mockProcessorSupplier = new MockProcessorSupplier();
 
     private ProcessorTopologyTestDriver driver;
     private StreamsConfig config;
@@ -89,7 +93,7 @@ public class ProcessorTopologyTest {
 
     @Test
     public void testTopologyMetadata() {
-        final TopologyBuilder builder = new TopologyBuilder().setApplicationId("X");
+        builder.setApplicationId("X");
 
         builder.addSource("source-1", "topic-1");
         builder.addSource("source-2", "topic-2", "topic-3");
@@ -206,7 +210,7 @@ public class ProcessorTopologyTest {
         final String global = "global";
         final String topic = "topic";
         final KeyValueStore<String, String> globalStore = (KeyValueStore<String,
String>) storeSupplier.get();
-        final TopologyBuilder topologyBuilder = new TopologyBuilder()
+        final TopologyBuilder topologyBuilder = this.builder
                 .addGlobalStore(globalStore, global, STRING_DESERIALIZER, STRING_DESERIALIZER,
topic, "processor", define(new StatefulProcessor("my-store")));
 
         driver = new ProcessorTopologyTestDriver(config, topologyBuilder, "my-store");
@@ -230,9 +234,51 @@ public class ProcessorTopologyTest {
         assertNoOutputRecord(OUTPUT_TOPIC_1);
     }
 
+    @Test
+    public void shouldCreateStringWithSourceAndTopics() throws Exception {
+        builder.addSource("source", "topic1", "topic2");
+        final ProcessorTopology topology = builder.build(null);
+        final String result = topology.toString();
+        assertThat(result, containsString("source:\n\t\ttopics:\t\t[topic1, topic2]\n"));
+    }
+
+    @Test
+    public void shouldCreateStringWithMultipleSourcesAndTopics() throws Exception {
+        builder.addSource("source", "topic1", "topic2");
+        builder.addSource("source2", "t", "t1", "t2");
+        final ProcessorTopology topology = builder.build(null);
+        final String result = topology.toString();
+        assertThat(result, containsString("source:\n\t\ttopics:\t\t[topic1, topic2]\n"));
+        assertThat(result, containsString("source2:\n\t\ttopics:\t\t[t, t1, t2]\n"));
+    }
 
+    @Test
+    public void shouldCreateStringWithProcessors() throws Exception {
+        builder.addSource("source", "t")
+                .addProcessor("processor", mockProcessorSupplier, "source")
+                .addProcessor("other", mockProcessorSupplier, "source");
+        final ProcessorTopology topology = builder.build(null);
+        final String result = topology.toString();
+        assertThat(result, containsString("\t\tchildren:\t[processor, other]"));
+        assertThat(result, containsString("processor:\n"));
+        assertThat(result, containsString("other:\n"));
+    }
+
+    @Test
+    public void shouldRecursivelyPrintChildren() throws Exception {
+        builder.addSource("source", "t")
+                .addProcessor("processor", mockProcessorSupplier, "source")
+                .addProcessor("child-one", mockProcessorSupplier, "processor")
+                .addProcessor("child-one-one", mockProcessorSupplier, "child-one")
+                .addProcessor("child-two", mockProcessorSupplier, "processor")
+                .addProcessor("child-two-one", mockProcessorSupplier, "child-two");
+
+        final String result = builder.build(null).toString();
+        assertThat(result, containsString("child-one:\n\t\tchildren:\t[child-one-one]"));
+        assertThat(result, containsString("child-two:\n\t\tchildren:\t[child-two-one]"));
+    }
 
-    protected void assertNextOutputRecord(String topic, String key, String value) {
+    private void assertNextOutputRecord(String topic, String key, String value) {
         ProducerRecord<String, String> record = driver.readOutput(topic, STRING_DESERIALIZER,
STRING_DESERIALIZER);
         assertEquals(topic, record.topic());
         assertEquals(key, record.key());
@@ -240,7 +286,7 @@ public class ProcessorTopologyTest {
         assertNull(record.partition());
     }
 
-    protected void assertNextOutputRecord(String topic, String key, String value, Integer
partition) {
+    private void assertNextOutputRecord(String topic, String key, String value, Integer partition)
{
         ProducerRecord<String, String> record = driver.readOutput(topic, STRING_DESERIALIZER,
STRING_DESERIALIZER);
         assertEquals(topic, record.topic());
         assertEquals(key, record.key());
@@ -248,11 +294,11 @@ public class ProcessorTopologyTest {
         assertEquals(partition, record.partition());
     }
 
-    protected void assertNoOutputRecord(String topic) {
+    private void assertNoOutputRecord(String topic) {
         assertNull(driver.readOutput(topic));
     }
 
-    protected StreamPartitioner<Object, Object> constantPartitioner(final Integer partition)
{
+    private StreamPartitioner<Object, Object> constantPartitioner(final Integer partition)
{
         return new StreamPartitioner<Object, Object>() {
             @Override
             public Integer partition(Object key, Object value, int numPartitions) {
@@ -261,28 +307,28 @@ public class ProcessorTopologyTest {
         };
     }
 
-    protected TopologyBuilder createSimpleTopology(int partition) {
-        return new TopologyBuilder().addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER,
INPUT_TOPIC_1)
+    private TopologyBuilder createSimpleTopology(int partition) {
+        return builder.addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_1)
                                     .addProcessor("processor", define(new ForwardingProcessor()),
"source")
                                     .addSink("sink", OUTPUT_TOPIC_1, constantPartitioner(partition),
"processor");
     }
 
-    protected TopologyBuilder createMultiplexingTopology() {
-        return new TopologyBuilder().addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER,
INPUT_TOPIC_1)
+    private TopologyBuilder createMultiplexingTopology() {
+        return builder.addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_1)
                                     .addProcessor("processor", define(new MultiplexingProcessor(2)),
"source")
                                     .addSink("sink1", OUTPUT_TOPIC_1, "processor")
                                     .addSink("sink2", OUTPUT_TOPIC_2, "processor");
     }
 
-    protected TopologyBuilder createMultiplexByNameTopology() {
-        return new TopologyBuilder().addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER,
INPUT_TOPIC_1)
+    private TopologyBuilder createMultiplexByNameTopology() {
+        return builder.addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_1)
             .addProcessor("processor", define(new MultiplexByNameProcessor(2)), "source")
             .addSink("sink0", OUTPUT_TOPIC_1, "processor")
             .addSink("sink1", OUTPUT_TOPIC_2, "processor");
     }
 
-    protected TopologyBuilder createStatefulTopology(String storeName) {
-        return new TopologyBuilder().addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER,
INPUT_TOPIC_1)
+    private TopologyBuilder createStatefulTopology(String storeName) {
+        return builder.addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_1)
                                     .addProcessor("processor", define(new StatefulProcessor(storeName)),
"source")
                                     .addStateStore(
                                             Stores.create(storeName).withStringKeys().withStringValues().inMemory().build(),
@@ -292,8 +338,8 @@ public class ProcessorTopologyTest {
     }
 
 
-    protected TopologyBuilder createSimpleMultiSourceTopology(int partition) {
-        return new TopologyBuilder().addSource("source-1", STRING_DESERIALIZER, STRING_DESERIALIZER,
INPUT_TOPIC_1)
+    private TopologyBuilder createSimpleMultiSourceTopology(int partition) {
+        return builder.addSource("source-1", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_1)
                 .addProcessor("processor-1", define(new ForwardingProcessor()), "source-1")
                 .addSink("sink-1", OUTPUT_TOPIC_1, constantPartitioner(partition), "processor-1")
                 .addSource("source-2", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_2)
@@ -414,7 +460,7 @@ public class ProcessorTopologyTest {
         }
     }
 
-    protected <K, V> ProcessorSupplier<K, V> define(final Processor<K, V>
processor) {
+    private <K, V> ProcessorSupplier<K, V> define(final Processor<K, V>
processor) {
         return new ProcessorSupplier<K, V>() {
             @Override
             public Processor<K, V> get() {


Mime
View raw message