fundingcircle / jackdaw Goto Github PK

Found this warning while working on another issue.

https://docs.confluent.io/current/streams/javadocs/org/apache/kafka/streams/kstream/KStream.html#groupByKey-org.apache.kafka.streams.kstream.Serialized-

https://github.com/FundingCircle/jackdaw/blob/master/src/jackdaw/streams/interop.clj#L211

So folks know how to use it, and what it does and doesn't do.

Occasional build failures due to gen/any-printable producing #NaN sometimes in the EDN test. This comes from the double generator used by any-printable.

Spotted by @r0man

Example failing build: https://circleci.com/gh/FundingCircle/jackdaw/65?utm_campaign=workflow-failed&utm_medium=email&utm_source=notification

Great for on the fly updating what topics are being listened to by the consumer.

https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#pause-java.util.Collection-

https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#paused--

https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#resume-java.util.Collection-

Astute followers may have noticed some flakeyness in the rest-proxy transport test. On closer inspection, this does indeed seem like a bug although I'm not quite sure yet how it's happening. When the transport is created, it is supposed to wait until the consumer has performed at least one poll before it "starts" the producer and allows any messages to be submitted by the test machine. This is intended to ensure that the journal captures all output initiated by the current test.

https://github.com/FundingCircle/jackdaw/blob/master/src/jackdaw/test/transports/rest_proxy.clj#L289

However careful inspection of the log for a test failure reveals that when the test intermittently fails, it looks like the producer is indeed being "started" before the consumer

I'm not sure if its how jackdaw creates its default client or not but I couldn't find a way to set or get it to pick up the piece of config to tell it to extract the basic auth creds from the URL. A simple workaround is to create one yourself

(defn cached-schema-registry-cli []
  (let [rest-service (RestService. "https://user:[email protected]:13584")]
    (CachedSchemaRegistryClient. rest-service 10 (doto (new java.util.HashMap) (.put "basic.auth.credentials.source" "URL")))))

(def serde-resolver
  (partial resolver/serde-resolver :schema-registry-url "" :schema-registry-client (cached-schema-registry-cli)))

Minor points but it would also be nice to not require a url for the resolver and also the topic-metadata map insists on a default schema even if using it from the registry.

Thanks.

Many of the operations in the admin namespace block on some upstream result before transforming the result into something a bit more idiomatic for Clojure programs. If the operation is unable to complete (e.g. it is unable to connect to the configured broker), the operation will never return, potentially blocking your REPL.

Returning a manifold deferred instead allows jackdaw to asynchronously perform the same transformation but instead of returning the transformed object after blocking, immediately return a deferred that eventually resolves to the transformed result.

In addition, functions like create-topics! currently return a boolean indicating whether all operations were completed. The underlying kafka operations allow us to infer which operations failed and provide representation of this to the caller.

Current jackdaw.admin

Proposed jackdaw.admin

When trying to test my topology I stumbled upon a bug. I'm trying to test a topology that groups a stream in time windows. When the topic's key is a String, the wrong method in the ConsumerRecordFactory is used when using an explicit time-ms.

This happens in jackdaw.streams.mock/procucer: (.pipeInput test-driver (.create record-factory k v time-ms)). The intention is to use public ConsumerRecord<byte[], byte[]> create(final K key, final V value, final long timestampMs) in line 227, but instead it then uses public ConsumerRecord<byte[], byte[]> create(final String topicName, final K key, final V value) in line 265.

I was able to create another test case for streams-test that reproduces the issue:

  (testing "windowed-by-time with string keys"
    (let [topic-a (assoc (mock/topic "topic-a") :key-serde (Serdes/String))
          topic-b (assoc (mock/topic "topic-b") :key-serde (Serdes/String))
          driver (mock/build-driver (fn [builder]
                                      (-> builder
                                          (k/kstream topic-a)
                                          (k/group-by-key)
                                          (k/window-by-time (TimeWindows/of 1000))
                                          (k/reduce + topic-a)
                                          (k/to-kstream)
                                          (k/map (fn [[k v]] [(.key k) v]))
                                          (k/to topic-b))))
          publish (partial mock/publish driver topic-a)]

      (publish 1000 "a" 1)
      (publish 1500 "a" 2)
      (publish 5000 "a" 4)

      (let [keyvals (mock/get-keyvals driver topic-b)]
        (is (= 3 (count keyvals)))
        (is (= [0 1] (first keyvals)))
        (is (= [0 3] (second keyvals)))
        (is (= [0 4] (nth keyvals 2))))))

Releasing the library manually is painful, and potentially error prone - we should have the tooling do it automatically when a release is created in Git hub (or something like that)

