diff --git a/CHANGELOG.md b/CHANGELOG.md index e493d4e5..22e05cb9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ # Changelog ### Unreleased +- Fix inability to materialize without supplying a topic name ### [0.9.6] - [2022-08-01] diff --git a/src/jackdaw/streams/interop.clj b/src/jackdaw/streams/interop.clj index e853087b..9c6b4d62 100644 --- a/src/jackdaw/streams/interop.clj +++ b/src/jackdaw/streams/interop.clj @@ -38,9 +38,11 @@ (Grouped/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..2eac60bd --- /dev/null +++ b/test/jackdaw/streams/interop_test.clj @@ -0,0 +1,15 @@ +(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)}))))