kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: KAFKA-2649: Add support for custom partitioning in topology sinks
Date Thu, 07 Jan 2016 22:46:08 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk ee1770e00 -> 4836e525c


KAFKA-2649: Add support for custom partitioning in topology sinks

Added option to use custom partitioning logic within each topology sink.

Author: Randall Hauch <rhauch@gmail.com>

Reviewers: Guozhang Wang

Closes #309 from rhauch/kafka-2649


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/4836e525
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/4836e525
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/4836e525

Branch: refs/heads/trunk
Commit: 4836e525c851640e0da9d8d11321621a2c70e8f0
Parents: ee1770e
Author: Randall Hauch <rhauch@gmail.com>
Authored: Thu Jan 7 14:46:02 2016 -0800
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Thu Jan 7 14:46:02 2016 -0800

----------------------------------------------------------------------
 .../streams/processor/StreamPartitioner.java    | 59 ++++++++++++++++
 .../streams/processor/TopologyBuilder.java      | 73 +++++++++++++++++++-
 .../processor/internals/RecordCollector.java    | 17 ++++-
 .../streams/processor/internals/SinkNode.java   |  7 +-
 .../internals/ProcessorTopologyTest.java        | 45 ++++++++----
 .../streams/state/KeyValueStoreTestDriver.java  |  6 ++
 .../apache/kafka/test/KStreamTestDriver.java    | 11 +++
 .../kafka/test/ProcessorTopologyTestDriver.java |  7 +-
 8 files changed, 203 insertions(+), 22 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/4836e525/streams/src/main/java/org/apache/kafka/streams/processor/StreamPartitioner.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/StreamPartitioner.java
