SingleProcessorTopology implements Processor

stream/src/main/java/org/apache/kafka/stream/examples/MapKStreamJob.java

@@ -34,7 +34,7 @@ public class MapKStreamJob extends KStreamTopology {
     @Override
     public void topology() {
 
-        // With overriden de-serializer
+        // With overridden de-serializer
         KStream stream1 = from(new StringDeserializer(), new StringDeserializer(), "topic1");
 
         stream1.map(new KeyValueMapper<String, Integer, String, String>() {

stream/src/main/java/org/apache/kafka/stream/examples/PrintKStreamJob.java

@@ -17,45 +17,51 @@
 
 package org.apache.kafka.stream.examples;
 
-import org.apache.kafka.stream.topology.Processor;
 import org.apache.kafka.stream.KStreamContext;
 import org.apache.kafka.stream.KafkaStreaming;
 import org.apache.kafka.stream.StreamingConfig;
-import org.apache.kafka.stream.topology.SingleProcessorTopology;
+import org.apache.kafka.stream.topology.KStreamTopology;
+import org.apache.kafka.stream.topology.Processor;
 
 import java.util.Properties;
 
-public class PrintKStreamJob<K, V> implements Processor<K, V> {
+public class PrintKStreamJob extends KStreamTopology {
 
-    private KStreamContext context;
+    private class MyProcessor<K, V> implements Processor<K, V> {
+        private KStreamContext context;
 
-    @Override
-    public void init(KStreamContext context) {
-        this.context = context;
-    }
+        @Override
+        public void init(KStreamContext context) {
+            this.context = context;
+        }
 
-    @Override
-    public void process(K key, V value) {
-        System.out.println("[" + key + ", " + value + "]");
+        @Override
+        public void process(K key, V value) {
+            System.out.println("[" + key + ", " + value + "]");
 
-        context.commit();
+            context.commit();
 
-        context.send("topic", key, value);
-    }
+            context.send("topic", key, value);
+        }
 
-    @Override
-    public void punctuate(long streamTime) {
-        // do nothing
+        @Override
+        public void punctuate(long streamTime) {
+            // do nothing
+        }
+
+        @Override
+        public void close() {
+            // do nothing
+        }
     }
 
+    @SuppressWarnings("unchecked")
     @Override
-    public void close() {
-        // do nothing
-    }
+    public void topology() { from("topic").process(new MyProcessor()); }
 
     public static void main(String[] args) {
         KafkaStreaming streaming = new KafkaStreaming(
-            new SingleProcessorTopology(PrintKStreamJob.class, args),
+            new PrintKStreamJob(),
             new StreamingConfig(new Properties())
         );
         streaming.run();

stream/src/main/java/org/apache/kafka/stream/examples/StatefulKStreamJob.java

@@ -17,61 +17,68 @@
 
 package org.apache.kafka.stream.examples;
 
-import org.apache.kafka.stream.topology.Processor;
+import org.apache.kafka.stream.KStream;
 import org.apache.kafka.stream.KStreamContext;
 import org.apache.kafka.stream.KafkaStreaming;
 import org.apache.kafka.stream.StreamingConfig;
 import org.apache.kafka.stream.state.Entry;
 import org.apache.kafka.stream.state.InMemoryKeyValueStore;
 import org.apache.kafka.stream.state.KeyValueIterator;
 import org.apache.kafka.stream.state.KeyValueStore;
-import org.apache.kafka.stream.topology.SingleProcessorTopology;
+import org.apache.kafka.stream.topology.KStreamTopology;
+import org.apache.kafka.stream.topology.Processor;
 
 import java.util.Properties;
 
-public class StatefulKStreamJob implements Processor<String, Integer> {
+public class StatefulKStreamJob extends KStreamTopology {
 
-    private KStreamContext context;
-    private KeyValueStore<String, Integer> kvStore;
+    private class MyProcessor implements Processor<String, Integer> {
+        private KStreamContext context;
+        private KeyValueStore<String, Integer> kvStore;
 
-    @Override
-    public void init(KStreamContext context) {
-        this.context = context;
-        this.context.schedule(this, 1000);
+        @Override
+        public void init(KStreamContext context) {
+            this.context = context;
+            this.context.schedule(this, 1000);
 
-        this.kvStore = new InMemoryKeyValueStore<>("local-state", context);
-    }
+            this.kvStore = new InMemoryKeyValueStore<>("local-state", context);
+        }
 
-    @Override
-    public void process(String key, Integer value) {
-        Integer oldValue = this.kvStore.get(key);
-        if (oldValue == null) {
-            this.kvStore.put(key, value);
-        } else {
-            int newValue = oldValue + value;
-            this.kvStore.put(key, newValue);
+        @Override
+        public void process(String key, Integer value) {
+            Integer oldValue = this.kvStore.get(key);
+            if (oldValue == null) {
+                this.kvStore.put(key, value);
+            } else {
+                int newValue = oldValue + value;
+                this.kvStore.put(key, newValue);
+            }
+
+            context.commit();
         }
 
-        context.commit();
-    }
+        @Override
+        public void punctuate(long streamTime) {
+            KeyValueIterator<String, Integer> iter = this.kvStore.all();
+            while (iter.hasNext()) {
+                Entry<String, Integer> entry = iter.next();
+                System.out.println("[" + entry.key() + ", " + entry.value() + "]");
+            }
+        }
 
-    @Override
-    public void punctuate(long streamTime) {
-        KeyValueIterator<String, Integer> iter = this.kvStore.all();
-        while (iter.hasNext()) {
-            Entry<String, Integer> entry = iter.next();
-            System.out.println("[" + entry.key() + ", " + entry.value() + "]");
+        @Override
+        public void close() {
+            // do nothing
         }
     }
 
+    @SuppressWarnings("unchecked")
     @Override
-    public void close() {
-        // do nothing
-    }
+    public void topology() { ((KStream<String, Integer>) from("topic")).process(new MyProcessor()); }
 
     public static void main(String[] args) {
         KafkaStreaming streaming = new KafkaStreaming(
-            new SingleProcessorTopology(StatefulKStreamJob.class, args),
+            new StatefulKStreamJob(),
             new StreamingConfig(new Properties())
         );
         streaming.run();