Add flat-transform wrappers for Kafka Streams #245

99-not-out · 2020-04-27T22:47:30Z

Spotted flatTransform for a KStream and it looks worth adding. Also added a few of the common bits of wrapper / helper / sugar code which pop up from uses of transformers.

Add support for flat-transform and flat-transform-values
Add some helpers for adding state stores
Add some sugar for common transform use cases
Use some of the above to provide simple de-duping stream support in extras

With flatTransformValues note that if you return an empty list or nil, it halts processing. With transformValues, if you return nil processing is not halted and the next processor in the stream sees [k nil]. This difference can be useful if you meant to halt processing, as it avoids a subsequent filter.

@creese We really need to get your transducers changes merged too!

99-not-out · 2020-04-27T22:53:17Z

Coverage doesn't seem to like jackdaw/streams/protocols.clj anymore 😞

.circleci/config.yml

kidpollo

This is great. Some comments, Nothing blocking

kidpollo · 2020-04-28T19:13:05Z

src/jackdaw/streams.clj

@@ -200,6 +215,15 @@
  ([kstream value-transformer-supplier-fn state-store-names]
   (p/transform-values kstream value-transformer-supplier-fn state-store-names)))

+(defn flat-transform-values


why not flat-transform-keys ?

kidpollo · 2020-04-28T19:34:39Z

src/jackdaw/streams/interop.clj

@@ -323,7 +333,20 @@
    (clj-kstream
     (.transform ^KStream kstream
                 ^TransformerSupplier (transformer-supplier transformer-supplier-fn)
-                 ^"[Ljava.lang.String;" (into-array String state-store-names))))
+                 ^"[Ljava.lang.String;" (into-array String


is that how a string array is annotated? 🤯

Yeah, although arrays are objects in Java they are not an instance of a class, so have a slightly funky type signature :-)

kidpollo · 2020-04-28T19:45:47Z

test/jackdaw/streams_test.clj