b/streams/src/main/java/org/apache/kafka/streams/processor/StreamPartitioner.java
new file mode 100644
index 0000000..f14d9d9
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/StreamPartitioner.java
@@ -0,0 +1,59 @@
+/**
+ * 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.streams.processor;
+
+/**
+ * Determine how messages are distributed among the partitions in a Kafka topic. If not specified,
the underlying producer's
+ * {@link org.apache.kafka.clients.producer.internals.DefaultPartitioner} will be used to
determine the partition.
+ * <p>
+ * Kafka topics are divided into one or more <i>partitions</i>. Since each partition
must fit on the servers that host it, so
+ * using multiple partitions allows the topic to scale beyond a size that will fit on a single
machine. Partitions also enable you
+ * to use multiple instances of your topology to process in parallel all of the messages
on the topology's source topics.
+ * <p>
+ * When a topology is instantiated, each of its sources are assigned a subset of that topic's
partitions. That means that only
+ * those processors in that topology instance will consume the messages from those partitions.
In many cases, Kafka Streams will
+ * automatically manage these instances, and adjust when new topology instances are added
or removed.
+ * <p>
+ * Some topologies, though, need more control over which messages appear in each partition.
For example, some topologies that have
+ * stateful processors may want all messages within a range of keys to always be delivered
to and handled by the same topology instance.
+ * An upstream topology producing messages to that topic can use a custom <i>stream
partitioner</i> to precisely and consistently
+ * determine to which partition each message should be written.
+ * <p>
+ * To do this, create a <code>StreamPartitioner</code> implementation, and when
you build your topology specify that custom partitioner
+ * when {@link TopologyBuilder#addSink(String, String, org.apache.kafka.common.serialization.Serializer,
org.apache.kafka.common.serialization.Serializer, StreamPartitioner, String...) adding a sink}
+ * for that topic.
+ * <p>
+ * All StreamPartitioner implementations should be stateless and a pure function so they
can be shared across topic and sink nodes.
+ * 
+ * @param <K> the type of keys
+ * @param <V> the type of values
+ * @see TopologyBuilder#addSink(String, String, org.apache.kafka.common.serialization.Serializer,
+ *      org.apache.kafka.common.serialization.Serializer, StreamPartitioner, String...)
+ * @see TopologyBuilder#addSink(String, String, StreamPartitioner, String...)
+ */
+public interface StreamPartitioner<K, V> {
+
+    /**
+     * Determine the partition number for a message with the given key and value and the
current number of partitions.
+     * 
+     * @param key the key of the message
+     * @param value the value of the message
+     * @param numPartitions the total number of partitions
+     * @return an integer between 0 and {@code numPartitions-1}, or {@code null} if the default
partitioning logic should be used
+     */
+    Integer partition(K key, V value, int numPartitions);
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/4836e525/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
b/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
index 3cfb22b..9cd80a4 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
@@ -132,19 +132,21 @@ public class TopologyBuilder {
         public final String topic;
         private Serializer keySerializer;
         private Serializer valSerializer;
+        private final StreamPartitioner partitioner;
 
-        private SinkNodeFactory(String name, String[] parents, String topic, Serializer keySerializer,
Serializer valSerializer) {
+        private SinkNodeFactory(String name, String[] parents, String topic, Serializer keySerializer,
Serializer valSerializer, StreamPartitioner partitioner) {
             super(name);
             this.parents = parents.clone();
             this.topic = topic;
             this.keySerializer = keySerializer;
             this.valSerializer = valSerializer;
+            this.partitioner = partitioner;
         }
 
         @SuppressWarnings("unchecked")
         @Override
         public ProcessorNode build() {
-            return new SinkNode(name, topic, keySerializer, valSerializer);
+            return new SinkNode(name, topic, keySerializer, valSerializer, partitioner);
         }
     }
 
@@ -235,15 +237,53 @@ public class TopologyBuilder {
      *
      * @param name the unique name of the sink
      * @param topic the name of the Kafka topic to which this sink should write its messages
+     * @param parentNames the name of one or more source or processor nodes whose output
message this sink should consume
+     * and write to its topic
      * @return this builder instance so methods can be chained together; never null
+     * @see #addSink(String, String, StreamPartitioner, String...)
+     * @see #addSink(String, String, Serializer, Serializer, String...)
+     * @see #addSink(String, String, Serializer, Serializer, StreamPartitioner, String...)
      */
     public final TopologyBuilder addSink(String name, String topic, String... parentNames)
{
         return addSink(name, topic, (Serializer) null, (Serializer) null, parentNames);
     }
 
     /**
+     * Add a new sink that forwards messages from upstream parent processor and/or source
nodes to the named Kafka topic, using
+     * the supplied partitioner.
+     * The sink will use the {@link StreamingConfig#KEY_SERIALIZER_CLASS_CONFIG default key
serializer} and
+     * {@link StreamingConfig#VALUE_SERIALIZER_CLASS_CONFIG default value serializer} specified
in the
+     * {@link StreamingConfig streaming configuration}.
+     * <p>
+     * The sink will also use the specified {@link StreamPartitioner} to determine how messages
are distributed among
+     * the named Kafka topic's partitions. Such control is often useful with topologies that
use
+     * {@link #addStateStore(StateStoreSupplier, String...) state stores}
+     * in its processors. In most other cases, however, a partitioner need not be specified
and Kafka will automatically distribute
+     * messages among partitions using Kafka's default partitioning logic.
+     *
+     * @param name the unique name of the sink
+     * @param topic the name of the Kafka topic to which this sink should write its messages
+     * @param partitioner the function that should be used to determine the partition for
each message processed by the sink
+     * @param parentNames the name of one or more source or processor nodes whose output
message this sink should consume
+     * and write to its topic
+     * @return this builder instance so methods can be chained together; never null
+     * @see #addSink(String, String, String...)
+     * @see #addSink(String, String, Serializer, Serializer, String...)
+     * @see #addSink(String, String, Serializer, Serializer, StreamPartitioner, String...)
+     */
+    public final TopologyBuilder addSink(String name, String topic, StreamPartitioner partitioner,
String... parentNames) {
+        return addSink(name, topic, (Serializer) null, (Serializer) null, partitioner, parentNames);
+    }
+
+    /**
      * Add a new sink that forwards messages from upstream parent processor and/or source
nodes to the named Kafka topic.
      * The sink will use the specified key and value serializers.
+     * <p>
+     * The sink will also use the specified {@link StreamPartitioner} to determine how messages
are distributed among
+     * the named Kafka topic's partitions. Such control is often useful with topologies that
use
+     * {@link #addStateStore(StateStoreSupplier, String...) state stores}
+     * in its processors. In most other cases, however, a partitioner need not be specified
and Kafka will automatically distribute
+     * messages among partitions using Kafka's default partitioning logic.
      *
      * @param name the unique name of the sink
      * @param topic the name of the Kafka topic to which this sink should write its messages
@@ -256,8 +296,35 @@ public class TopologyBuilder {
      * @param parentNames the name of one or more source or processor nodes whose output
message this sink should consume
      * and write to its topic
      * @return this builder instance so methods can be chained together; never null
+     * @see #addSink(String, String, String...)
+     * @see #addSink(String, String, StreamPartitioner, String...)
+     * @see #addSink(String, String, Serializer, Serializer, StreamPartitioner, String...)
      */
     public final TopologyBuilder addSink(String name, String topic, Serializer keySerializer,
Serializer valSerializer, String... parentNames) {
+        return addSink(name, topic, keySerializer, valSerializer, (StreamPartitioner) null,
parentNames);
+    }
+
+    /**
+     * Add a new sink that forwards messages from upstream parent processor and/or source
nodes to the named Kafka topic.
+     * The sink will use the specified key and value serializers, and the supplied partitioner.
+     *
+     * @param name the unique name of the sink
+     * @param topic the name of the Kafka topic to which this sink should write its messages
+     * @param keySerializer the {@link Serializer key serializer} used when consuming messages;
may be null if the sink
+     * should use the {@link StreamingConfig#KEY_SERIALIZER_CLASS_CONFIG default key serializer}
specified in the
+     * {@link StreamingConfig streaming configuration}
+     * @param valSerializer the {@link Serializer value serializer} used when consuming messages;
may be null if the sink
+     * should use the {@link StreamingConfig#VALUE_SERIALIZER_CLASS_CONFIG default value
serializer} specified in the
+     * {@link StreamingConfig streaming configuration}
+     * @param partitioner the function that should be used to determine the partition for
each message processed by the sink
+     * @param parentNames the name of one or more source or processor nodes whose output
message this sink should consume
+     * and write to its topic
+     * @return this builder instance so methods can be chained together; never null
+     * @see #addSink(String, String, String...)
+     * @see #addSink(String, String, StreamPartitioner, String...)
+     * @see #addSink(String, String, Serializer, Serializer, String...)
+     */
+    public final <K, V> TopologyBuilder addSink(String name, String topic, Serializer<K>
keySerializer, Serializer<V> valSerializer, StreamPartitioner<K, V> partitioner,
String... parentNames) {
         if (nodeFactories.containsKey(name))
             throw new TopologyException("Processor " + name + " is already added.");
 
@@ -272,7 +339,7 @@ public class TopologyBuilder {
             }
         }
 
-        nodeFactories.put(name, new SinkNodeFactory(name, parentNames, topic, keySerializer,
valSerializer));
+        nodeFactories.put(name, new SinkNodeFactory(name, parentNames, topic, keySerializer,
valSerializer, partitioner));
         nodeGrouper.add(name);
         nodeGrouper.unite(name, parentNames);
         return this;

http://git-wip-us.apache.org/repos/asf/kafka/blob/4836e525/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java
index f0dbf35..fe0472e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollector.java
@@ -21,12 +21,15 @@ import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.streams.processor.StreamPartitioner;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 public class RecordCollector {
@@ -47,6 +50,7 @@ public class RecordCollector {
     private final Producer<byte[], byte[]> producer;
     private final Map<TopicPartition, Long> offsets;
     private final Callback callback = new Callback() {
+        @Override
         public void onCompletion(RecordMetadata metadata, Exception exception) {
             if (exception == null) {
                 TopicPartition tp = new TopicPartition(metadata.topic(), metadata.partition());
@@ -64,9 +68,20 @@ public class RecordCollector {
     }
 
     public <K, V> void send(ProducerRecord<K, V> record, Serializer<K>
keySerializer, Serializer<V> valueSerializer) {
+        send(record, keySerializer, valueSerializer, null);
+    }
+
+    public <K, V> void send(ProducerRecord<K, V> record, Serializer<K>
keySerializer, Serializer<V> valueSerializer,
+                            StreamPartitioner<K, V> partitioner) {
         byte[] keyBytes = keySerializer.serialize(record.topic(), record.key());
         byte[] valBytes = valueSerializer.serialize(record.topic(), record.value());
-        this.producer.send(new ProducerRecord<>(record.topic(), keyBytes, valBytes),
callback);
+        Integer partition = null;
+        if (partitioner != null) {
+            List<PartitionInfo> partitions = this.producer.partitionsFor(record.topic());
+            if (partitions != null)
+                partition = partitioner.partition(record.key(), record.value(), partitions.size());
+        }
+        this.producer.send(new ProducerRecord<>(record.topic(), partition, keyBytes,
valBytes), callback);
     }
 
     public void flush() {

http://git-wip-us.apache.org/repos/asf/kafka/blob/4836e525/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
index 9f01727..7ab59ee 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java
@@ -20,21 +20,24 @@ package org.apache.kafka.streams.processor.internals;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.StreamPartitioner;
 
 public class SinkNode<K, V> extends ProcessorNode<K, V> {
 
     private final String topic;
     private Serializer<K> keySerializer;
     private Serializer<V> valSerializer;
+    private final StreamPartitioner<K, V> partitioner;
 
     private ProcessorContext context;
 
-    public SinkNode(String name, String topic, Serializer<K> keySerializer, Serializer<V>
valSerializer) {
+    public SinkNode(String name, String topic, Serializer<K> keySerializer, Serializer<V>
valSerializer, StreamPartitioner<K, V> partitioner) {
         super(name);
 
         this.topic = topic;
         this.keySerializer = keySerializer;
         this.valSerializer = valSerializer;
+        this.partitioner = partitioner;
     }
 
     @Override
@@ -54,7 +57,7 @@ public class SinkNode<K, V> extends ProcessorNode<K, V> {
     public void process(K key, V value) {
         // send to all the registered topics
         RecordCollector collector = ((RecordCollector.Supplier) context).recordCollector();
-        collector.send(new ProducerRecord<>(topic, key, value), keySerializer, valSerializer);
+        collector.send(new ProducerRecord<>(topic, key, value), keySerializer, valSerializer,
partitioner);
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/kafka/blob/4836e525/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
index 2f359bc..f2ef2ea 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
@@ -32,6 +32,7 @@ import org.apache.kafka.streams.processor.AbstractProcessor;
 import org.apache.kafka.streams.processor.Processor;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.ProcessorSupplier;
+import org.apache.kafka.streams.processor.StreamPartitioner;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 import org.apache.kafka.streams.processor.TopologyBuilder;
 import org.apache.kafka.streams.state.KeyValueIterator;
@@ -114,22 +115,23 @@ public class ProcessorTopologyTest {
 
     @Test
     public void testDrivingSimpleTopology() {
-        driver = new ProcessorTopologyTestDriver(config, createSimpleTopology());
+        int partition = 10;
+        driver = new ProcessorTopologyTestDriver(config, createSimpleTopology(partition));
         driver.process(INPUT_TOPIC, "key1", "value1", STRING_SERIALIZER, STRING_SERIALIZER);
-        assertNextOutputRecord(OUTPUT_TOPIC_1, "key1", "value1");
+        assertNextOutputRecord(OUTPUT_TOPIC_1, "key1", "value1", partition);
         assertNoOutputRecord(OUTPUT_TOPIC_2);
 
         driver.process(INPUT_TOPIC, "key2", "value2", STRING_SERIALIZER, STRING_SERIALIZER);
-        assertNextOutputRecord(OUTPUT_TOPIC_1, "key2", "value2");
+        assertNextOutputRecord(OUTPUT_TOPIC_1, "key2", "value2", partition);
         assertNoOutputRecord(OUTPUT_TOPIC_2);
 
         driver.process(INPUT_TOPIC, "key3", "value3", STRING_SERIALIZER, STRING_SERIALIZER);
         driver.process(INPUT_TOPIC, "key4", "value4", STRING_SERIALIZER, STRING_SERIALIZER);
         driver.process(INPUT_TOPIC, "key5", "value5", STRING_SERIALIZER, STRING_SERIALIZER);
         assertNoOutputRecord(OUTPUT_TOPIC_2);
-        assertNextOutputRecord(OUTPUT_TOPIC_1, "key3", "value3");
-        assertNextOutputRecord(OUTPUT_TOPIC_1, "key4", "value4");
-        assertNextOutputRecord(OUTPUT_TOPIC_1, "key5", "value5");
+        assertNextOutputRecord(OUTPUT_TOPIC_1, "key3", "value3", partition);
+        assertNextOutputRecord(OUTPUT_TOPIC_1, "key4", "value4", partition);
+        assertNextOutputRecord(OUTPUT_TOPIC_1, "key5", "value5", partition);
     }
 
     @Test
@@ -172,25 +174,38 @@ public class ProcessorTopologyTest {
     }
 
     protected void assertNextOutputRecord(String topic, String key, String value) {
-        assertProducerRecord(driver.readOutput(topic, STRING_DESERIALIZER, STRING_DESERIALIZER),
topic, key, value);
+        ProducerRecord<String, String> record = driver.readOutput(topic, STRING_DESERIALIZER,
STRING_DESERIALIZER);
+        assertEquals(topic, record.topic());
+        assertEquals(key, record.key());
+        assertEquals(value, record.value());
+        assertNull(record.partition());
+    }
+
+    protected void assertNextOutputRecord(String topic, String key, String value, Integer
partition) {
+        ProducerRecord<String, String> record = driver.readOutput(topic, STRING_DESERIALIZER,
STRING_DESERIALIZER);
+        assertEquals(topic, record.topic());
+        assertEquals(key, record.key());
+        assertEquals(value, record.value());
+        assertEquals(partition, record.partition());
     }
 
     protected void assertNoOutputRecord(String topic) {
         assertNull(driver.readOutput(topic));
     }
 
-    private void assertProducerRecord(ProducerRecord<String, String> record, String
topic, String key, String value) {
-        assertEquals(topic, record.topic());
-        assertEquals(key, record.key());
-        assertEquals(value, record.value());
-        // Kafka Streaming doesn't set the partition, so it's always null
-        assertNull(record.partition());
+    protected <K, V> StreamPartitioner<K, V> constantPartitioner(final Integer
partition) {
+        return new StreamPartitioner<K, V>() {
+            @Override
+            public Integer partition(K key, V value, int numPartitions) {
+                return partition;
+            }
+        };
     }
 
-    protected TopologyBuilder createSimpleTopology() {
+    protected TopologyBuilder createSimpleTopology(int partition) {
         return new TopologyBuilder().addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER,
INPUT_TOPIC)
                                     .addProcessor("processor", define(new ForwardingProcessor()),
"source")
-                                    .addSink("sink", OUTPUT_TOPIC_1, "processor");
+                                    .addSink("sink", OUTPUT_TOPIC_1, constantPartitioner(partition),
"processor");
     }
 
     protected TopologyBuilder createMultiplexingTopology() {

http://git-wip-us.apache.org/repos/asf/kafka/blob/4836e525/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
index 6f74da8..108797a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
@@ -28,6 +28,7 @@ import org.apache.kafka.streams.StreamingMetrics;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.StateRestoreCallback;
 import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.StreamPartitioner;
 import org.apache.kafka.streams.processor.TaskId;
 import org.apache.kafka.streams.processor.internals.RecordCollector;
 import org.apache.kafka.test.MockProcessorContext;
@@ -245,6 +246,11 @@ public class KeyValueStoreTestDriver<K, V> {
             public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1>
keySerializer, Serializer<V1> valueSerializer) {
                 recordFlushed(record.key(), record.value());
             }
+            @Override
+            public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1>
keySerializer, Serializer<V1> valueSerializer,
+                                    StreamPartitioner<K1, V1> partitioner) {
+                recordFlushed(record.key(), record.value());
+            }
         };
         this.stateDir = StateUtils.tempDir();
         this.stateDir.mkdirs();

http://git-wip-us.apache.org/repos/asf/kafka/blob/4836e525/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java b/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
index 5275545..2dc567e 100644
--- a/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java
@@ -24,6 +24,7 @@ import org.apache.kafka.streams.kstream.KStreamBuilder;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.processor.StateStoreSupplier;
+import org.apache.kafka.streams.processor.StreamPartitioner;
 import org.apache.kafka.streams.processor.internals.ProcessorNode;
 import org.apache.kafka.streams.processor.internals.ProcessorTopology;
 import org.apache.kafka.streams.processor.internals.RecordCollector;
@@ -126,15 +127,25 @@ public class KStreamTestDriver {
         public MockRecordCollector() {
             super(null);
         }
+        
+        @Override
+        public <K, V> void send(ProducerRecord<K, V> record, Serializer<K>
keySerializer, Serializer<V> valueSerializer,
+                                StreamPartitioner<K, V> partitioner) {
+            // The serialization is skipped.
+            process(record.topic(), record.key(), record.value());
+        }
 
+        @Override
         public <K, V> void send(ProducerRecord<K, V> record, Serializer<K>
keySerializer, Serializer<V> valueSerializer) {
             // The serialization is skipped.
             process(record.topic(), record.key(), record.value());
         }
 
+        @Override
         public void flush() {
         }
 
+        @Override
         public void close() {
         }
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/4836e525/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
b/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
index 879c172..eaeed09 100644
--- a/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
@@ -149,7 +149,12 @@ public class ProcessorTopologyTestDriver {
 
         // Set up the consumer and producer ...
         consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
-        producer = new MockProducer<>(true, bytesSerializer, bytesSerializer);
+        producer = new MockProducer<byte[], byte[]>(true, bytesSerializer, bytesSerializer)
{
+            @Override
+            public List<PartitionInfo> partitionsFor(String topic) {
+                return Collections.emptyList();
+            }
+        };
         restoreStateConsumer = createRestoreConsumer(id, storeNames);
 
         // Set up all of the topic+partition information and subscribe the consumer to each
...


Mime
View raw message