Simplified join processor

stream/src/main/java/org/apache/kafka/streaming/KafkaStreaming.java

@@ -22,8 +22,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
-
 /**
  * Kafka Streaming allows for performing continuous computation on input coming from one or more input topics and
  * sends output to zero or more output topics.
@@ -118,5 +116,5 @@ public synchronized void close() {
         state = STOPPED;
 
         log.info("Stopped Kafka Stream process");
-   }
+    }
 }

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

@@ -31,47 +31,23 @@ private static abstract class Finder<K, T> {
         abstract Iterator<T> find(K key, long timestamp);
     }
 
-    private final String windowName1;
-    private final String windowName2;
+    private final String windowName;
     private final ValueJoiner<V1, V2, V> joiner;
 
-    private Processor processorForOtherStream = null;
-    public final ProcessorDef processorDefForOtherStream = new ProcessorDef() {
-        @Override
-        public Processor instance() {
-            return processorForOtherStream;
-        }
-    };
-
-    KStreamJoin(String windowName1, String windowName2, ValueJoiner<V1, V2, V> joiner) {
-        this.windowName1 = windowName1;
-        this.windowName2 = windowName2;
+    KStreamJoin(String windowName, ValueJoiner<V1, V2, V> joiner) {
+        this.windowName = windowName;
         this.joiner = joiner;
     }
 
     @Override
     public Processor instance() {
-        // create a processor instance for the other stream
-        processorForOtherStream = new KStreamJoinProcessor<K, V2, V1>(windowName1) {
-            @Override
-            protected void doJoin(K key, V2 value2, V1 value1) {
-                context.forward(key, joiner.apply(value1, value2));
-            }
-        };
-
-        // create a processor instance for the primary stream
-        return new KStreamJoinProcessor<K, V1, V2>(windowName2) {
-            @Override
-            protected void doJoin(K key, V1 value1, V2 value2) {
-                context.forward(key, joiner.apply(value1, value2));
-            }
-        };
+        return new KStreamJoinProcessor(windowName);
     }
 
-    private abstract class KStreamJoinProcessor<K, T1, T2> extends KStreamProcessor<K, T1> {
+    private class KStreamJoinProcessor extends KStreamProcessor<K, V1> {
 
         private final String windowName;
-        protected Finder<K, T2> finder;
+        protected Finder<K, V2> finder;
 
         public KStreamJoinProcessor(String windowName) {
             this.windowName = windowName;
@@ -86,27 +62,34 @@ public void init(ProcessorContext context) {
             if (!context.joinable())
                 throw new IllegalStateException("Streams are not joinable.");
 
-            final Window<K, T2> window = (Window<K, T2>) context.getStateStore(windowName);
+            final Window<K, V2> window = (Window<K, V2>) context.getStateStore(windowName);
 
-            this.finder = new Finder<K, T2>() {
-                Iterator<T2> find(K key, long timestamp) {
+            this.finder = new Finder<K, V2>() {
+                Iterator<V2> find(K key, long timestamp) {
                     return window.find(key, timestamp);
                 }
             };
         }
 
         @Override
-        public void process(K key, T1 value) {
+        public void process(K key, V1 value) {
             long timestamp = context.timestamp();
-            Iterator<T2> iter = finder.find(key, timestamp);
+            Iterator<V2> iter = finder.find(key, timestamp);
             if (iter != null) {
                 while (iter.hasNext()) {
-                    doJoin(key, value, iter.next());
+                    context.forward(key, joiner.apply(value, iter.next()));
                 }
             }
         }
+    }
 
-        abstract protected void doJoin(K key, T1 value1, T2 value2);
+    public static <T2, T1, R> ValueJoiner<T2, T1, R> reserveJoiner(final ValueJoiner<T1, T2, R> joiner) {
+        return new ValueJoiner<T2, T1, R>() {
+            @Override
+            public R apply(T2 value2, T1 value1) {
+                return joiner.apply(value1, value2);
+            }
+        };
     }
 
 }

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

@@ -37,15 +37,16 @@ public <V1, V2> KStream<K, V2> join(KStreamWindowed<K, V1> other, ValueJoiner<V,
         String thisWindowName = this.windowDef.name();
         String otherWindowName = ((KStreamWindowedImpl<K, V1>) other).windowDef.name();
 
+        KStreamJoin<K, V2, V, V1> joinThis = new KStreamJoin<>(otherWindowName, valueJoiner);
+        KStreamJoin<K, V2, V1, V> joinOther = new KStreamJoin<>(thisWindowName, KStreamJoin.reserveJoiner(valueJoiner));
+        KStreamPassThrough<K, V2> joinMerge = new KStreamPassThrough<>();
+
         String joinThisName = JOINTHIS_NAME + INDEX.getAndIncrement();
         String joinOtherName = JOINOTHER_NAME + INDEX.getAndIncrement();
         String joinMergeName = JOINMERGE_NAME + INDEX.getAndIncrement();
 
-        KStreamJoin<K, V2, V, V1> join = new KStreamJoin<>(thisWindowName, otherWindowName, valueJoiner);
-        KStreamPassThrough<K, V2> joinMerge = new KStreamPassThrough<>();
-
-        topology.addProcessor(joinThisName, join, this.name);
-        topology.addProcessor(joinOtherName, join.processorDefForOtherStream, ((KStreamImpl) other).name);
+        topology.addProcessor(joinThisName, joinThis, this.name);
+        topology.addProcessor(joinOtherName, joinOther, ((KStreamImpl) other).name);
         topology.addProcessor(joinMergeName, joinMerge, joinThisName, joinOtherName);
 
         return new KStreamImpl<>(topology, joinMergeName);