KGroupedStream and KGroupedTable now offer simplified arities (that don't need to be provided a Materialized parameter).

These could be added to jackdaw.

value param not used

I dont know if its a typo or something that was intended for use in the future?

The current stable version of Kafka streams (2.3) supports naming some internal topics per KIP's.

https://cwiki.apache.org/confluence/display/KAFKA/KIP-372%3A+Naming+Repartition+Topics+for+Joins+and+Grouping
https://cwiki.apache.org/confluence/display/KAFKA/KIP+230%3A+Name+Windowing+Joins

It seems that it is possible (in 2.3) to name some internal topics now but not windowed join topics. We actually discovered that in trunk it is implemented but not released yet. https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java#L1109

Being able to name joins is important for long lived JOIN windows so that changes in the topology dont change the internal topic name and ignore the join history upon topology restart.

In the meant time we are gettin by a custom build 2.2 of kafka with support for Named Joins.
https://github.com/FundingCircle/kafka/pull/5/files#diff-5142e1d4a6410459d6bf6df98828e5afR920-R921

And a patch to join impl functions:

(defn join-windowed
  "Combines the values of two streams that share the same key using a windowed
  inner join. Adds the `join-name` parameter, which is used to name the internal
  storage topics. Requires patched version of the kafka streams jar."
  [this-kstream other-kstream value-joiner-fn windows
   {key-serde :key-serde this-value-serde :value-serde}
   {other-value-serde :value-serde}
   join-name]
  (clj-kstream
   (.join (kstream* this-kstream)
          (kstream* other-kstream)
          (value-joiner value-joiner-fn)
          windows
          (Joined/with key-serde this-value-serde other-value-serde join-name))))

(defn left-join-windowed
  "Combines the values of two streams that share the same key using a windowed
  left join. Adds the `join-name` parameter, which is used to name the internal
  storage topics. Requires patched version of the kafka streams jar."
  [this-kstream other-kstream value-joiner-fn windows
   {key-serde :key-serde this-value-serde :value-serde}
   {other-value-serde :value-serde}
   join-name]
  (clj-kstream
   (.leftJoin (kstream* this-kstream)
              (kstream* other-kstream)
              (value-joiner value-joiner-fn)
              windows
              (Joined/with key-serde this-value-serde other-value-serde join-name))))

Supporting naming of windowed joins seems quite critical as explained above but looking into supporting other internal topic custom naming support should also be looked at.

Avro has a very interesting ability to use a different schema for reading a message than the one used to writing the message. Both schemas must be made available to the reader (this is the raison d'être for the schema registry) but as long as the schemas are compatible, it means that consumers and producers are not forced to share their schemas via some centralized github repo.

At the moment jackdaw doesn't support this feature. Even though it requires a :value-serde in many consuming contexts, it does not actually use it to deserialize the message.

It should be possible to add a :reader-schema key to avro serde multi-method which allows users to take advantage of this feature without breaking any existing uses.

(ja/describe-topics-configs client [{:topic-name "connect-offsets"}]) throws a ClassCastException when invoked. I think the fix is relatively simple just have to add topics to (-> client (.describeConfigs (map #(-> % :topic-name jd/->topic-resource) topics)) .all deref vals first jd/datafy)

Tagging/creating a release does not deploy to Clojars. This looks like a problem with the credentials.

Does jackdaw provide a convenient way to use schema registry with basic auth? I think it's a good feature, considering it's how Confluent Cloud secures its Schema Registry instances.

After lots of source reading I was able to produce messages by instantiating a custom CachedSchemaRegistryClient object and passing it as :schema-registry-client, for instance:

(RestService. ^str "https://xxxxx.confluent.cloud")
(def cli (CachedSchemaRegistryClient. rs 200 (select-keys (basic-config) ["basic.auth.credentials.source" "schema.registry.basic.auth.user.info"])))
(def serde-resolver
  (partial resolver/serde-resolver :schema-registry-url (get (basic-config) "schema.registry.url")
           :schema-registry-client cli))

The API docs should be generated on merge to master.

Hello and thx for the great lib.

I have a window-by-time aggregation that works w/ reduce when this only has an adder. But, I need to also have a subtracter and this fails w/:

java.lang.AbstractMethodError: jackdaw.streams.interop.reduce

I believe that the signature of the reduce for CljTimeWindowedKStream is missing for this case.

Thx

With Clojure 1.9 or lower, this casques a problem

Caused by: java.lang.IllegalArgumentException: Unbound: #'clojure.core.protocols/Datafiable is not a protocol
 at clojure.core$extend.invokeStatic(core_deftype.clj:784)
 at clojure.core$extend.doInvoke(core_deftype.clj:746)
 at clojure.lang.RestFn.invoke(RestFn.java:439)
 at jackdaw.data__init.load(Unknown Source)
 at jackdaw.data__init.<clinit>(Unknown Source)

Need to re-work in and around

While working on #90, I noticed that k/count was mis-specced which meant that you could not fall back to the default serdes configured in the app config. Suspect there might be a few more. Would be cool to locate and fix them all.

At current, the CircleCI configuration uses the org-global context. This brings private configuration data into scope. As a matter of policy, open source projects must not make use of the same private infrastructure and secrets as private projects.

The intent of send! is to write the data but not block for the result. It uses a delay around the actual result to delay the datafy of the response when it arrives.

Unfortunately, the use of delay also wraps the send. call to kafka, so calling send! does nothing until you deref it, at which point you are blocked until the write succeeds. The use of the threading macro obfuscates things here also (for me at least ...)

Can be fixed by moving the send. call outside of the delay, so the kafka write is started and we just delaying the datafying of the eventual result, without the threading macro use so its more explicit. i.e.

(let [send-future (.send ^Producer producer ^ProducerRecord record)] 
     (delay (jd/datafy @send-future)))

I'll raise a PR tmrw,

This happens consistently for me with the word-count example:

• Open word_count_test.clj
• m-x cider-jack-in
• m-x cider-load-buffer
• m-x cider-test-run-ns-tests works (first time)
• m-x cider-test-run-ns-tests fails - every time after the first

Error in test-word-count-example
Uncaught exception, not in assertion
   error: org.apache.kafka.streams.errors.ProcessorStateException: Error opening store KSTREAM-AGGREGATE-STATE-STORE-0000000004 at location /tmp/kafka-streams/word-count/0_0/rocksdb/KSTREAM-AGGREGATE-STATE-STORE-0000000004

(Related: #114)

At the moment, the avro serde requires a key of :key? to be present in the topic config (see the code here). If the topic config doesn't include it, you get a NPE. AFAICT, this key determines whether the serde is for a key or value - so the topic config doesn't seem like the right place for it, as there can be an avro serde for both the key and the value.

Am I missing something here? Or would it be best to pass key? as a separate parameter to avro-serde?

The anchor text should describe what the link points to.

Create a test machine test for the Word Count example.

As a test-author,
I'd like to ensure a topic does not exist before re-creating the topic according to the fixture before my test runs
So that I can be sure the topic used is the one my test expects

One option for this is a :delete-first? option on the existing topic-fixture but there are a few things to consider in relation to giving the caller a chance to abort the fixture if the topics were not created to it's satisfaction within the expected time.

The avro serde resolver requires the specification of a local schema even though the schema is only actually used for producing messages. When consuming messages, we just let the KafkaAvroDeserializer lookup the schema-id in the registry so when "resolving" a topic-metadata definition we should not require the presence of a schema.

Great functionality to have for interactive development. It is also a great way to excercise the untested admin APIs like alter-topic-config! and describe-topics-configs

The generated documentation contains a SNAPSHOT version. This should correspond to the last release.

Given that the test machine is just called test we end up with some interesting file names, like

• src/jackdaw/test.clj
• test/jackdaw/test_test.clj

I'm not sure what could be a better name tbh, but I think that the whole machinery under src/test is not really just useful to write tests, it's something that can help just setting up kafka topics, so it might be worth to rename it something else.

jackdaw.serdes.avro/+base-schema-type-registry+ and
jackdaw.serdes.avro/+UUID-type-registry+
are part of the public API but this namespace has been deprecated.

Copy these to jackdaw.serdes.avro.confluent.

When doing so, omit these lines:
https://github.com/FundingCircle/jackdaw/blob/master/src/jackdaw/serdes/avro.clj#L581-L583
and update the doc-string.

At the moment only Clojure 1.10 works, so we should either fix it or at least document clearly what versions of Clojure are currently supported, to avoid surprises.

When an exception happens while reading EDN data it would be nice to know what the data is that could not be read. Can we add error handling to the EDN serde and include the string that failed to read in the ex-data of the thrown exception?

Similar to what the Avro serdedoes here:
https://github.com/FundingCircle/jackdaw/blob/master/src/jackdaw/serdes/avro.clj#L532

I can work on a PR if you are fine with this proposal.

produce! accepts key as an argument and does nothing with it.

(defn produce!
 "Helper wrapping `#'send!`.

 Builds and sends a `ProducerRecord` so you don't have to. Returns
 a future which will produce datafied record metadata when forced."
 ([producer topic value]
  (send! producer
         (jd/->ProducerRecord topic value)))
 ([producer topic key value]
  (send! producer
         (jd/->ProducerRecord topic value)))
 ([producer topic partition key value]
  (send! producer
         (jd/->ProducerRecord topic partition topic value)))
 ([producer topic partition timestamp key value]
  (send! producer
         (jd/->ProducerRecord topic partition timestamp topic value)))
 ([producer topic partition timestamp key value headers]
  (send! producer
         (jd/->ProducerRecord topic partition timestamp topic value headers))))

Documentation refers to avro-serde fn but the fn is actually called serde

Currently the "do" and "do!" commands require that the function provided by the user returns a map that is assoc'd into the result. We should just ignore whatever is returned by the user provided function and explicitly return {:result :ok} as long as the function completes.

I'm writing a utility which has to process messages within a time period and my implementation uses seek-to-timestamp. Documentation for seek-to-timestamp says the following:

After seeking, the first message read from each partition will be
the EARLIEST message whose timestamp is greater than or equal to the
timestamp sought.

This behavior is exactly what I need and what I'd expect. However, it is not what I actually get. Calling seek-to-timestamp with the current time as argument (e.g. Instant/now) positions consumer at the beginning instead of the end of partitions. Consequently, consumer reads all messages in a topic instead of none.

The following (existing) test case from client_test.clj demonstrates the problem:

   (testing "seek to ts-next=1000"
          (let [ts-next 1000]
            (as-> consumer $
              (client/assign-all $ (map :topic-name [topic-config]))
              (client/seek-to-timestamp $ ts-next [topic-config])
              (client/position-all $)
              (is (= [0]
                     (vals $))))))

Contrary to the documentation and expectation, the first message read would not be the one whose timestamp is greater than or equal to the timestamp sought, but the message with timestamp 1.

I suppose the implementation which would be more aligned with the documentation/expectation would handle nils returned by KafkaConsumer/offsetsForTimes and instead of returning position 0, return the last offset in partition. Unfortunately, trying to maintain the expectation in case of timestamps in the future is probably infeasible so that would not be covered.

I'd like to hear your thoughts on this.

quay.io/fundingcircle/clojure is private.
This repo should build from a public image.

jackdow is lacking the support to specify the suppress operator when building a time-windowed kafka streams topology.

See:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-328%3A+Ability+to+suppress+updates+for+KTables
and
https://kafka.apache.org/21/documentation/streams/developer-guide/dsl-api#id31

All other client fn's that expect a topic expect topic metadata. This fn expects a topic name string. I thing it should extract the name from the metadata as other fn's do."

As assing-all might already be used be sure to add a deprecation warning.

Function jackdaw.admin/topics-ready? seems broken (at least from 0.6.8 to the latest commit a68f3d7, probably even earlier). I think it uses return value of describe-topics* without datafying it and it throws exception.

Attempting to use reset-application-fixture or, by extension, integration-fixture with a Kafka bootstrap server other than localhost:9092 results in a connection timeout. Examining the source, it looks like this value has been hardcoded as an argument to StreamsResetter.

Hi! I'm trying to use various jackdaw serdes, but unfortunately I'm not managing to get avro serdes to work. The following exception is thrown:

Execution error (IllegalArgumentException) at jackdaw.serdes.avro/eval4495$fn$G (avro.clj:123).
No implementation of method: :clj->avro of protocol: #'jackdaw.serdes.avro/SchemaCoercion found for class: nil

I'm using clojure 1.10.0 and jackdaw 0.6.6. Kafka clusters/schema registry is managed by Confluent Cloud. Using Confluent Platform 5.2.1. Key/value schemas are already in registry. I'd really appreciate any help or suggestion! This code is mostly the same as examples/serdes. String and JSON serdes are working fine.

(ns trying-jackdaw.core
  (:require [clojure.algo.generic.functor :refer [fmap]]
            [clojure.java.shell :refer [sh]]
            [jackdaw.client :as jc]
            [jackdaw.client.log :as jcl]
            [jackdaw.admin :as ja]
            [jackdaw.serdes.edn :as jse]
            [jackdaw.streams :as j]
            [jackdaw.serdes]
            [jackdaw.serdes.json :as json]
            [jackdaw.serdes.avro.confluent :as jsac]
            [jackdaw.serdes.edn :as jse]
            [jackdaw.serdes.resolver :as resolver]))

;; of course not pasting my keys here
#_(defn basic-config
  []
  {"ssl.endpoint.identification.algorithm" "https"
   "sasl.mechanism"                        "PLAIN"
   "request.timeout.ms"                    (int 20000)
   "bootstrap.servers"                     "xxxxx.confluent.cloud:9092"
   "retry.backoff.ms"                      (int 500)
   "sasl.jaas.config"                      (str "org.apache.kafka.common.security.plain.PlainLoginModule required username=\""
                                                ccloud-key "\" password=\"" ccloud-secret "\";")
   "security.protocol"                     "SASL_SSL"

   ; Schema Registry specific settings
   "basic.auth.credentials.source"         "USER_INFO"
   "schema.registry.basic.auth.user.info"  (str sr-key ":" sr-secret)
   "schema.registry.url"                   "https://xxxxxxx.confluent.cloud"

   ; Enable Avro serializer with Schema Registry
   "key.serializer"                        "io.confluent.kafka.serializers.KafkaAvroSerializer"
   "value.serializer"                      "io.confluent.kafka.serializers.KafkaAvroSerializer"
   })
(defn basic-config
  "Returns the application config."
  []
  {"application.id" "serdes"
   "bootstrap.servers" "localhost:9092"
   "cache.max.bytes.buffering" "0"})

(defn producer-config [] (basic-config))
(defn admin-config [] (basic-config))

;;; ------------------------------------------------------------
;;;
;;; Configure topics
;;;

(def +topic-metadata+
  {"example_avro"
                  {:topic-name  "example_avro"
                   :key-serde   {:serde-keyword   :jackdaw.serdes.avro.confluent/serde
                                 :schema-filename "key_schema.json"
                                 :key?            true}
                   :value-serde {:serde-keyword   :jackdaw.serdes.avro.confluent/serde
                                 :schema-filename "value_schema.json"
                                 :key?            false}}
   "example_json" {
                   :topic-name  "example_json"
                   :key-serde   (json/serde)
                   :value-serde (json/serde)
                   }
   "example_str"
                  {:topic-name  "example_str"
                   :key-serde   {:serde-keyword :jackdaw.serdes/string-serde}
                   :value-serde {:serde-keyword :jackdaw.serdes/string-serde}}})

#_(def serde-resolver
  (partial resolver/serde-resolver :schema-registry-url (get (basic-config) "schema.registry.url")))
(def serde-resolver
  (partial resolver/serde-resolver :schema-registry-url "http://localhost:8081"))

(def topic-metadata
  (memoize (fn []
             (fmap #(if (= (:topic-name %) "example_json")
                      %
                      (assoc % :key-serde ((serde-resolver) (:key-serde %))
                               :value-serde ((serde-resolver) (:value-serde %))))
                   +topic-metadata+))))

(defn publish
  "Takes a topic config and record value, and (optionally) a key and
  parition number, and produces to a Kafka topic."
  ([topic-config value]
   (with-open [client (jc/producer (producer-config) topic-config)]
     @(jc/produce! client topic-config value))
   nil)

  ([topic-config key value]
   (with-open [client (jc/producer (producer-config) topic-config)]
     @(jc/produce! client topic-config key value))
   nil)

  ([topic-config partition key value]
   (with-open [client (jc/producer (producer-config) topic-config)]
     @(jc/produce! client topic-config partition key value))
   nil))

(def example-key {"timestamp" 1562266979037
                  "name"      "example-name"})

(def example-value {"int_field"    1
                    "string_field" ""})

(defn -main
  [& args]
  (publish (get (topic-metadata) "example_str") "example-key" "example-value")
  (publish (get (topic-metadata) "example_json") example-key example-value)
  (publish (get (topic-metadata) "example_avro") example-key example-value))

resources/key_schema.json

{
  "type" : "record",
  "name" : "key_example",
  "namespace" : "avro",
  "fields" : [
    {
      "name" : "timestamp",
      "type" : "long"
    },
    {
      "name" : "name",
      "type" : "string"
    }
  ]
}

resources/value_schema.json

{
  "type" : "record",
  "name" : "value_example",
  "namespace" : "avro",
  "fields" : [
    {
      "name" : "int_field",
      "type" : "int"
    },
    {
      "name" : "string_field",
      "type" : "string"
    }
  ]
}

Update: Using Confluent Platform, since issue doesn't depend on Confluent Cloud.

Right now examples depend on non-released versions of Jackdaw. Needs definition.

It hides a lot of output in the logs (well, its not strictly hidden but its not "in your face" when there are errors). It also uses stack trace pretty formatting which may not work with the 1.10 errors - and also means control sequences appear in the logs making the errors harder to read.