You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2017/01/25 00:36:34 UTC
kafka git commit: KAFKA-3835;
Streams is creating two ProducerRecords for each send via
RecordCollector
Repository: kafka
Updated Branches:
refs/heads/trunk 63cf0e587 -> abe19fe69
KAFKA-3835; Streams is creating two ProducerRecords for each send via RecordCollector
Author: Jeyhun Karimov <je...@gmail.com>
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 <je...@gmail.com>
Authored: Tue Jan 24 16:36:31 2017 -0800
Committer: Guozhang Wang <wa...@gmail.com>
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
}