Repository: kafka Updated Branches: refs/heads/trunk 63cf0e587 -> abe19fe69
KAFKA-3835; Streams is creating two ProducerRecords for each send via RecordCollector Author: Jeyhun Karimov <[email protected]> Reviewers: Matthias J. Sax, Guozhang Wang Closes #2417 from jeyhunkarimov/KAFKA-3835 Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/abe19fe6 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/abe19fe6 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/abe19fe6 Branch: refs/heads/trunk Commit: abe19fe693d81ea65a018041f157c58ebecd014d Parents: 63cf0e5 Author: Jeyhun Karimov <[email protected]> Authored: Tue Jan 24 16:36:31 2017 -0800 Committer: Guozhang Wang <[email protected]> Committed: Tue Jan 24 16:36:31 2017 -0800 ---------------------------------------------------------------------- .../processor/internals/RecordCollector.java | 19 ++- .../internals/RecordCollectorImpl.java | 34 +++-- .../streams/processor/internals/SinkNode.java | 3 +- .../processor/internals/StandbyContextImpl.java | 19 ++- .../state/internals/StoreChangeLogger.java | 3 +- .../internals/RecordCollectorTest.java | 48 +++---- .../streams/state/KeyValueStoreTestDriver.java | 33 +++-- .../ChangeLoggingKeyValueBytesStoreTest.java | 11 +- .../ChangeLoggingKeyValueStoreTest.java | 11 +- .../ChangeLoggingSegmentedBytesStoreTest.java | 11 +- .../RocksDBKeyValueStoreSupplierTest.java | 20 ++- .../RocksDBSessionStoreSupplierTest.java | 20 ++- .../RocksDBWindowStoreSupplierTest.java | 20 ++- .../state/internals/RocksDBWindowStoreTest.java | 133 ++++++++++++++----- .../state/internals/StoreChangeLoggerTest.java | 23 +++- .../apache/kafka/test/KStreamTestDriver.java | 23 +++- .../apache/kafka/test/NoOpRecordCollector.java | 20 ++- 17 files changed, 324 insertions(+), 127 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/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 ec20d46..730b37d 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 @@ -16,7 +16,6 @@ */ package org.apache.kafka.streams.processor.internals; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streams.processor.StreamPartitioner; @@ -24,9 +23,21 @@ import org.apache.kafka.streams.processor.StreamPartitioner; import java.util.Map; public interface RecordCollector { - <K, V> void send(ProducerRecord<K, V> record, Serializer<K> keySerializer, Serializer<V> valueSerializer); - - <K, V> void send(ProducerRecord<K, V> record, Serializer<K> keySerializer, Serializer<V> valueSerializer, + <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer); + + <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer, StreamPartitioner<? super K, ? super V> partitioner); void flush(); http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java ---------------------------------------------------------------------- diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java index afb8999..f243d82 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java @@ -40,7 +40,7 @@ public class RecordCollectorImpl implements RecordCollector { private static final long SEND_RETRY_BACKOFF = 100L; private static final Logger log = LoggerFactory.getLogger(RecordCollectorImpl.class); - + private final Producer<byte[], byte[]> producer; private final Map<TopicPartition, Long> offsets; private final String logPrefix; @@ -54,26 +54,36 @@ public class RecordCollectorImpl implements RecordCollector { } @Override - 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(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + send(topic, key, value, partition, timestamp, keySerializer, valueSerializer, null); } @Override - public <K, V> void send(ProducerRecord<K, V> record, Serializer<K> keySerializer, Serializer<V> valueSerializer, - StreamPartitioner<? super K, ? super V> partitioner) { + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer, + StreamPartitioner<? super K, ? super V> partitioner) { checkForException(); - byte[] keyBytes = keySerializer.serialize(record.topic(), record.key()); - byte[] valBytes = valueSerializer.serialize(record.topic(), record.value()); - Integer partition = record.partition(); + byte[] keyBytes = keySerializer.serialize(topic, key); + byte[] valBytes = valueSerializer.serialize(topic, value); if (partition == null && partitioner != null) { - List<PartitionInfo> partitions = this.producer.partitionsFor(record.topic()); + List<PartitionInfo> partitions = this.producer.partitionsFor(topic); if (partitions != null && partitions.size() > 0) - partition = partitioner.partition(record.key(), record.value(), partitions.size()); + partition = partitioner.partition(key, value, partitions.size()); } ProducerRecord<byte[], byte[]> serializedRecord = - new ProducerRecord<>(record.topic(), partition, record.timestamp(), keyBytes, valBytes); - final String topic = serializedRecord.topic(); + new ProducerRecord<>(topic, partition, timestamp, keyBytes, valBytes); for (int attempt = 1; attempt <= MAX_SEND_ATTEMPTS; attempt++) { try { http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/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 c4db740..f144079 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 @@ -17,7 +17,6 @@ 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.errors.StreamsException; import org.apache.kafka.streams.kstream.internals.ChangedSerializer; @@ -77,7 +76,7 @@ public class SinkNode<K, V> extends ProcessorNode<K, V> { } try { - collector.send(new ProducerRecord<>(topic, null, timestamp, key, value), keySerializer, valSerializer, partitioner); + collector.send(topic, key, value, null, timestamp, keySerializer, valSerializer, partitioner); } catch (ClassCastException e) { throw new StreamsException( String.format("A serializer (key: %s / value: %s) is not compatible to the actual key or value type " + http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java ---------------------------------------------------------------------- diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java index 954c710..82f2714 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyContextImpl.java @@ -17,7 +17,6 @@ package org.apache.kafka.streams.processor.internals; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streams.StreamsConfig; @@ -33,12 +32,24 @@ class StandbyContextImpl extends AbstractProcessorContext implements RecordColle private static final RecordCollector NO_OP_COLLECTOR = new RecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { } @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final StreamPartitioner<? super K, ? super V> partitioner) { + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer, + StreamPartitioner<? super K, ? super V> partitioner) { } http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreChangeLogger.java ---------------------------------------------------------------------- diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreChangeLogger.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreChangeLogger.java index 44cda5b..39c8270 100644 --- a/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreChangeLogger.java +++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreChangeLogger.java @@ -17,7 +17,6 @@ package org.apache.kafka.streams.state.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.internals.ProcessorStateManager; @@ -58,7 +57,7 @@ class StoreChangeLogger<K, V> { if (collector != null) { final Serializer<K> keySerializer = serialization.keySerializer(); final Serializer<V> valueSerializer = serialization.valueSerializer(); - collector.send(new ProducerRecord<>(this.topic, this.partition, context.timestamp(), key, value), keySerializer, valueSerializer); + collector.send(this.topic, key, value, this.partition, context.timestamp(), keySerializer, valueSerializer); } } http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java index b54ce34..f0d06f7 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java @@ -74,14 +74,14 @@ public class RecordCollectorTest { new MockProducer<>(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer), "RecordCollectorTest-TestSpecificPartition"); - collector.send(new ProducerRecord<>("topic1", 0, "999", "0"), stringSerializer, stringSerializer); - collector.send(new ProducerRecord<>("topic1", 0, "999", "0"), stringSerializer, stringSerializer); - collector.send(new ProducerRecord<>("topic1", 0, "999", "0"), stringSerializer, stringSerializer); + collector.send("topic1", "999", "0", 0, null, stringSerializer, stringSerializer); + collector.send("topic1", "999", "0", 0, null, stringSerializer, stringSerializer); + collector.send("topic1", "999", "0", 0, null, stringSerializer, stringSerializer); - collector.send(new ProducerRecord<>("topic1", 1, "999", "0"), stringSerializer, stringSerializer); - collector.send(new ProducerRecord<>("topic1", 1, "999", "0"), stringSerializer, stringSerializer); + collector.send("topic1", "999", "0", 1, null, stringSerializer, stringSerializer); + collector.send("topic1", "999", "0", 1, null, stringSerializer, stringSerializer); - collector.send(new ProducerRecord<>("topic1", 2, "999", "0"), stringSerializer, stringSerializer); + collector.send("topic1", "999", "0", 2, null, stringSerializer, stringSerializer); Map<TopicPartition, Long> offsets = collector.offsets(); @@ -90,9 +90,9 @@ public class RecordCollectorTest { assertEquals((Long) 0L, offsets.get(new TopicPartition("topic1", 2))); // ignore StreamPartitioner - collector.send(new ProducerRecord<>("topic1", 0, "999", "0"), stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", 1, "999", "0"), stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", 2, "999", "0"), stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "999", "0", 0, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "999", "0", 1, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "999", "0", 2, null, stringSerializer, stringSerializer, streamPartitioner); assertEquals((Long) 3L, offsets.get(new TopicPartition("topic1", 0))); assertEquals((Long) 2L, offsets.get(new TopicPartition("topic1", 1))); @@ -106,17 +106,17 @@ public class RecordCollectorTest { new MockProducer<>(cluster, true, new DefaultPartitioner(), byteArraySerializer, byteArraySerializer), "RecordCollectorTest-TestStreamPartitioner"); - collector.send(new ProducerRecord<>("topic1", "3", "0"), stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", "9", "0"), stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", "27", "0"), stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", "81", "0"), stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", "243", "0"), stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "9", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "27", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "81", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "243", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", "28", "0"), stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", "82", "0"), stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", "244", "0"), stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "28", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "82", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "244", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", "245", "0"), stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "245", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); Map<TopicPartition, Long> offsets = collector.offsets(); @@ -141,7 +141,7 @@ public class RecordCollectorTest { }, "test"); - collector.send(new ProducerRecord<>("topic1", "3", "0"), stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); final Long offset = collector.offsets().get(new TopicPartition("topic1", 0)); assertEquals(Long.valueOf(0L), offset); } @@ -158,7 +158,7 @@ public class RecordCollectorTest { }, "test"); - collector.send(new ProducerRecord<>("topic1", "3", "0"), stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); } @@ -174,8 +174,8 @@ public class RecordCollectorTest { } }, "test"); - collector.send(new ProducerRecord<>("topic1", "3", "0"), stringSerializer, stringSerializer, streamPartitioner); - collector.send(new ProducerRecord<>("topic1", "3", "0"), stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); } @SuppressWarnings("unchecked") @@ -190,7 +190,7 @@ public class RecordCollectorTest { } }, "test"); - collector.send(new ProducerRecord<>("topic1", "3", "0"), stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); collector.flush(); } @@ -206,7 +206,7 @@ public class RecordCollectorTest { } }, "test"); - collector.send(new ProducerRecord<>("topic1", "3", "0"), stringSerializer, stringSerializer, streamPartitioner); + collector.send("topic1", "3", "0", null, null, stringSerializer, stringSerializer, streamPartitioner); collector.close(); } http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/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 efa0e0e..2f5b368 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,7 +28,6 @@ import java.util.Properties; import java.util.Set; import org.apache.kafka.clients.producer.MockProducer; import org.apache.kafka.clients.producer.Producer; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.metrics.JmxReporter; import org.apache.kafka.common.metrics.MetricConfig; import org.apache.kafka.common.metrics.Metrics; @@ -203,20 +202,32 @@ public class KeyValueStoreTestDriver<K, V> { this.recordCollector = new RecordCollectorImpl(producer, "KeyValueStoreTestDriver") { @SuppressWarnings("unchecked") @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { - // for byte arrays we need to wrap it for comparison - - K key = serdes.keyFrom(keySerializer.serialize(record.topic(), record.key())); - V value = serdes.valueFrom(valueSerializer.serialize(record.topic(), record.value())); - - recordFlushed(key, value); + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { + // for byte arrays we need to wrap it for comparison + + K keyTest = serdes.keyFrom(keySerializer.serialize(topic, key)); + V valueTest = serdes.valueFrom(valueSerializer.serialize(topic, value)); + + recordFlushed(keyTest, valueTest); } @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer, - StreamPartitioner<? super K1, ? super V1> partitioner) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer, + StreamPartitioner<? super K1, ? super V1> partitioner) { // ignore partitioner - send(record, keySerializer, valueSerializer); + send(topic, key, value, partition, timestamp, keySerializer, valueSerializer); } }; this.stateDir = TestUtils.tempDirectory(); http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java index 99b1347..58b42e3 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java @@ -16,7 +16,6 @@ */ package org.apache.kafka.streams.state.internals; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.serialization.Serializer; @@ -53,8 +52,14 @@ public class ChangeLoggingKeyValueBytesStoreTest { public void before() { final NoOpRecordCollector collector = new NoOpRecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - sent.put(record.key(), record.value()); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + sent.put(key, value); } }; final MockProcessorContext context = new MockProcessorContext(TestUtils.tempDirectory(), http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStoreTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStoreTest.java index 2dc2017..d45f82e 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueStoreTest.java @@ -16,7 +16,6 @@ */ package org.apache.kafka.streams.state.internals; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; @@ -61,8 +60,14 @@ public class ChangeLoggingKeyValueStoreTest { public void before() { final NoOpRecordCollector collector = new NoOpRecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - sent.put(record.key(), record.value()); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + sent.put(key, value); } }; final MockProcessorContext context = new MockProcessorContext(TestUtils.tempDirectory(), http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java index 51f31bf..6ba7b14 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingSegmentedBytesStoreTest.java @@ -17,7 +17,6 @@ package org.apache.kafka.streams.state.internals; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.serialization.Serializer; @@ -48,8 +47,14 @@ public class ChangeLoggingSegmentedBytesStoreTest { public void setUp() throws Exception { final NoOpRecordCollector collector = new NoOpRecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - sent.put(record.key(), record.value()); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + sent.put(key, value); } }; final MockProcessorContext context = new MockProcessorContext(TestUtils.tempDirectory(), http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueStoreSupplierTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueStoreSupplierTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueStoreSupplierTest.java index c510089..304772b 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueStoreSupplierTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueStoreSupplierTest.java @@ -62,8 +62,14 @@ public class RocksDBKeyValueStoreSupplierTest { final List<ProducerRecord> logged = new ArrayList<>(); final NoOpRecordCollector collector = new NoOpRecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - logged.add(record); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + logged.add(new ProducerRecord<K, V>(topic, partition, timestamp, key, value)); } }; final MockProcessorContext context = new MockProcessorContext(TestUtils.tempDirectory(), @@ -83,8 +89,14 @@ public class RocksDBKeyValueStoreSupplierTest { final List<ProducerRecord> logged = new ArrayList<>(); final NoOpRecordCollector collector = new NoOpRecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - logged.add(record); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + logged.add(new ProducerRecord<K, V>(topic, partition, timestamp, key, value)); } }; final MockProcessorContext context = new MockProcessorContext(TestUtils.tempDirectory(), http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplierTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplierTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplierTest.java index 6677624..48618b7 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplierTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreSupplierTest.java @@ -65,8 +65,14 @@ public class RocksDBSessionStoreSupplierTest { final List<ProducerRecord> logged = new ArrayList<>(); final NoOpRecordCollector collector = new NoOpRecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - logged.add(record); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + logged.add(new ProducerRecord<K, V>(topic, partition, timestamp, key, value)); } }; final MockProcessorContext context = new MockProcessorContext(TestUtils.tempDirectory(), @@ -86,8 +92,14 @@ public class RocksDBSessionStoreSupplierTest { final List<ProducerRecord> logged = new ArrayList<>(); final NoOpRecordCollector collector = new NoOpRecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - logged.add(record); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + logged.add(new ProducerRecord<K, V>(topic, partition, timestamp, key, value)); } }; final MockProcessorContext context = new MockProcessorContext(TestUtils.tempDirectory(), http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplierTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplierTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplierTest.java index d9a0d4f..98fac08 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplierTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreSupplierTest.java @@ -62,8 +62,14 @@ public class RocksDBWindowStoreSupplierTest { final List<ProducerRecord> logged = new ArrayList<>(); final NoOpRecordCollector collector = new NoOpRecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - logged.add(record); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + logged.add(new ProducerRecord<K, V>(topic, partition, timestamp, key, value)); } }; final MockProcessorContext context = new MockProcessorContext(TestUtils.tempDirectory(), @@ -83,8 +89,14 @@ public class RocksDBWindowStoreSupplierTest { final List<ProducerRecord> logged = new ArrayList<>(); final NoOpRecordCollector collector = new NoOpRecordCollector() { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - logged.add(record); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + logged.add(new ProducerRecord<K, V>(topic, partition, timestamp, key, value)); } }; final MockProcessorContext context = new MockProcessorContext(TestUtils.tempDirectory(), http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java index ee846f7..d606e84 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java @@ -19,7 +19,6 @@ package org.apache.kafka.streams.state.internals; import org.apache.kafka.clients.producer.MockProducer; import org.apache.kafka.clients.producer.Producer; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; @@ -95,7 +94,13 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-ShouldOnlyIterateOpenSegments") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { } }; @@ -142,10 +147,16 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestPutAndFetch") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { changeLog.add(new KeyValue<>( - keySerializer.serialize(record.topic(), record.key()), - valueSerializer.serialize(record.topic(), record.value())) + keySerializer.serialize(topic, key), + valueSerializer.serialize(topic, value)) ); } }; @@ -216,10 +227,16 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestPutAndFetchBefore") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { changeLog.add(new KeyValue<>( - keySerializer.serialize(record.topic(), record.key()), - valueSerializer.serialize(record.topic(), record.value())) + keySerializer.serialize(topic, key), + valueSerializer.serialize(topic, value)) ); } }; @@ -305,10 +322,16 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestPutAndFetchAfter") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { changeLog.add(new KeyValue<>( - keySerializer.serialize(record.topic(), record.key()), - valueSerializer.serialize(record.topic(), record.value())) + keySerializer.serialize(topic, key), + valueSerializer.serialize(topic, value)) ); } }; @@ -392,10 +415,16 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestPutSameKeyTimestamp") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { changeLog.add(new KeyValue<>( - keySerializer.serialize(record.topic(), record.key()), - valueSerializer.serialize(record.topic(), record.value())) + keySerializer.serialize(topic, key), + valueSerializer.serialize(topic, value)) ); } }; @@ -448,10 +477,16 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "anyTaskID") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { changeLog.add(new KeyValue<>( - keySerializer.serialize(record.topic(), record.key()), - valueSerializer.serialize(record.topic(), record.value())) + keySerializer.serialize(topic, key), + valueSerializer.serialize(topic, value)) ); } }; @@ -476,10 +511,16 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestRolling") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { changeLog.add(new KeyValue<>( - keySerializer.serialize(record.topic(), record.key()), - valueSerializer.serialize(record.topic(), record.value())) + keySerializer.serialize(topic, key), + valueSerializer.serialize(topic, value)) ); } }; @@ -604,10 +645,16 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestRestore") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { changeLog.add(new KeyValue<>( - keySerializer.serialize(record.topic(), record.key()), - valueSerializer.serialize(record.topic(), record.value())) + keySerializer.serialize(topic, key), + valueSerializer.serialize(topic, value)) ); } }; @@ -653,10 +700,16 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestRestoreII") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { changeLog.add(new KeyValue<>( - keySerializer.serialize(record.topic(), record.key()), - valueSerializer.serialize(record.topic(), record.value())) + keySerializer.serialize(topic, key), + valueSerializer.serialize(topic, value)) ); } }; @@ -704,8 +757,14 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestSegmentMaintenance") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { - // do nothing + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { + // do nothing } }; @@ -804,8 +863,14 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-TestInitialLoading") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { - // do nothing + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { + // do nothing } }; @@ -865,7 +930,13 @@ public class RocksDBWindowStoreTest { Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); RecordCollector recordCollector = new RecordCollectorImpl(producer, "RocksDBWindowStoreTest-ShouldOnlyIterateOpenSegments") { @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { } }; http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/streams/state/internals/StoreChangeLoggerTest.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/StoreChangeLoggerTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/StoreChangeLoggerTest.java index 82e524e..9b2bd53 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/StoreChangeLoggerTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/StoreChangeLoggerTest.java @@ -23,7 +23,6 @@ package org.apache.kafka.streams.state.internals; import java.util.HashMap; import java.util.Map; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streams.processor.StreamPartitioner; import org.apache.kafka.streams.processor.internals.RecordCollectorImpl; @@ -45,15 +44,27 @@ public class StoreChangeLoggerTest { new RecordCollectorImpl(null, "StoreChangeLoggerTest") { @SuppressWarnings("unchecked") @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { - logged.put((Integer) record.key(), (String) record.value()); + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer) { + logged.put((Integer) key, (String) value); } @Override - public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer, - StreamPartitioner<? super K1, ? super V1> partitioner) { + public <K1, V1> void send(final String topic, + K1 key, + V1 value, + Integer partition, + Long timestamp, + Serializer<K1> keySerializer, + Serializer<V1> valueSerializer, + StreamPartitioner<? super K1, ? super V1> partitioner) { // ignore partitioner - send(record, keySerializer, valueSerializer); + send(topic, key, value, partition, timestamp, keySerializer, valueSerializer); } } ); http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/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 207705c..011532b 100644 --- a/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java +++ b/streams/src/test/java/org/apache/kafka/test/KStreamTestDriver.java @@ -17,7 +17,6 @@ package org.apache.kafka.test; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; @@ -213,16 +212,28 @@ public class KStreamTestDriver { } @Override - public <K, V> void send(ProducerRecord<K, V> record, Serializer<K> keySerializer, Serializer<V> valueSerializer, + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer, StreamPartitioner<? super K, ? super V> partitioner) { // The serialization is skipped. - process(record.topic(), record.key(), record.value()); + process(topic, key, 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()); + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + // The serialization is skipped. + process(topic, key, value); } @Override http://git-wip-us.apache.org/repos/asf/kafka/blob/abe19fe6/streams/src/test/java/org/apache/kafka/test/NoOpRecordCollector.java ---------------------------------------------------------------------- diff --git a/streams/src/test/java/org/apache/kafka/test/NoOpRecordCollector.java b/streams/src/test/java/org/apache/kafka/test/NoOpRecordCollector.java index 0fe4d63..c2833ad 100644 --- a/streams/src/test/java/org/apache/kafka/test/NoOpRecordCollector.java +++ b/streams/src/test/java/org/apache/kafka/test/NoOpRecordCollector.java @@ -16,7 +16,6 @@ */ package org.apache.kafka.test; -import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streams.processor.StreamPartitioner; @@ -28,12 +27,25 @@ import java.util.Map; public class NoOpRecordCollector implements RecordCollector { @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) { - // no-op + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer) { + // no-op } @Override - public <K, V> void send(final ProducerRecord<K, V> record, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final StreamPartitioner<? super K, ? super V> partitioner) { + public <K, V> void send(final String topic, + K key, + V value, + Integer partition, + Long timestamp, + Serializer<K> keySerializer, + Serializer<V> valueSerializer, + StreamPartitioner<? super K, ? super V> partitioner) { // no-op }
