davewm / willa Goto Github PK

A left-join of a KStream and a KTable is legal (and in my use-cases, somewhat common) operation. However, the experiment namespace doesn't support this case.

where it reads:
(w/build-workflow! topology)) should be
(w/build-topology! topology))

I wanted to check in to see if you have put any thought onto how support for kstream branching via jackdaw.streams/branch might work. Seems like it might be representable in the willa build-topology! map but there's probably options there about the exact structure of the data form and how the code to build would look.

I couldn't figure this one out and it's not really referenced in the documentation that I could see. My use-case is establishing a ktable with the intention to query it outside of the topology as a local state store. I need a consistent name instead of a gensym for obvious reasons. I looked in the code and believe I determined that this is not supported, so I have pushed a PR for you to check out, to see if it's something you'd like to support.

#14

Hello! Thanks for the library.

I noticed that the recently added custom store-name key is not used if an :aggregate-adder-fn is also defined. Is this working as intended?

Also, the key name mentioned in the issue for the original PR mentions :willa.core/store-name, but tE code seems to use :willa.streams/store.name, is that right?

(def topology 
  {:workflow workflow
   :entities entities
   :joins joins})

joins is not defined anywhere above this.
I guess it will be {} in this case?

Hi, our org just switched from Apache Camel to your lib, and we are liking it so far.

Question: is there any way to call a KStream directly with a function? Maybe I'm asking too high up the stack here but perhaps you know. With Camel, you could call! any route directly from your code synchronously or asynchronously, and it would flow through all the downstream routes as you've defined them. We'd like to do the same thing with Kafka Streams, given the key describing the stream in our Willa topology.

I'm trying to figure out if it is possible to access a local state store inside
a willa workflow.

Is their something like transform from jackdaw?
https://cljdoc.org/d/fundingcircle/jackdaw/0.8.0/api/jackdaw.streams#transform

For context I'm trying to do something like:
https://github.com/brycecovert/clojure-event-sourcing/blob/8dd6974eb050c41bf8f5600bc2ed9e213ff2f71e/src/event_sourcing/passenger_counting.clj#L64

Hi, thanks for this library and its nice readme.
I think there is an error where you define the entities:

(def entities
  {:input-topic {::w/entity-type :topic
                 :topic-name "output-topic" ;; <======== HERE =======
                 :replication-factor 1
                 :partition-count 1
                 :key-serde (serdes.edn/serde)
                 :value-serde (serdes.edn/serde)}
   :increment-stream {::w/entity-type :kstream
                      :willa.core/xform (map (fn [[k v]] [k (inc v)])) ;; Note that the mapping function expects a key-value tuple
                      }
   :output-topic {::w/entity-type :topic
                  :topic-name "output-topic"
                  :replication-factor 1
                  :partition-count 1
                  :key-serde (serdes.edn/serde)
                  :value-serde (serdes.edn/serde)}})

I think you meant "input-topic" there!

Thanks again.

When I try to build this topology I get a No method in multimethod 'coerce-to-ktable' for dispatch value: class jackdaw.streams.interop.CljTimeWindowedKStream exception.

Changing the :ktable to :kstream will allow the topology to build but I think it produces different semantics of the stream.

(def workflow
  [[:input-topic :aggregate-stream]
   [:aggregate-stream :output-topic]])

(def entities
 {:input-topic {::w/entity-type :topic
                :topic-name "timed-readings-topic"
                :replication-factor 1
                :partition-count 1
                :key-serde (serdes.edn/serde)
                :value-serde (serdes.edn/serde)}
  :aggregate-stream {::w/entity-type :ktable
                     :willa.core/group-by-fn (fn [[k v]] (:device-number v))
                     :willa.core/window (TimeWindows/of (* 2 60 1000))} 
  :output-topic {::w/entity-type :topic
                 :topic-name "timed-readings-output-topic"
                 :replication-factor 1
                 :partition-count 1
                 :key-serde (serdes.edn/serde)
                 :value-serde (serdes.edn/serde)}})

