diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java
index 688cb65f9fc2a..91ab5fef36ef9 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamJoin.java
@@ -17,20 +17,13 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.streaming.kstream.ValueJoiner;
 import org.apache.kafka.streaming.kstream.Window;
 import org.apache.kafka.streaming.processor.Processor;
 import org.apache.kafka.streaming.processor.ProcessorContext;
 import org.apache.kafka.streaming.processor.ProcessorDef;
-import org.apache.kafka.streaming.processor.internals.ProcessorContextImpl;
 
-import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
 
 class KStreamJoin<K, V, V1, V2> implements ProcessorDef {
 
@@ -74,7 +67,7 @@ public void init(ProcessorContext context) {
             super.init(context);
 
             // check if these two streams are joinable
-            if (!joinable())
+            if (!context.joinable())
                 throw new IllegalStateException("Streams are not joinable.");
 
             final Window<K, V1> window1 = (Window<K, V1>) context.getStateStore(windowName1);
@@ -129,34 +122,6 @@ public void process(K key, V1 value) {
             }
         }
 
-        private boolean joinable() {
-            Set<TopicPartition> partitions = ((ProcessorContextImpl) this.context).task().partitions();
-            Map<Integer, List<String>> partitionsById = new HashMap<>();
-            int firstId = -1;
-            for (TopicPartition partition : partitions) {
-                if (!partitionsById.containsKey(partition.partition())) {
-                    partitionsById.put(partition.partition(), new ArrayList<String>());
-                }
-                partitionsById.get(partition.partition()).add(partition.topic());
-
-                if (firstId < 0)
-                    firstId = partition.partition();
-            }
-
-            List<String> topics = partitionsById.get(firstId);
-            for (List<String> topicsPerPartition : partitionsById.values()) {
-                if (topics.size() != topicsPerPartition.size())
-                    return false;
-
-                for (String topic : topicsPerPartition) {
-                    if (!topics.contains(topic))
-                        return false;
-                }
-            }
-
-            return true;
-        }
-
         // TODO: use the "outer-stream" topic as the resulted join stream topic
         private void doJoin(K key, V1 value1, V2 value2) {
             context.forward(key, joiner.apply(value1, value2));