diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index c526bd04ea642..696b90771d089 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -103,12 +103,12 @@
 		</subpackage>
 	</subpackage>
 
-	<subpackage name="stream">
+	<subpackage name="streaming">
 		<allow pkg="org.apache.kafka.common"/>
 		<allow pkg="org.apache.kafka.test"/>
+		<allow pkg="org.apache.kafka.clients"/>
 		<allow pkg="org.apache.kafka.clients.producer" exact-match="true"/>
 		<allow pkg="org.apache.kafka.clients.consumer" exact-match="true"/>
-		<allow pkg="org.apache.kafka.clients.processor"/>
 
 		<allow pkg="org.apache.kafka.streaming"/>
 
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
index b07e760c28b17..5609dec7467ab 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java
@@ -21,6 +21,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -38,12 +39,14 @@ public class MockConsumer<K, V> implements Consumer<K, V> {
     private final Map<String, List<PartitionInfo>> partitions;
     private final SubscriptionState subscriptions;
     private Map<TopicPartition, List<ConsumerRecord<K, V>>> records;
+    private Set<TopicPartition> paused;
     private boolean closed;
 
     public MockConsumer(OffsetResetStrategy offsetResetStrategy) {
         this.subscriptions = new SubscriptionState(offsetResetStrategy);
-        this.partitions = new HashMap<String, List<PartitionInfo>>();
-        this.records = new HashMap<TopicPartition, List<ConsumerRecord<K, V>>>();
+        this.partitions = new HashMap<>();
+        this.records = new HashMap<>();
+        this.paused = new HashSet<>();
         this.closed = false;
     }
     
@@ -197,14 +200,18 @@ public synchronized void updatePartitions(String topic, List<PartitionInfo> part
 
     @Override
     public void pause(TopicPartition... partitions) {
-        for (TopicPartition partition : partitions)
+        for (TopicPartition partition : partitions) {
             subscriptions.pause(partition);
+            paused.add(partition);
+        }
     }
 
     @Override
     public void resume(TopicPartition... partitions) {
-        for (TopicPartition partition : partitions)
+        for (TopicPartition partition : partitions) {
             subscriptions.resume(partition);
+            paused.remove(partition);
+        }
     }
 
     @Override
@@ -218,6 +225,10 @@ public void wakeup() {
 
     }
 
+    public Set<TopicPartition> paused() {
+        return paused;
+    }
+
     private void ensureNotClosed() {
         if (this.closed)
             throw new IllegalStateException("This consumer has already been closed.");
diff --git a/stream/src/main/java/org/apache/kafka/streaming/StreamingConfig.java b/stream/src/main/java/org/apache/kafka/streaming/StreamingConfig.java
index 5b59197388dc3..cad0b6dfec3bb 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/StreamingConfig.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/StreamingConfig.java
@@ -17,6 +17,7 @@
 
 package org.apache.kafka.streaming;
 
+import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.config.AbstractConfig;
@@ -25,7 +26,6 @@
 import org.apache.kafka.common.config.ConfigDef.Type;
 
 import java.util.Map;
-import java.util.Properties;
 
 public class StreamingConfig extends AbstractConfig {
 
@@ -80,8 +80,10 @@ public class StreamingConfig extends AbstractConfig {
     /** <code>value.deserializer</code> */
     public static final String VALUE_DESERIALIZER_CLASS_CONFIG = ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
 
-
-
+    /**
+     * <code>bootstrap.servers</code>
+     */
+    public static final String BOOTSTRAP_SERVERS_CONFIG = CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
 
     private static final String SYSTEM_TEMP_DIRECTORY = System.getProperty("java.io.tmpdir");
 
@@ -100,12 +102,12 @@ public class StreamingConfig extends AbstractConfig {
                                         Type.LONG,
                                         100,
                                         Importance.LOW,
-                                    POLL_MS_DOC)
+                                        POLL_MS_DOC)
                                 .define(NUM_STREAM_THREADS_CONFIG,
-                                    Type.INT,
-                                    1,
-                                    Importance.LOW,
-                                    NUM_STREAM_THREADS_DOC)
+                                        Type.INT,
+                                        1,
+                                        Importance.LOW,
+                                        NUM_STREAM_THREADS_DOC)
                                 .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
                                         Type.INT,
                                         1000,
@@ -115,7 +117,7 @@ public class StreamingConfig extends AbstractConfig {
                                         Type.LONG,
                                         60000,
                                         Importance.LOW,
-                                    STATE_CLEANUP_DELAY_MS_DOC)
+                                        STATE_CLEANUP_DELAY_MS_DOC)
                                 .define(TOTAL_RECORDS_TO_PROCESS,
                                         Type.LONG,
                                         -1L,
@@ -145,28 +147,32 @@ public class StreamingConfig extends AbstractConfig {
                                 .define(TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
                                         Type.CLASS,
                                         Importance.HIGH,
-                                        TIMESTAMP_EXTRACTOR_CLASS_DOC);
+                                        TIMESTAMP_EXTRACTOR_CLASS_DOC)
+                                .define(BOOTSTRAP_SERVERS_CONFIG,
+                                        Type.STRING,
+                                        Importance.HIGH,
+                                        CommonClientConfigs.BOOSTRAP_SERVERS_DOC);
     }
 
     public StreamingConfig(Map<?, ?> props) {
         super(CONFIG, props);
     }
 
-    public Properties getConsumerProperties() {
-        Properties props = new Properties();
+    public Map<String, Object> getConsumerConfigs() {
+        Map<String, Object> props = this.originals();
 
         // set consumer default property values
-        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
-        props.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "range");
+        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "range");
 
         return props;
     }
 
-    public Properties getProducerProperties() {
-        Properties props = new Properties();
+    public Map<String, Object> getProducerConfigs() {
+        Map<String, Object> props = this.originals();
 
         // set producer default property values
-        props.setProperty(ProducerConfig.LINGER_MS_CONFIG, "100");
+        props.put(ProducerConfig.LINGER_MS_CONFIG, "100");
 
         return props;
     }
diff --git a/stream/src/main/java/org/apache/kafka/streaming/examples/ProcessorJob.java b/stream/src/main/java/org/apache/kafka/streaming/examples/ProcessorJob.java
new file mode 100644
index 0000000000000..c7defec9115db
--- /dev/null
+++ b/stream/src/main/java/org/apache/kafka/streaming/examples/ProcessorJob.java
@@ -0,0 +1,101 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.streaming.examples;
+
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+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.Processor;
+import org.apache.kafka.streaming.processor.ProcessorDef;
+import org.apache.kafka.streaming.processor.TopologyBuilder;
+import org.apache.kafka.streaming.processor.ProcessorContext;
+import org.apache.kafka.streaming.StreamingConfig;
+import org.apache.kafka.streaming.state.Entry;
+import org.apache.kafka.streaming.state.InMemoryKeyValueStore;
+import org.apache.kafka.streaming.state.KeyValueIterator;
+import org.apache.kafka.streaming.state.KeyValueStore;
+
+import java.util.Properties;
+
+public class ProcessorJob {
+
+    private static class MyProcessorDef implements ProcessorDef {
+
+        @Override
+        public Processor<String, Integer> instance() {
+            return new Processor<String, Integer>() {
+                private ProcessorContext context;
+                private KeyValueStore<String, Integer> kvStore;
+
+                @Override
+                public void init(ProcessorContext context) {
+                    this.context = context;
+                    this.context.schedule(this, 1000);
+                    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);
+                    }
+
+                    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() + "]");
+
+                        context.forward(entry.key(), entry.value());
+                    }
+                }
+
+                @Override
+                public void close() {
+                    this.kvStore.close();
+                }
+            };
+        }
+    }
+
+    public static void main(String[] args) throws Exception {
+        StreamingConfig config = new StreamingConfig(new Properties());
+        TopologyBuilder builder = new TopologyBuilder();
+
+        builder.addSource("SOURCE", new StringDeserializer(), new IntegerDeserializer(), "topic-source");
+
+        builder.addProcessor("PROCESS", new MyProcessorDef(), null, "SOURCE");
+
+        builder.addSink("SINK", "topic-sink", new StringSerializer(), new IntegerSerializer(), "PROCESS");
+
+        KafkaStreaming streaming = new KafkaStreaming(builder, config);
+        streaming.run();
+    }
+}
diff --git a/stream/src/main/java/org/apache/kafka/streaming/examples/SimpleProcessJob.java b/stream/src/main/java/org/apache/kafka/streaming/examples/SimpleProcessJob.java
deleted file mode 100644
index 22d6425e7aadf..0000000000000
--- a/stream/src/main/java/org/apache/kafka/streaming/examples/SimpleProcessJob.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kafka.streaming.examples;
-
-import org.apache.kafka.streaming.KafkaStreaming;
-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;
-import org.apache.kafka.common.serialization.IntegerDeserializer;
-import org.apache.kafka.common.serialization.StringDeserializer;
-
-import java.util.Properties;
-
-public class SimpleProcessJob {
-
-    private static class MyProcessor extends Processor<String, Integer, Object, Object> {
-        private ProcessorContext context;
-
-        public MyProcessor(String name) {
-            super(name);
-        }
-
-        @Override
-        public void init(ProcessorContext context) {
-            this.context = context;
-        }
-
-        @Override
-        public void process(String key, Integer value) {
-            System.out.println("[" + key + ", " + value + "]");
-
-            context.commit();
-
-            context.send("topic-dest", key, value);
-        }
-
-        @Override
-        public void close() {
-            // do nothing
-        }
-    }
-
-    public static void main(String[] args) throws Exception {
-        StreamingConfig config = new StreamingConfig(new Properties());
-        TopologyBuilder builder = new TopologyBuilder();
-
-        builder.addSource("SOURCE", new StringDeserializer(), new IntegerDeserializer(), "topic-source");
-        builder.addProcessor("PROCESS", MyProcessor.class, null, "SOURCE");
-
-        KafkaStreaming streaming = new KafkaStreaming(builder, config);
-        streaming.run();
-    }
-}
diff --git a/stream/src/main/java/org/apache/kafka/streaming/examples/StatefulProcessJob.java b/stream/src/main/java/org/apache/kafka/streaming/examples/StatefulProcessJob.java
deleted file mode 100644
index 6a01551b4972a..0000000000000
--- a/stream/src/main/java/org/apache/kafka/streaming/examples/StatefulProcessJob.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kafka.streaming.examples;
-
-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.Processor;
-import org.apache.kafka.streaming.processor.TopologyBuilder;
-import org.apache.kafka.streaming.StreamingConfig;
-import org.apache.kafka.streaming.processor.ProcessorContext;
-import org.apache.kafka.streaming.state.Entry;
-import org.apache.kafka.streaming.state.InMemoryKeyValueStore;
-import org.apache.kafka.streaming.state.KeyValueIterator;
-import org.apache.kafka.streaming.state.KeyValueStore;
-
-import java.util.Properties;
-
-public class StatefulProcessJob {
-
-    private static class MyProcessor extends Processor<String, Integer, Object, Object> {
-        private ProcessorContext context;
-        private KeyValueStore<String, Integer> kvStore;
-
-        public MyProcessor(String name) {
-            super(name);
-        }
-
-        @Override
-        public void init(ProcessorContext context) {
-            this.context = context;
-            this.context.schedule(this, 1000);
-
-            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);
-            }
-
-            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 close() {
-            this.kvStore.close();
-        }
-    }
-
-    public static void main(String[] args) throws Exception {
-        StreamingConfig config = new StreamingConfig(new Properties());
-        TopologyBuilder builder = new TopologyBuilder();
-
-        builder.addSource("SOURCE", new StringDeserializer(), new IntegerDeserializer(), "topic-source");
-        builder.addProcessor("PROCESS", MyProcessor.class, null, "SOURCE");
-
-        KafkaStreaming streaming = new KafkaStreaming(builder, config);
-        streaming.run();
-    }
-}
diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java
index 4cf99cf6ed842..bf7456e5236df 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamImpl.java
@@ -61,8 +61,6 @@ public class KStreamImpl<K, V> implements KStream<K, V> {
 
     public static final String SEND_NAME = "KAFKA-SEND-";
 
-    public static final String WINDOW_NAME = "KAFKA-WINDOW-";
-
     public static final AtomicInteger INDEX = new AtomicInteger(1);
 
     protected TopologyBuilder topology;
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 2c0020b12afb8..688cb65f9fc2a 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,13 +17,20 @@
 
 package org.apache.kafka.streaming.kstream.internals;
 
-import org.apache.kafka.streaming.processor.Processor;
-import org.apache.kafka.streaming.processor.ProcessorContext;
+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 {
 
@@ -67,7 +74,7 @@ public void init(ProcessorContext context) {
             super.init(context);
 
             // check if these two streams are joinable
-            if (!context.joinable())
+            if (!joinable())
                 throw new IllegalStateException("Streams are not joinable.");
 
             final Window<K, V1> window1 = (Window<K, V1>) context.getStateStore(windowName1);
@@ -122,6 +129,34 @@ 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));
diff --git a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedImpl.java b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedImpl.java
index 62351a4e3607f..ec1e2c10f1f67 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedImpl.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/kstream/internals/KStreamWindowedImpl.java
@@ -1,3 +1,20 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package org.apache.kafka.streaming.kstream.internals;
 
 import org.apache.kafka.streaming.kstream.KStream;
@@ -6,9 +23,6 @@
 import org.apache.kafka.streaming.kstream.WindowDef;
 import org.apache.kafka.streaming.processor.TopologyBuilder;
 
-/**
- * Created by yasuhiro on 8/27/15.
- */
 public final class KStreamWindowedImpl<K, V> extends KStreamImpl<K, V> implements KStreamWindowed<K, V> {
 
     private final WindowDef<K, V> windowDef;
diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorContext.java b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorContext.java
index db78fe52e365c..06e219ae45304 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorContext.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/processor/ProcessorContext.java
@@ -25,9 +25,6 @@
 
 public interface ProcessorContext {
 
-    // TODO: this is better moved to a KStreamContext
-    boolean joinable();
-
     /**
      * Returns the partition group id
      *
@@ -77,6 +74,11 @@ public interface ProcessorContext {
      */
     Metrics metrics();
 
+    /**
+     * Check if this process's incoming streams are joinable
+     */
+    boolean joinable();
+
     /**
      * Registers and possibly restores the specified storage engine.
      *
diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java b/stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java
index 596a4c6c49885..8bf8ee8c152d5 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/processor/TopologyBuilder.java
@@ -175,7 +175,7 @@ public ProcessorTopology build() {
                         processorMap.get(parent).addChild(node);
                     }
                 } else if (factory instanceof SourceNodeFactory) {
-                    for (String topic : ((SourceNodeFactory)factory).topics) {
+                    for (String topic : ((SourceNodeFactory) factory).topics) {
                         topicSourceMap.put(topic, (SourceNode) node);
                     }
                 } else if (factory instanceof SinkNodeFactory) {
diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java
index a4954b408f806..d4fc4e63a3beb 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorContextImpl.java
@@ -79,7 +79,7 @@ public ProcessorContextImpl(int id,
         File stateFile = new File(config.getString(StreamingConfig.STATE_DIR_CONFIG), Integer.toString(id));
 
         Consumer restoreConsumer = new KafkaConsumer<>(
-            config.getConsumerProperties(),
+            config.getConsumerConfigs(),
             null /* no callback for restore consumer */,
             new ByteArrayDeserializer(),
             new ByteArrayDeserializer());
@@ -97,6 +97,10 @@ public RecordCollector recordCollector() {
         return this.collector;
     }
 
+    public StreamTask task() {
+        return this.task;
+    }
+
     @Override
     public boolean joinable() {
         Set<TopicPartition> partitions = this.task.partitions();
@@ -169,6 +173,7 @@ public void register(StateStore store, RestoreFunc restoreFunc) {
         stateMgr.register(store, restoreFunc);
     }
 
+    @Override
     public StateStore getStateStore(String name) {
         return stateMgr.getStore(name);
     }
diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorTopology.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorTopology.java
index bf4ddd796ece8..9ebbe5a5702d8 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorTopology.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/ProcessorTopology.java
@@ -20,7 +20,6 @@
 import org.apache.kafka.streaming.processor.ProcessorContext;
 
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/SinkNode.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/SinkNode.java
index 14bec7ea5fc65..412824c517578 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/SinkNode.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/SinkNode.java
@@ -21,9 +21,6 @@
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streaming.processor.ProcessorContext;
 
-import java.util.ArrayList;
-import java.util.List;
-
 public class SinkNode<K, V> extends ProcessorNode<K, V> {
 
     private final String topic;
@@ -48,7 +45,7 @@ public void init(ProcessorContext context) {
     @Override
     public void process(K key, V value) {
         // send to all the registered topics
-        RecordCollector collector = ((ProcessorContextImpl)context).recordCollector();
+        RecordCollector collector = ((ProcessorContextImpl) context).recordCollector();
         collector.send(new ProducerRecord<>(topic, key, value), keySerializer, valSerializer);
     }
 
diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamTask.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamTask.java
index 7f48c1864c46b..6a5b868fade10 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamTask.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamTask.java
@@ -105,8 +105,8 @@ public StreamTask(int id,
 
         // create the record recordCollector that maintains the produced offsets
         this.recordCollector = new RecordCollector(producer,
-            (Serializer<Object>) config.getConfiguredInstance(StreamingConfig.KEY_DESERIALIZER_CLASS_CONFIG, Serializer.class),
-            (Serializer<Object>) config.getConfiguredInstance(StreamingConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Serializer.class));
+            (Serializer<Object>) config.getConfiguredInstance(StreamingConfig.KEY_SERIALIZER_CLASS_CONFIG, Serializer.class),
+            (Serializer<Object>) config.getConfiguredInstance(StreamingConfig.VALUE_SERIALIZER_CLASS_CONFIG, Serializer.class));
 
         // initialize the topology with its own context
         try {
@@ -156,7 +156,7 @@ public void addRecords(TopicPartition partition, Iterator<ConsumerRecord<byte[],
 
         // if after adding these records, its partition queue's buffered size has been
         // increased beyond the threshold, we can then pause the consumption for this partition
-        if (partitionGroup.numbuffered(partition) == this.maxBufferedSize) {
+        if (partitionGroup.numbuffered(partition) > this.maxBufferedSize) {
             consumer.pause(partition);
         }
     }
diff --git a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamThread.java b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamThread.java
index 5807fbf9d52a8..791b5cb1c295c 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamThread.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/processor/internals/StreamThread.java
@@ -93,11 +93,11 @@ public StreamThread(TopologyBuilder builder, StreamingConfig config) throws Exce
         this.builder = builder;
 
         // create the producer and consumer clients
-        this.producer = new KafkaProducer<>(config.getProducerProperties(),
+        this.producer = new KafkaProducer<>(config.getProducerConfigs(),
             new ByteArraySerializer(),
             new ByteArraySerializer());
 
-        this.consumer = new KafkaConsumer<>(config.getConsumerProperties(),
+        this.consumer = new KafkaConsumer<>(config.getConsumerConfigs(),
             rebalanceCallback,
             new ByteArrayDeserializer(),
             new ByteArrayDeserializer());
diff --git a/stream/src/main/java/org/apache/kafka/streaming/state/MeteredKeyValueStore.java b/stream/src/main/java/org/apache/kafka/streaming/state/MeteredKeyValueStore.java
index 93f7c093a7107..805de51c8acab 100644
--- a/stream/src/main/java/org/apache/kafka/streaming/state/MeteredKeyValueStore.java
+++ b/stream/src/main/java/org/apache/kafka/streaming/state/MeteredKeyValueStore.java
@@ -213,7 +213,7 @@ public void flush() {
     }
 
     private void logChange() {
-        RecordCollector collector = ((ProcessorContextImpl)context).recordCollector();
+        RecordCollector collector = ((ProcessorContextImpl) context).recordCollector();
         Serializer<K> keySerializer = (Serializer<K>) context.keySerializer();
         Serializer<V> valueSerializer = (Serializer<V>) context.valueSerializer();
 
diff --git a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/FilteredIteratorTest.java b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/FilteredIteratorTest.java
index ff8ed098f0977..6c71742403d1a 100644
--- a/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/FilteredIteratorTest.java
+++ b/stream/src/test/java/org/apache/kafka/streaming/kstream/internals/FilteredIteratorTest.java
@@ -19,7 +19,6 @@
 
 import static org.junit.Assert.assertEquals;
 
-import org.apache.kafka.streaming.kstream.internals.FilteredIterator;
 import org.junit.Test;
 
 import java.util.ArrayList;
diff --git a/stream/src/test/java/org/apache/kafka/streaming/processor/internals/StreamGroupTest.java b/stream/src/test/java/org/apache/kafka/streaming/processor/internals/StreamGroupTest.java
deleted file mode 100644
index ac82bab0cd538..0000000000000
--- a/stream/src/test/java/org/apache/kafka/streaming/processor/internals/StreamGroupTest.java
+++ /dev/null
@@ -1,162 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.kafka.streaming.processor.internals;
-
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.streaming.processor.TimestampExtractor;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.serialization.Deserializer;
-import org.apache.kafka.common.serialization.IntegerDeserializer;
-import org.apache.kafka.common.serialization.IntegerSerializer;
-import org.apache.kafka.common.serialization.Serializer;
-import org.apache.kafka.streaming.processor.internals.StreamGroup;
-import org.apache.kafka.test.MockProcessorContext;
-import org.apache.kafka.test.MockSourceNode;
-import org.junit.Test;
-
-import java.util.Arrays;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-public class StreamGroupTest {
-
-    private static Serializer<Integer> serializer = new IntegerSerializer();
-    private static Deserializer<Integer> deserializer = new IntegerDeserializer();
-
-    @SuppressWarnings("unchecked")
-    @Test
-    public void testAddPartition() {
-
-        MockIngestor mockIngestor = new MockIngestor();
-
-        StreamGroup streamGroup = new StreamGroup(
-            new MockProcessorContext(serializer, deserializer),
-            mockIngestor,
-            new TimestampExtractor() {
-                public long extract(String topic, Object key, Object value) {
-                    if (topic.equals("topic1"))
-                        return ((Integer) key).longValue();
-                    else
-                        return ((Integer) key).longValue() / 10L + 5L;
-                }
-            },
-            3
-        );
-
-        TopicPartition partition1 = new TopicPartition("topic1", 1);
-        TopicPartition partition2 = new TopicPartition("topic2", 1);
-        MockSourceNode<Integer, Integer> source1 = new MockSourceNode(deserializer, deserializer);
-        MockSourceNode<Integer, Integer> source2 = new MockSourceNode(deserializer, deserializer);
-        MockSourceNode<Integer, Integer> source3 = new MockSourceNode(deserializer, deserializer);
-
-        streamGroup.addPartition(partition1, source1);
-        mockIngestor.addPartitionStreamToGroup(streamGroup, partition1);
-
-        streamGroup.addPartition(partition2, source2);
-        mockIngestor.addPartitionStreamToGroup(streamGroup, partition2);
-
-        Exception exception = null;
-        try {
-            streamGroup.addPartition(partition1, source3);
-        } catch (Exception ex) {
-            exception = ex;
-        }
-        assertTrue(exception != null);
-
-        byte[] recordValue = serializer.serialize(null, new Integer(10));
-
-        mockIngestor.addRecords(partition1, records(
-            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 1, serializer.serialize(partition1.topic(), new Integer(10)), recordValue),
-            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 2, serializer.serialize(partition1.topic(), new Integer(20)), recordValue)
-        ));
-
-        mockIngestor.addRecords(partition2, records(
-            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 1, serializer.serialize(partition1.topic(), new Integer(300)), recordValue),
-            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 2, serializer.serialize(partition1.topic(), new Integer(400)), recordValue),
-            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 3, serializer.serialize(partition1.topic(), new Integer(500)), recordValue),
-            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 4, serializer.serialize(partition1.topic(), new Integer(600)), recordValue)
-        ));
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 1);
-        assertEquals(source2.numReceived, 0);
-
-        assertEquals(mockIngestor.paused.size(), 1);
-        assertTrue(mockIngestor.paused.contains(partition2));
-
-        mockIngestor.addRecords(partition1, records(
-            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 3, serializer.serialize(partition1.topic(), new Integer(30)), recordValue),
-            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 4, serializer.serialize(partition1.topic(), new Integer(40)), recordValue),
-            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 5, serializer.serialize(partition1.topic(), new Integer(50)), recordValue)
-        ));
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 2);
-        assertEquals(source2.numReceived, 0);
-
-        assertEquals(mockIngestor.paused.size(), 2);
-        assertTrue(mockIngestor.paused.contains(partition1));
-        assertTrue(mockIngestor.paused.contains(partition2));
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 3);
-        assertEquals(source2.numReceived, 0);
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 3);
-        assertEquals(source2.numReceived, 1);
-
-        assertEquals(mockIngestor.paused.size(), 1);
-        assertTrue(mockIngestor.paused.contains(partition2));
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 4);
-        assertEquals(source2.numReceived, 1);
-
-        assertEquals(mockIngestor.paused.size(), 1);
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 4);
-        assertEquals(source2.numReceived, 2);
-
-        assertEquals(mockIngestor.paused.size(), 0);
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 5);
-        assertEquals(source2.numReceived, 2);
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 5);
-        assertEquals(source2.numReceived, 3);
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 5);
-        assertEquals(source2.numReceived, 4);
-
-        assertEquals(mockIngestor.paused.size(), 0);
-
-        streamGroup.process();
-        assertEquals(source1.numReceived, 5);
-        assertEquals(source2.numReceived, 4);
-    }
-
-    private Iterable<ConsumerRecord<byte[], byte[]>> records(ConsumerRecord<byte[], byte[]>... recs) {
-        return Arrays.asList(recs);
-    }
-}
diff --git a/stream/src/test/java/org/apache/kafka/streaming/processor/internals/StreamTaskTest.java b/stream/src/test/java/org/apache/kafka/streaming/processor/internals/StreamTaskTest.java
new file mode 100644
index 0000000000000..c269d47bc2d49
--- /dev/null
+++ b/stream/src/test/java/org/apache/kafka/streaming/processor/internals/StreamTaskTest.java
@@ -0,0 +1,182 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.streaming.processor.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streaming.StreamingConfig;
+import org.apache.kafka.test.MockSourceNode;
+import org.junit.Test;
+import org.junit.Before;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class StreamTaskTest {
+
+    private final Serializer<Integer> intSerializer = new IntegerSerializer();
+    private final Deserializer<Integer> intDeserializer = new IntegerDeserializer();
+    private final Serializer<byte[]> bytesSerializer = new ByteArraySerializer();
+
+    private final TopicPartition partition1 = new TopicPartition("topic1", 1);
+    private final TopicPartition partition2 = new TopicPartition("topic2", 1);
+    private final HashSet<TopicPartition> partitions = new HashSet<>(Arrays.asList(partition1, partition2));
+
+    private final MockSourceNode source1 = new MockSourceNode<>(intDeserializer, intDeserializer);
+    private final MockSourceNode source2 = new MockSourceNode<>(intDeserializer, intDeserializer);
+    private final ProcessorTopology topology = new ProcessorTopology(
+        Arrays.asList((ProcessorNode) source1, (ProcessorNode) source2),
+        new HashMap<String, SourceNode>() {
+            {
+                put("topic1", source1);
+                put("topic2", source2);
+            }
+        },
+        Collections.<String, SinkNode>emptyMap()
+    );
+
+    private final StreamingConfig config = new StreamingConfig(new Properties() {
+        {
+            setProperty(StreamingConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
+            setProperty(StreamingConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+            setProperty(StreamingConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
+            setProperty(StreamingConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+            setProperty(StreamingConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, "org.apache.kafka.test.MockTimestampExtractor");
+            setProperty(StreamingConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:2171");
+            setProperty(StreamingConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG, "3");
+        }
+    });
+
+    private final MockConsumer<byte[], byte[]> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+    private final MockProducer<byte[], byte[]> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer);
+    private final StreamTask task = new StreamTask(0, consumer, producer, partitions, topology, config);
+
+    @Before
+    public void setup() {
+        consumer.subscribe(partition1, partition2);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void testProcessOrder() {
+        byte[] recordValue = intSerializer.serialize(null, 10);
+
+        task.addRecords(partition1, records(
+            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 1, intSerializer.serialize(partition1.topic(), 10), recordValue),
+            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 2, intSerializer.serialize(partition1.topic(), 20), recordValue),
+            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 3, intSerializer.serialize(partition1.topic(), 30), recordValue)
+        ));
+
+        task.addRecords(partition2, records(
+            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 1, intSerializer.serialize(partition1.topic(), 25), recordValue),
+            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 2, intSerializer.serialize(partition1.topic(), 35), recordValue),
+            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 3, intSerializer.serialize(partition1.topic(), 45), recordValue)
+        ));
+
+        assertEquals(task.process(), 5);
+        assertEquals(source1.numReceived, 1);
+        assertEquals(source2.numReceived, 0);
+
+        assertEquals(task.process(), 4);
+        assertEquals(source1.numReceived, 1);
+        assertEquals(source2.numReceived, 1);
+
+        assertEquals(task.process(), 3);
+        assertEquals(source1.numReceived, 2);
+        assertEquals(source2.numReceived, 1);
+
+        assertEquals(task.process(), 2);
+        assertEquals(source1.numReceived, 3);
+        assertEquals(source2.numReceived, 1);
+
+        assertEquals(task.process(), 1);
+        assertEquals(source1.numReceived, 3);
+        assertEquals(source2.numReceived, 2);
+
+        assertEquals(task.process(), 0);
+        assertEquals(source1.numReceived, 3);
+        assertEquals(source2.numReceived, 3);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void testPauseResume() {
+        byte[] recordValue = intSerializer.serialize(null, 10);
+
+        task.addRecords(partition1, records(
+            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 1, intSerializer.serialize(partition1.topic(), 10), recordValue),
+            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 2, intSerializer.serialize(partition1.topic(), 20), recordValue)
+        ));
+
+        task.addRecords(partition2, records(
+            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 3, intSerializer.serialize(partition1.topic(), 35), recordValue),
+            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 4, intSerializer.serialize(partition1.topic(), 45), recordValue),
+            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 5, intSerializer.serialize(partition1.topic(), 55), recordValue),
+            new ConsumerRecord<>(partition2.topic(), partition2.partition(), 6, intSerializer.serialize(partition1.topic(), 65), recordValue)
+        ));
+
+        assertEquals(task.process(), 5);
+        assertEquals(source1.numReceived, 1);
+        assertEquals(source2.numReceived, 0);
+
+        assertEquals(consumer.paused().size(), 1);
+        assertTrue(consumer.paused().contains(partition2));
+
+        task.addRecords(partition1, records(
+            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 3, intSerializer.serialize(partition1.topic(), 30), recordValue),
+            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 4, intSerializer.serialize(partition1.topic(), 40), recordValue),
+            new ConsumerRecord<>(partition1.topic(), partition1.partition(), 5, intSerializer.serialize(partition1.topic(), 50), recordValue)
+        ));
+
+        assertEquals(consumer.paused().size(), 2);
+        assertTrue(consumer.paused().contains(partition1));
+        assertTrue(consumer.paused().contains(partition2));
+
+        assertEquals(task.process(), 7);
+        assertEquals(source1.numReceived, 1);
+        assertEquals(source2.numReceived, 1);
+
+        assertEquals(consumer.paused().size(), 1);
+        assertTrue(consumer.paused().contains(partition1));
+
+        assertEquals(task.process(), 6);
+        assertEquals(source1.numReceived, 2);
+        assertEquals(source2.numReceived, 1);
+
+        assertEquals(consumer.paused().size(), 0);
+    }
+
+    private Iterator<ConsumerRecord<byte[], byte[]>> records(ConsumerRecord<byte[], byte[]>... recs) {
+        return Arrays.asList(recs).iterator();
+    }
+}
diff --git a/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java b/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java
index b8f65252bea39..5854e4b817a30 100644
--- a/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java
+++ b/stream/src/test/java/org/apache/kafka/test/MockProcessorContext.java
@@ -50,15 +50,13 @@ public void setTime(long timestamp) {
         this.timestamp = timestamp;
     }
 