When running an experiment in willa where I have a ktable that has a key added to it then removed from it the aggregate-subtractor-fn I supply does not seem to get called. When looking in the willa.experiment namespace I see references to aggregate-adder-fn though not to aggregate-subtractor-fun. Does the experiment module currently not support where you add a key to a ktable then remove it it?

If I load willa.core in a repl running in docker, I see this error:

Syntax error (HeadlessException) compiling at (viz.clj:27:1).

No X11 DISPLAY variable was set, but this program performed an operation which requires it.
class clojure.lang.Compiler$CompilerException

Could willa.core be restructured to not load viz or rhizome stuff?

Something may have changed within the Kafka Streams API. Trying to bump the jackdaw version for Willa causes tests to fail because:

java.lang.NoSuchMethodError: 'org.apache.kafka.streams.processor.internals.ProcessorTopology org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.build(java.lang.Integer)

I'm just starting to keep up with this stuff, it looks like InternalTopologyBuilder has been deprecated.

Any idea what needs to be done @DaveWM?

Problem

When using ::willa/xform kafka streams will always repartition the topic, since we might have changed the key. There should be some way of transforming only the values, to make it possible to avoid repartitioning.

Suggested solution

Add ::willa/xform-values which takes a transducer of the shape (map (fn [[k v]] v)).

Hey there! My co-worker recommended your library to me and I was reading through your blog post.

It looks great!

When I ran (wv/view-topology topology), the following exception was thrown:

Execution error (IOException) at java.lang.UNIXProcess/forkAndExec (UNIXProcess.java:-2). error=2, No such file or directory

After some digging, it looks like rhizome has a dependency on graphviz, which has to be installed by an os package manager. Installing graphviz solves the issue.

I think it would be helpful to have some documentation noting the need to install the dependency to use this method & maybe a try around the usage of rhizome to help suggest this as a possible solution.

Not sure if this is supposed to work, but FYI:

Syntax error (ClassNotFoundException) compiling new at (jackdaw/test.clj:250:5). org.apache.kafka.streams.processor.internals.ChangelogRegister

and the second time I run lein test I get:

Syntax error (ClassNotFoundException) compiling new at (jackdaw/test.clj:250:5). org.apache.kafka.streams.processor.internals.ClientUtils$QuietStreamsConfig

Here is the project.clj I tried to use:

(defproject willa "0.2.0"
  :description "A Clojure DSL for Kafka Streams"
  :url "https://github.com/DaveWM/willa"
  :license {:name "GPL V3"
            :url "https://www.gnu.org/licenses/gpl-3.0.en.html"}
  :dependencies [[org.clojure/clojure "1.10.3"]
                 [fundingcircle/jackdaw "0.7.8"]
                 [aysylu/loom "1.0.2"]
                 [rhizome "0.2.9"]
                 [org.clojure/math.combinatorics "0.1.6"]]
  :repl-options {:init-ns willa.core}
  :repositories [["confluent" "https://packages.confluent.io/maven/"]
                 ["clojars" "https://clojars.org/repo/"]]
  :profiles {:test {:dependencies [[org.apache.kafka/kafka-streams-test-utils "2.7.0"]
                                   [log4j/log4j "1.2.17"]]}
             :dev {:dependencies [[org.clojure/test.check "1.1.0"]]}
             :ci {:jvm-opts ["-Djava.awt.headless=true"]}})

I rarely use lein test, so maybe I don't know what I am doing....

Then I went back to your master branch, and that branch tests fine (as one would expect):

lein test willa.test-utils
lein test willa.unit.core-test
lein test willa.unit.experiment-and-ttd-congruity-test
lein test willa.unit.streams-test
lein test willa.unit.utils-test

Ran 19 tests containing 43 assertions.
0 failures, 0 errors.