Processor unit tests

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"/>
 

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.");

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;
     }

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();
+    }
+}