-    @Override
-    public boolean joinable() {
-        // TODO
-        return true;
+    public int id() {
+        return -1;
     }
 
     @Override
-    public int id() {
-        return -1;
+    public boolean joinable() {
+        return true;
     }
 
     @Override
diff --git a/stream/src/test/java/org/apache/kafka/test/MockSourceNode.java b/stream/src/test/java/org/apache/kafka/test/MockSourceNode.java
index cd29a317c4ff9..9b6e4b01b6128 100644
--- a/stream/src/test/java/org/apache/kafka/test/MockSourceNode.java
+++ b/stream/src/test/java/org/apache/kafka/test/MockSourceNode.java
@@ -17,25 +17,25 @@
 
 package org.apache.kafka.test;
 
-import org.apache.kafka.streaming.processor.internals.SourceNode;
+
 import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.streaming.processor.internals.SourceNode;
 
 import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class MockSourceNode<K, V> extends SourceNode<K, V> {
 
-    public Deserializer<? extends K> keyDeserializer;
-    public Deserializer<? extends V> valDeserializer;
+    public static final String NAME = "MOCK-SOURCE-";
+
+    public static final AtomicInteger INDEX = new AtomicInteger(1);
 
     public int numReceived = 0;
     public ArrayList<K> keys = new ArrayList<>();
     public ArrayList<V> values = new ArrayList<>();
 
-    public MockSourceNode(Deserializer<? extends K> keyDeserializer, Deserializer<? extends V> valDeserializer) {
-        super(keyDeserializer, valDeserializer);
-
-        this.keyDeserializer = keyDeserializer;
-        this.valDeserializer = valDeserializer;
+    public MockSourceNode(Deserializer<K> keyDeserializer, Deserializer<V> valDeserializer) {
+        super(NAME + INDEX.getAndIncrement(), keyDeserializer, valDeserializer);
     }
 
     @Override
diff --git a/stream/src/test/java/org/apache/kafka/test/MockTimestampExtractor.java b/stream/src/test/java/org/apache/kafka/test/MockTimestampExtractor.java
new file mode 100644
index 0000000000000..331f01533edbd
--- /dev/null
+++ b/stream/src/test/java/org/apache/kafka/test/MockTimestampExtractor.java
@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.test;
+
+import org.apache.kafka.streaming.processor.TimestampExtractor;
+
+public class MockTimestampExtractor implements TimestampExtractor {
+
+    @Override
+    public long extract(String topic, Object key, Object value) {
+        return ((Integer) key).longValue();
+    }
+}
diff --git a/stream/src/test/java/org/apache/kafka/test/UnlimitedWindowDef.java b/stream/src/test/java/org/apache/kafka/test/UnlimitedWindowDef.java
index c6d929725c65f..b20a4b7ae2ede 100644
--- a/stream/src/test/java/org/apache/kafka/test/UnlimitedWindowDef.java
+++ b/stream/src/test/java/org/apache/kafka/test/UnlimitedWindowDef.java
@@ -48,26 +48,26 @@ public class UnlimitedWindow implements Window<K, V> {
         private LinkedList<Stamped<KeyValue<K, V>>> list = new LinkedList<>();
 
         @Override
-        public void init (ProcessorContext context){
+        public void init(ProcessorContext context) {
             context.register(this, null);
         }
 
         @Override
-        public Iterator<V> find(final K key, long timestamp){
+        public Iterator<V> find(final K key, long timestamp) {
             return find(key, Long.MIN_VALUE, timestamp);
         }
 
         @Override
-        public Iterator<V> findAfter(final K key, long timestamp){
+        public Iterator<V> findAfter(final K key, long timestamp) {
             return find(key, timestamp, Long.MAX_VALUE);
         }
 
         @Override
-        public Iterator<V> findBefore(final K key, long timestamp){
+        public Iterator<V> findBefore(final K key, long timestamp) {
             return find(key, Long.MIN_VALUE, Long.MAX_VALUE);
         }
 
-        private Iterator<V> find(final K key, final long startTime, final long endTime){
+        private Iterator<V> find(final K key, final long startTime, final long endTime) {
             return new FilteredIterator<V, Stamped<KeyValue<K, V>>>(list.iterator()) {
                 protected V filter(Stamped<KeyValue<K, V>> item) {
                     if (item.value.key.equals(key) && startTime <= item.timestamp && item.timestamp <= endTime)
@@ -79,25 +79,25 @@ protected V filter(Stamped<KeyValue<K, V>> item) {
         }
 
         @Override
-        public void put (K key, V value,long timestamp){
+        public void put(K key, V value, long timestamp) {
             list.add(new Stamped<>(KeyValue.pair(key, value), timestamp));
         }
 
         @Override
-        public String name () {
+        public String name() {
             return null;
         }
 
         @Override
-        public void flush () {
+        public void flush() {
         }
 
         @Override
-        public void close () {
+        public void close() {
         }
 
         @Override
-        public boolean persistent () {
+        public boolean persistent() {
             return false;
         }
     }