kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: MINOR: Allow timestamp parameter in `ProcessorTopologyTestDriver.process`
Date Thu, 31 Aug 2017 21:57:57 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 75070bdb5 -> 51af6fb65


MINOR: Allow timestamp parameter in `ProcessorTopologyTestDriver.process`

All current implementations process records using the same timestamp. This makes it difficult
to test operations that require time windows, like `KStream-KStream joins`.

This change would allow tests to simulate records created at different times, thus making
it possible to test operations like the above mentioned joins.

Author: Sebastian Gavril <sgavril@wehkamp.nl>

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Closes #3753 from sebigavril/allow-timestamps-in-test-driver


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

Branch: refs/heads/trunk
Commit: 51af6fb654c482fb6fdd85126991149238640fd7
Parents: 75070bd
Author: Sebastian Gavril <sgavril@wehkamp.nl>
Authored: Thu Aug 31 14:57:54 2017 -0700
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Thu Aug 31 14:57:54 2017 -0700

----------------------------------------------------------------------
 .../internals/ProcessorTopologyTest.java        | 46 +++++++++++++++-----
 .../kafka/test/ProcessorTopologyTestDriver.java | 23 +++++++++-
 2 files changed, 56 insertions(+), 13 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/51af6fb6/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
index fd3afa8..1ea09cd 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
@@ -63,7 +63,6 @@ public class ProcessorTopologyTest {
     private static final String OUTPUT_TOPIC_2 = "output-topic-2";
     private static final String THROUGH_TOPIC_1 = "through-topic-1";
 
-    private static long timestamp = 1000L;
     private final TopologyBuilder builder = new TopologyBuilder();
     private final MockProcessorSupplier mockProcessorSupplier = new MockProcessorSupplier();
 
@@ -315,20 +314,43 @@ public class ProcessorTopologyTest {
         assertThat(result, containsString("child-two:\n\t\tchildren:\t[child-two-one]"));
     }
 
-    private void assertNextOutputRecord(String topic, String key, String value) {
-        ProducerRecord<String, String> record = driver.readOutput(topic, STRING_DESERIALIZER,
STRING_DESERIALIZER);
-        assertEquals(topic, record.topic());
-        assertEquals(key, record.key());
-        assertEquals(value, record.value());
-        assertNull(record.partition());
+    @Test
+    public void shouldConsiderTimeStamps() throws Exception {
+        final int partition = 10;
+        driver = new ProcessorTopologyTestDriver(config, createSimpleTopology(partition).internalTopologyBuilder);
+        driver.process(INPUT_TOPIC_1, "key1", "value1", STRING_SERIALIZER, STRING_SERIALIZER,
10L);
+        driver.process(INPUT_TOPIC_1, "key2", "value2", STRING_SERIALIZER, STRING_SERIALIZER,
20L);
+        driver.process(INPUT_TOPIC_1, "key3", "value3", STRING_SERIALIZER, STRING_SERIALIZER,
30L);
+        assertNextOutputRecord(OUTPUT_TOPIC_1, "key1", "value1", partition, 10L);
+        assertNextOutputRecord(OUTPUT_TOPIC_1, "key2", "value2", partition, 20L);
+        assertNextOutputRecord(OUTPUT_TOPIC_1, "key3", "value3", partition, 30L);
     }
 
-    private void assertNextOutputRecord(String topic, String key, String value, Integer partition)
{
+
+    private void assertNextOutputRecord(final String topic,
+                                        final String key,
+                                        final String value) {
+        assertNextOutputRecord(topic, key, value, null, 0L);
+    }
+
+    private void assertNextOutputRecord(final String topic,
+                                        final String key,
+                                        final String value,
+                                        final Integer partition) {
+        assertNextOutputRecord(topic, key, value, partition, 0L);
+    }
+
+    private void assertNextOutputRecord(final String topic,
+                                        final String key,
+                                        final String value,
+                                        final Integer partition,
+                                        final Long timestamp) {
         ProducerRecord<String, String> record = driver.readOutput(topic, STRING_DESERIALIZER,
STRING_DESERIALIZER);
         assertEquals(topic, record.topic());
         assertEquals(key, record.key());
         assertEquals(value, record.value());
         assertEquals(partition, record.partition());
+        assertEquals(timestamp, record.timestamp());
     }
 
     private void assertNoOutputRecord(String topic) {
@@ -543,18 +565,20 @@ public class ProcessorTopologyTest {
 
     /**
      * A custom timestamp extractor that extracts the timestamp from the record's value if
the value is in ".*@[0-9]+"
-     * format. Otherwise, it returns the record's timestamp or the default timestamp if the
record's timestamp is zero.
+     * format. Otherwise, it returns the record's timestamp or the default timestamp if the
record's timestamp is negative.
     */
     public static class CustomTimestampExtractor implements TimestampExtractor {
+        private static final long DEFAULT_TIMESTAMP = 1000L;
+
         @Override
         public long extract(final ConsumerRecord<Object, Object> record, final long
previousTimestamp) {
             if (record.value().toString().matches(".*@[0-9]+"))
                 return Long.parseLong(record.value().toString().split("@")[1]);
 
-            if (record.timestamp() > 0L)
+            if (record.timestamp() >= 0L)
                 return record.timestamp();
 
-            return timestamp;
+            return DEFAULT_TIMESTAMP;
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/51af6fb6/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
b/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
index 47124ed..deba860 100644
--- a/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
+++ b/streams/src/test/java/org/apache/kafka/test/ProcessorTopologyTestDriver.java
@@ -240,7 +240,7 @@ public class ProcessorTopologyTestDriver {
      * @param value the raw message value
      * @param timestamp the raw message timestamp
      */
-    private void process(final String topicName,
+    public void process(final String topicName,
                          final byte[] key,
                          final byte[] value,
                          final long timestamp) {
@@ -309,7 +309,26 @@ public class ProcessorTopologyTestDriver {
                                final V value,
                                final Serializer<K> keySerializer,
                                final Serializer<V> valueSerializer) {
-        process(topicName, keySerializer.serialize(topicName, key), valueSerializer.serialize(topicName,
value));
+        process(topicName, key, value, keySerializer, valueSerializer, 0L);
+    }
+
+    /**
+     * Send an input message with the given key and value and timestamp on the specified
topic to the topology.
+     *
+     * @param topicName the name of the topic on which the message is to be sent
+     * @param key the raw message key
+     * @param value the raw message value
+     * @param keySerializer the serializer for the key
+     * @param valueSerializer the serializer for the value
+     * @param timestamp the raw message timestamp
+     */
+    public <K, V> void process(final String topicName,
+                               final K key,
+                               final V value,
+                               final Serializer<K> keySerializer,
+                               final Serializer<V> valueSerializer,
+                               final long timestamp) {
+        process(topicName, keySerializer.serialize(topicName, key), valueSerializer.serialize(topicName,
value), timestamp);
     }
 
     /**


Mime
View raw message