Repository: FundingCircle/jackdaw (clojure)

Issue:
Cannot create record with ConsumerRecordFactory when the key is of type String
When trying to test my topology I stumbled upon a bug. I'm trying to test a topology that groups a stream in time windows. When the topic's `key` is a String, the wrong method in the `ConsumerRecordFactory` is used when using an explicit `time-ms`.

This happens in `jackdaw.streams.mock/procucer`: `(.pipeInput test-driver (.create record-factory k v time-ms))`. The intention is to use `public ConsumerRecord<byte[], byte[]> create(final K key, final V value, final long timestampMs)` in line 227, but instead it then uses `public ConsumerRecord<byte[], byte[]> create(final String topicName, final K key, final V value) ` in line 265.

I was able to create another test case for `streams-test` that reproduces the issue:

```clojure
  (testing "windowed-by-time with string keys"
    (let [topic-a (assoc (mock/topic "topic-a") :key-serde (Serdes/String))
          topic-b (assoc (mock/topic "topic-b") :key-serde (Serdes/String))
          driver (mock/build-driver (fn [builder]
                                      (-> builder
                                          (k/kstream topic-a)
                                          (k/group-by-key)
                                          (k/window-by-time (TimeWindows/of 1000))
                                          (k/reduce + topic-a)
                                          (k/to-kstream)
                                          (k/map (fn [[k v]] [(.key k) v]))
                                          (k/to topic-b))))
          publish (partial mock/publish driver topic-a)]

      (publish 1000 "a" 1)
      (publish 1500 "a" 2)
      (publish 5000 "a" 4)

      (let [keyvals (mock/get-keyvals driver topic-b)]
        (is (= 3 (count keyvals)))
        (is (= [0 1] (first keyvals)))
        (is (= [0 3] (second keyvals)))
        (is (= [0 4] (nth keyvals 2))))))
```

Pull request description:
When using strings as keys, the wrong method for creating
ConsumerRecords is used. This fixes that by explicitly using the most
specific underlying method.

Closes #19

Patch:
diff --git a/src/jackdaw/streams/mock.clj b/src/jackdaw/streams/mock.clj
index 3298e64a..ac9e9e78 100644
--- a/src/jackdaw/streams/mock.clj
+++ b/src/jackdaw/streams/mock.clj
@@ -11,6 +11,7 @@
            org.apache.kafka.streams.TopologyTestDriver
            java.util.Properties
            org.apache.kafka.streams.test.ConsumerRecordFactory
+           org.apache.kafka.common.header.internals.RecordHeaders
            [org.apache.kafka.common.serialization Serde Serdes Serializer]))
 
 (defn streams-builder
@@ -69,7 +70,7 @@
       ([k v]
        (.pipeInput test-driver (.create record-factory k v)))
       ([time-ms k v]
-       (.pipeInput test-driver (.create record-factory k v time-ms))))))
+       (.pipeInput test-driver (.create record-factory topic-name k v (RecordHeaders.) time-ms))))))
 
 (defn publish
   ([test-driver topic-config k v]


Test patch:
diff --git a/test/jackdaw/streams_test.clj b/test/jackdaw/streams_test.clj
index 21df9cc9..e3b20d34 100644
--- a/test/jackdaw/streams_test.clj
+++ b/test/jackdaw/streams_test.clj
@@ -14,7 +14,8 @@
   (:import [org.apache.kafka.streams.kstream
             JoinWindows SessionWindows TimeWindows Transformer
             ValueTransformer]
-           org.apache.kafka.streams.StreamsBuilder))
+           org.apache.kafka.streams.StreamsBuilder
+           [org.apache.kafka.common.serialization Serdes]))
 
 (stest/instrument)
 
@@ -696,6 +697,30 @@
         (is (= [0 3] (second keyvals)))
         (is (= [0 4] (nth keyvals 2))))))
 
+  (testing "windowed-by-time with string keys"
+    (let [topic-a (assoc (mock/topic "topic-a") :key-serde (Serdes/String))
+          topic-b (assoc (mock/topic "topic-b") :key-serde (Serdes/String))
+          driver (mock/build-driver (fn [builder]
+                                      (-> builder
+                                          (k/kstream topic-a)
+                                          (k/group-by-key)
+                                          (k/window-by-time (TimeWindows/of 1000))
+                                          (k/reduce + topic-a)
+                                          (k/to-kstream)
+                                          (k/map (fn [[k v]] [(.key k) v]))
+                                          (k/to topic-b))))
+          publish (partial mock/publish driver topic-a)]
+
+      (publish 1000 "a" 1)
+      (publish 1500 "a" 2)
+      (publish 5000 "a" 4)
+
+      (let [keyvals (mock/get-keyvals driver topic-b)]
+        (is (= 3 (count keyvals)))
+        (is (= ["a" 1] (first keyvals)))
+        (is (= ["a" 3] (second keyvals)))
+        (is (= ["a" 4] (nth keyvals 2))))))
+
   (testing "windowed-by-session"
     (let [topic-a (mock/topic "topic-a")
           topic-b (mock/topic "topic-b")