Adds IKStreamBase/join with arity 5 to configure left and right serdes and possibly name the join

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ### Unreleased
+- Adds IKStreamBase/join with arity 5 to configure left and right serdes and possibly name the join
 
 ### [0.9.12] - [2023-12-05]
 - Support for Foreign Key joins [#365](https://github.com/FundingCircle/jackdaw/pull/365) (Issue [#364])

src/jackdaw/streams.clj

@@ -64,6 +64,8 @@
   share the same key or a foreign key using an inner join."
   ([kstream-or-ktable ktable value-joiner-fn]
    (p/join kstream-or-ktable ktable value-joiner-fn))
+  ([kstream-or-ktable ktable value-joiner-fn this-topic-config other-topic-config]
+   (p/join kstream-or-ktable ktable value-joiner-fn this-topic-config other-topic-config))
   ([kstream-or-ktable ktable foreign-key-extractor-fn value-joiner-fn]
    (p/join kstream-or-ktable ktable foreign-key-extractor-fn value-joiner-fn)))
 

src/jackdaw/streams/configured.clj

@@ -81,6 +81,12 @@
      config
      (join kstream ktable value-joiner-fn)))
 
+  (join
+    [_ ktable value-joiner-fn topic-config other-topic-config]
+    (configured-kstream
+     config
+     (join kstream ktable value-joiner-fn topic-config other-topic-config)))
+
   (left-join
     [_ ktable value-joiner-fn]
     (configured-kstream

src/jackdaw/streams/interop.clj

@@ -44,6 +44,14 @@
     key-serde (.withKeySerde key-serde)
     value-serde (.withValueSerde value-serde)))
 
+(defn ->joined
+  "Builds a Joined object that represents optional params that can be
+  passed to join and left-join operations."
+  ([key-serde value-serde other-value-serde]
+   (Joined/with key-serde value-serde other-value-serde))
+  ([key-serde value-serde other-value-serde joined-name]
+   (Joined/with key-serde value-serde other-value-serde joined-name)))
+
 (defn suppress-config->suppressed
   [{:keys [max-records max-bytes until-time-limit-ms]}]
   (let [config (cond
@@ -154,6 +162,16 @@
             ^KTable (ktable* ktable)
             ^ValueJoiner (value-joiner value-joiner-fn))))
 
+  (join
+    [_ ktable value-joiner-fn
+     {key-serde :key-serde this-value-serde :value-serde joined-name :name}
+     {other-value-serde :value-serde}]
+    (clj-kstream
+     (.join ^KStream kstream
+            ^KTable (ktable* ktable)
+            ^ValueJoiner (value-joiner value-joiner-fn)
+            ^Joined (->joined key-serde this-value-serde other-value-serde joined-name))))
+
   (left-join
     [_ ktable value-joiner-fn]
     (clj-kstream
@@ -163,13 +181,13 @@
 
   (left-join
     [_ ktable value-joiner-fn
-     {key-serde :key-serde this-value-serde :value-serde}
+     {key-serde :key-serde this-value-serde :value-serde joined-name :name}
      {other-value-serde :value-serde}]
     (clj-kstream
      (.leftJoin kstream
                 ^KTable (ktable* ktable)
                 ^ValueJoiner (value-joiner value-joiner-fn)
-                (Joined/with key-serde this-value-serde other-value-serde))))
+                ^Joined (->joined key-serde this-value-serde other-value-serde joined-name))))
 
   (peek
     [_ peek-fn]

src/jackdaw/streams/protocols.clj

@@ -42,6 +42,7 @@
   "Methods common to KStream and KTable."
   (join
     [kstream-or-ktable ktable value-joiner-fn]
+    [kstream-or-ktable ktable value-joiner-fn this-topic-config other-topic-config]
     [kstream-or-ktable ktable foreign-key-extractor-fn value-joiner-fn]
     "Combines the values of the KStream-or-KTable with the values of the
     KTable that share the same key using an inner join. For foreign key

src/jackdaw/streams/specs.clj

@@ -62,6 +62,12 @@
 
 (s/def ::topic-configs (s/coll-of ::topic-config))
 
+(s/def ::name string?)
+(s/def ::join-config
+  (s/keys :req-un [::key-serde
+                   ::value-serde]
+          :opt-un [::name]))
+
 (s/def ::kstreams (s/coll-of kstream?))
 (s/def ::kstream-or-ktable (s/or :kstream kstream? :ktable ktable?))
 
@@ -103,7 +109,9 @@
 (s/fdef k/join
         :args (s/or :pk-join (s/cat :kstream-or-ktable ::kstream-or-ktable
                                     :ktable ktable?
-                                    :value-joiner-fn ifn?)
+                                    :value-joiner-fn ifn?
+                                    :this-topic-config (s/? ::join-config)
+                                    :other-topic-config (s/? ::join-config))
                     :fk-join (s/cat :kstream-or-ktable ::kstream-or-ktable
                                     :ktable ktable?
                                     :fk-extractor-fn ifn?
@@ -114,8 +122,8 @@
         :args (s/cat :kstream-or-ktable ::kstream-or-ktable
                      :ktable ktable?
                      :value-joiner-fn ifn?
-                     :this-topic-config (s/? ::topic-config)
-                     :other-topic-config (s/? ::topic-config))
+                     :this-topic-config (s/? ::join-config)
+                     :other-topic-config (s/? ::join-config))
         :ret ::kstream-or-ktable)
 
 (s/fdef k/for-each!

test/jackdaw/streams_test.clj

@@ -61,12 +61,17 @@
   (testing "join"
     (let [topic-a (mock/topic "table-a")
           topic-b (mock/topic "table-b")
-          topic-c (mock/topic "topic-c")]
+          topic-c (mock/topic "topic-c")
+          join-config {:key-serde (Serdes/Long)
+                       :value-serde (Serdes/Long)
+                       :name "a-b-join"}]
 
       (with-open [driver (mock/build-driver (fn [builder]
                                               (let [left (k/kstream builder topic-a)
                                                     right (k/ktable builder topic-b)]
-                                                (-> (k/join left right +)
+                                                (-> (k/join left right +
+                                                            join-config
+                                                            topic-b)
                                                     (k/to topic-c)))))]
         (let [publish-left (partial mock/publish driver topic-a)
               publish-right (partial mock/publish driver topic-b)]