Stream task

stream/src/main/java/org/apache/kafka/streaming/examples/SimpleProcessJob.java

@@ -18,7 +18,7 @@
 package org.apache.kafka.streaming.examples;
 
 import org.apache.kafka.streaming.KafkaStreaming;
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.TopologyBuilder;
 import org.apache.kafka.streaming.StreamingConfig;
 import org.apache.kafka.streaming.processor.ProcessorContext;
@@ -29,7 +29,7 @@
 
 public class SimpleProcessJob {
 
-    private static class MyProcessor extends KafkaProcessor<String, Integer, Object, Object> {
+    private static class MyProcessor extends Processor<String, Integer, Object, Object> {
         private ProcessorContext context;
 
         public MyProcessor(String name) {

stream/src/main/java/org/apache/kafka/streaming/examples/StatefulProcessJob.java

@@ -20,7 +20,7 @@
 import org.apache.kafka.common.serialization.IntegerDeserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.streaming.KafkaStreaming;
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.TopologyBuilder;
 import org.apache.kafka.streaming.StreamingConfig;
 import org.apache.kafka.streaming.processor.ProcessorContext;
@@ -33,7 +33,7 @@
 
 public class StatefulProcessJob {
 
-    private static class MyProcessor extends KafkaProcessor<String, Integer, Object, Object> {
+    private static class MyProcessor extends Processor<String, Integer, Object, Object> {
         private ProcessorContext context;
         private KeyValueStore<String, Integer> kvStore;
 

stream/src/main/java/org/apache/kafka/streaming/kstream/KStream.java

@@ -19,7 +19,6 @@
 
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
-import org.apache.kafka.streaming.processor.KafkaProcessor;
 import org.apache.kafka.streaming.processor.Processor;
 
 /**

stream/src/main/java/org/apache/kafka/streaming/kstream/SlidingWindow.java

@@ -25,8 +25,9 @@
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streaming.kstream.internals.FilteredIterator;
 import org.apache.kafka.streaming.kstream.internals.WindowSupport;
+import org.apache.kafka.streaming.processor.internals.ProcessorContextImpl;
+import org.apache.kafka.streaming.processor.internals.RecordCollector;
 import org.apache.kafka.streaming.processor.ProcessorContext;
-import org.apache.kafka.streaming.processor.RecordCollector;
 import org.apache.kafka.streaming.processor.RestoreFunc;
 import org.apache.kafka.streaming.processor.internals.Stamped;
 
@@ -173,7 +174,7 @@ public void flush() {
         IntegerSerializer intSerializer = new IntegerSerializer();
         ByteArraySerializer byteArraySerializer = new ByteArraySerializer();
 
-        RecordCollector collector = context.recordCollector();
+        RecordCollector collector = ((ProcessorContextImpl)context).recordCollector();
 
         for (Map.Entry<K, ValueList<V>> entry : map.entrySet()) {
             ValueList<V> values = entry.getValue();

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamBranch.java

@@ -18,11 +18,11 @@
 package org.apache.kafka.streaming.kstream.internals;
 
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.ProcessorMetadata;
 import org.apache.kafka.streaming.kstream.Predicate;
 
-class KStreamBranch<K, V> extends KafkaProcessor<K, V, K, V> {
+class KStreamBranch<K, V> extends Processor<K, V, K, V> {
 
     private final Predicate<K, V>[] predicates;
 

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFilter.java

@@ -17,11 +17,11 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.kstream.Predicate;
 import org.apache.kafka.streaming.processor.ProcessorMetadata;
 
-class KStreamFilter<K, V> extends KafkaProcessor<K, V> {
+class KStreamFilter<K, V> extends Processor<K, V> {
 
     private final PredicateOut<K, V> predicateOut;
 

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMap.java

@@ -17,12 +17,12 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.ProcessorMetadata;
 import org.apache.kafka.streaming.kstream.KeyValue;
 import org.apache.kafka.streaming.kstream.KeyValueFlatMap;
 
-class KStreamFlatMap<K1, V1, K2, V2> extends KafkaProcessor<K1, V1> {
+class KStreamFlatMap<K1, V1, K2, V2> extends Processor<K1, V1> {
 
     private final KeyValueFlatMap<K1, V1, K2, V2> mapper;
 

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamFlatMapValues.java

@@ -17,11 +17,11 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.kstream.ValueMapper;
 import org.apache.kafka.streaming.processor.ProcessorMetadata;
 
-class KStreamFlatMapValues<K1, V1, V2> extends KafkaProcessor<K1, V1, K1, V2> {
+class KStreamFlatMapValues<K1, V1, V2> extends Processor<K1, V1, K1, V2> {
 
     private final ValueMapper<V1, ? extends Iterable<V2>> mapper;
 

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java

@@ -20,7 +20,6 @@
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streaming.kstream.KeyValueFlatMap;
-import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.ProcessorMetadata;
 import org.apache.kafka.streaming.processor.TopologyBuilder;
 import org.apache.kafka.streaming.kstream.KStreamWindowed;

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java

@@ -17,14 +17,14 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.ProcessorContext;
 import org.apache.kafka.streaming.kstream.ValueJoiner;
 import org.apache.kafka.streaming.kstream.Window;
 
 import java.util.Iterator;
 
-class KStreamJoin<K, V, V1, V2> extends KafkaProcessor<K, V1, K, V> {
+class KStreamJoin<K, V, V1, V2> extends Processor<K, V1, K, V> {
 
     private static final String JOIN_NAME = "KAFKA-JOIN";
     private static final String JOIN_OTHER_NAME = "KAFKA-JOIN-OTHER";
@@ -38,7 +38,7 @@ private static abstract class Finder<K, T> {
     private final Finder<K, V1> finder1;
     private final Finder<K, V2> finder2;
     private final ValueJoiner<V1, V2, V> joiner;
-    final KafkaProcessor<K, V2, K, V> processorForOtherStream;
+    final Processor<K, V2, K, V> processorForOtherStream;
 
     private ProcessorContext context;
 
@@ -100,8 +100,8 @@ public void process(K key, V1 value) {
         }
     }
 
-    private KafkaProcessor<K, V2, K, V> processorForOther() {
-        return new KafkaProcessor<K, V2, K, V>(JOIN_OTHER_NAME) {
+    private Processor<K, V2, K, V> processorForOther() {
+        return new Processor<K, V2, K, V>(JOIN_OTHER_NAME) {
 
             @SuppressWarnings("unchecked")
             @Override

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMap.java

@@ -17,13 +17,12 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.kstream.KeyValue;
 import org.apache.kafka.streaming.kstream.KeyValueMapper;
-import org.apache.kafka.streaming.processor.ProcessorContext;
 import org.apache.kafka.streaming.processor.ProcessorMetadata;
 
-class KStreamMap<K1, V1, K2, V2> extends KafkaProcessor<K1, V1> {
+class KStreamMap<K1, V1, K2, V2> extends Processor<K1, V1> {
 
     private final KeyValueMapper<K1, V1, K2, V2> mapper;
 

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamMapValues.java

@@ -17,11 +17,11 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.kstream.ValueMapper;
 import org.apache.kafka.streaming.processor.ProcessorMetadata;
 
-class KStreamMapValues<K1, V1, V2> extends KafkaProcessor<K1, V1, K1, V2> {
+class KStreamMapValues<K1, V1, V2> extends Processor<K1, V1, K1, V2> {
 
     private final ValueMapper<V1, V2> mapper;
 

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamProcessor.java

@@ -17,11 +17,10 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.streaming.processor.KafkaProcessor;
 import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.ProcessorMetadata;
 
-public class KStreamProcessor<K, V> extends KafkaProcessor<K, V> {
+public class KStreamProcessor<K, V> extends Processor<K, V> {
 
     private final Processor<K, V> processor;
 

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamSend.java

@@ -18,11 +18,11 @@
 package org.apache.kafka.streaming.kstream.internals;
 
 import org.apache.kafka.common.serialization.Serializer;
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.ProcessorMetadata;
 import org.apache.kafka.streaming.processor.ProcessorContext;
 
-class KStreamSend<K, V> extends KafkaProcessor<K, V, Object, Object> {
+class KStreamSend<K, V> extends Processor<K, V, Object, Object> {
 
     private ProcessorContext context;
 

stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindow.java

@@ -17,15 +17,15 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.streaming.processor.KafkaProcessor;
+import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.TopologyBuilder;
 import org.apache.kafka.streaming.processor.ProcessorContext;
 import org.apache.kafka.streaming.kstream.KStream;
 import org.apache.kafka.streaming.kstream.KStreamWindowed;
 import org.apache.kafka.streaming.kstream.ValueJoiner;
 import org.apache.kafka.streaming.kstream.Window;
 
-public class KStreamWindow<K, V> extends KafkaProcessor<K, V, K, V> {
+public class KStreamWindow<K, V> extends Processor<K, V, K, V> {
 
     public static final class KStreamWindowedImpl<K, V> extends KStreamImpl<K, V> implements KStreamWindowed<K, V> {
 

stream/src/main/java/org/apache/kafka/streaming/processor/Processor.java

@@ -23,5 +23,7 @@ public interface Processor<K, V> {
 
     void process(K key, V value);
 
+    void punctuate(long streamTime);
+
     void close();
 }

stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorContext.java

@@ -63,13 +63,6 @@ public interface ProcessorContext {
      */
     Deserializer<?> valueDeserializer();
 
-    /**
-     * Returns a RecordCollector
-     *
-     * @return RecordCollector
-     */
-    RecordCollector recordCollector();
-
     /**
      * Returns the state directory for the partition.
      *
@@ -91,19 +84,10 @@ public interface ProcessorContext {
      */
     void register(StateStore store, RestoreFunc restoreFunc);
 
-    /**
-     * Flush the local state of this context
-     */
-    void flush();
+    void schedule(Processor processor, long interval);
 
     <K, V> void forward(K key, V value);
 
-    void send(String topic, Object key, Object value);
-
-    void send(String topic, Object key, Object value, Serializer<Object> keySerializer, Serializer<Object> valSerializer);
-
-    void schedule(KafkaProcessor processor, long interval);
-
     void commit();
 
     String topic();

stream/src/main/java/org/apache/kafka/streaming/processor/Punctuator.java → stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorFactory.java

@@ -17,7 +17,7 @@
 
 package org.apache.kafka.streaming.processor;
 
-public interface Punctuator {
+public interface ProcessorFactory {
 
-    void punctuate(long streamTime);
+    Processor build();
 }