+                                          (k/with-state-store {:store-name "test-store"})
+                                          (k/kstream topic-a)
+                                          (k/transform
+                                           (lambdas/transformer-with-ctx


Wondering if a macro could make this feel more 'clojurey' like

(transformer-with-context [test-store k v] (update test-store k some-fn) )

Yeah there is possible a macro here as the state store names have to be repeated. I try to avoid reaching for macro's immediately however until the basic functions are there.

We could perhaps start a macros namespace, as there are probably several syntax things you could add over the basic functions for use in streams.

99-not-out · 2020-04-29T14:45:49Z

src/jackdaw/streams/configured.clj

@@ -52,6 +52,10 @@
    [_]
    (source-topics streams-builder))

+  (with-state-store


I am not 100% on the naming here - maybe add-state-store. However this name make the code using it read nicer IMHO. (-> builder (k/with-state-store ...) (k/stream ...)) etc.

99-not-out · 2020-04-29T15:13:58Z

src/jackdaw/streams/interop.clj

+  (kstream* [_]
+    kstream)
+
+  ITransformingKStream


Added a new protocol for the transform fns on a stream, which is always used wherever IKStream is used. This had to be done to allow Coverage to run over this IKStream protocol.

Its not ideal, and perhaps there is a newer / different code coverage tool to fix this.

The change is however invisible to users of Jackdaw.

codecov · 2020-04-29T15:16:34Z

Codecov Report

Merging #245 into master will increase coverage by 0.70%.
The diff coverage is 90.12%.

@@            Coverage Diff             @@
##           master     #245      +/-   ##
==========================================
+ Coverage   80.82%   81.53%   +0.70%     
==========================================
  Files          41       41              
  Lines        2519     2583      +64     
  Branches      149      149              
==========================================
+ Hits         2036     2106      +70     
+ Misses        334      328       -6     
  Partials      149      149

Impacted Files	Coverage Δ
src/jackdaw/streams/configured.clj	`69.23% <ø> (ø)`
src/jackdaw/streams/interop.clj	`80.28% <ø> (ø)`
src/jackdaw/streams/mock.clj	`74.24% <25.00%> (-3.18%)`	⬇️
src/jackdaw/streams.clj	`83.33% <85.71%> (+1.07%)`	⬆️
src/jackdaw/streams/specs.clj	`97.04% <91.30%> (+0.15%)`	⬆️
src/jackdaw/streams/extras.clj	`35.29% <95.45%> (+28.77%)`	⬆️
src/jackdaw/streams/lambdas.clj	`75.71% <100.00%> (+7.78%)`	⬆️
src/jackdaw/streams/protocols.clj	`100.00% <100.00%> (ø)`
src/jackdaw/serdes/edn2.clj	`90.90% <0.00%> (+45.45%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update d53da3f...0d7f9e1. Read the comment docs.

* Add support for flat-transform and flat-transform-values * Add some helpers for adding state stores * Add some sugar for common transform use cases * Use some of the above to provide simple de-duping stream support in extras

* It cannot process the interop.clj namespace, which is otherwsie fine.

kidpollo · 2021-11-16T20:02:55Z

This PR and the #305 seem to have related/similar changes. @99-not-out Is there any insight on which will win hahaha or if both would be merged. I kinda want some of these stuff. Would love this merged and a new release cut sometime soon :P

kidpollo · 2021-11-16T23:27:29Z

@99-not-out would it make sense to add a simple interop to set message headers?

Right now one would have to do something like:

...

(defn set-headers [context headers]
  (let [record-headers (.headers context)]
    (doseq [[k v] headers]
      (.add record-headers
            k
            (.getBytes v))))
  context)
...
(-> stream
          (j/transform-values
           (jl/value-transformer-with-ctx
            (fn [ctx v]
              (set-headers ctx {"foo" "bar"})
              v)))

99-not-out · 2021-11-17T09:37:45Z

Going to close this PR in favour of #305
(which takes the core parts of this PR and will hopefully me merged soon!)

99-not-out · 2021-11-17T09:38:10Z

This PR and the #305 seem to have related/similar changes. @99-not-out Is there any insight on which will win hahaha or if both would be merged. I kinda want some of these stuff. Would love this merged and a new release cut sometime soon :P

#305 will win :-)

99-not-out · 2021-11-17T09:40:41Z

@99-not-out would it make sense to add a simple interop to set message headers?

Right now one would have to do something like:

...

(defn set-headers [context headers]
  (let [record-headers (.headers context)]
    (doseq [[k v] headers]
      (.add record-headers
            k
            (.getBytes v))))
  context)
...
(-> stream
          (j/transform-values
           (jl/value-transformer-with-ctx
            (fn [ctx v]
              (set-headers ctx {"foo" "bar"})
              v)))

I did have a bit of a think about headers a while a go - there is even a PR somewhere I think to add support in test machine.

Lets start a new PR specifically for headers @kidpollo - there are some nuances in there (like supporting encodings, so the interop isn't all byte[] based ...). Will tag you wen we start one (or feel free to start one yourself!)

gphilipp · 2021-11-17T11:22:21Z

Superseded by #305.

99-not-out requested a review from a team as a code owner April 27, 2020 22:47

99-not-out commented Apr 28, 2020

View reviewed changes

.circleci/config.yml Outdated Show resolved Hide resolved

kidpollo reviewed Apr 28, 2020

View reviewed changes

99-not-out commented Apr 29, 2020

View reviewed changes

Matt Searle added 3 commits November 11, 2021 12:28

Add flat-transform wrappers for Kafka Streams

d4bfd1b

* Add support for flat-transform and flat-transform-values * Add some helpers for adding state stores * Add some sugar for common transform use cases * Use some of the above to provide simple de-duping stream support in extras

Remove code coverage check for now

51587cf

* It cannot process the interop.clj namespace, which is otherwsie fine.

Split IKStream protocol up to allow Cloverage to run on it

d115dea

99-not-out force-pushed the more-interop branch from 0d7f9e1 to d115dea Compare November 11, 2021 12:29

99-not-out closed this Nov 17, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add flat-transform wrappers for Kafka Streams #245

Add flat-transform wrappers for Kafka Streams #245

99-not-out commented Apr 27, 2020 •

edited

Loading

99-not-out commented Apr 27, 2020

kidpollo left a comment

kidpollo Apr 28, 2020

kidpollo Apr 28, 2020

99-not-out Apr 29, 2020

kidpollo Apr 28, 2020

99-not-out Apr 29, 2020

99-not-out Apr 29, 2020

99-not-out Apr 29, 2020

codecov bot commented Apr 29, 2020 •

edited

Loading

kidpollo commented Nov 16, 2021

kidpollo commented Nov 16, 2021

99-not-out commented Nov 17, 2021

99-not-out commented Nov 17, 2021

99-not-out commented Nov 17, 2021

gphilipp commented Nov 17, 2021 •

edited

Loading

Add flat-transform wrappers for Kafka Streams #245

Add flat-transform wrappers for Kafka Streams #245

Conversation

99-not-out commented Apr 27, 2020 • edited Loading

99-not-out commented Apr 27, 2020

kidpollo left a comment

Choose a reason for hiding this comment

kidpollo Apr 28, 2020

Choose a reason for hiding this comment

kidpollo Apr 28, 2020

Choose a reason for hiding this comment

99-not-out Apr 29, 2020

Choose a reason for hiding this comment

kidpollo Apr 28, 2020

Choose a reason for hiding this comment

99-not-out Apr 29, 2020

Choose a reason for hiding this comment

99-not-out Apr 29, 2020

Choose a reason for hiding this comment

99-not-out Apr 29, 2020

Choose a reason for hiding this comment

codecov bot commented Apr 29, 2020 • edited Loading

Codecov Report

kidpollo commented Nov 16, 2021

kidpollo commented Nov 16, 2021

99-not-out commented Nov 17, 2021

99-not-out commented Nov 17, 2021

99-not-out commented Nov 17, 2021

gphilipp commented Nov 17, 2021 • edited Loading

99-not-out commented Apr 27, 2020 •

edited

Loading

codecov bot commented Apr 29, 2020 •

edited

Loading

gphilipp commented Nov 17, 2021 •

edited

Loading