diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a58c2fa..34e1d6dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## Unreleased * Fix `ktable` constructor to use supplied `store-name` +* Fix inability to materialize without supplying a topic name ## [0.7.3] - [2020-04-08] diff --git a/src/jackdaw/streams/interop.clj b/src/jackdaw/streams/interop.clj index 6f4254bc..362636fc 100644 --- a/src/jackdaw/streams/interop.clj +++ b/src/jackdaw/streams/interop.clj @@ -40,9 +40,12 @@ (Serialized/with key-serde value-serde)) (defn topic->materialized [{:keys [topic-name key-serde value-serde]}] - (cond-> (Materialized/as ^String topic-name) - key-serde (.withKeySerde key-serde) - value-serde (.withValueSerde value-serde))) + (if topic-name + (cond-> (Materialized/as ^String topic-name) + key-serde (.withKeySerde key-serde) + value-serde (.withValueSerde value-serde)) + (Materialized/with key-serde + value-serde))) (defn suppress-config->suppressed [{:keys [max-records max-bytes until-time-limit-ms]}] diff --git a/test/jackdaw/streams/interop_test.clj b/test/jackdaw/streams/interop_test.clj new file mode 100644 index 00000000..3c8c5b83 --- /dev/null +++ b/test/jackdaw/streams/interop_test.clj @@ -0,0 +1,16 @@ +(ns jackdaw.streams.interop-test + (:require + [clojure.test :refer :all] + [jackdaw.streams.interop :as interop]) + (:import + (org.apache.kafka.common.serialization + Serdes))) + +(deftest topic->materialized-test + (testing "materialized with serde only doesn't smoke" + (is (interop/topic->materialized {:key-serde (Serdes/String) + :value-serde (Serdes/String)}))) + (testing "materialized with topic name and serde doesn't smoke" + (is (interop/topic->materialized {:key-serde (Serdes/String) + :value-serde (Serdes/String)})))) +