ziggurat.producer — tech.gojek/ziggurat 3.4.2-alpha.1

This namespace defines methods for publishing data to Kafka topics. The methods defined here are essentially wrapper around variants of send methods defined in org.apache.kafka.clients.producer.KafkaProducer.

At the time of initialization, an instance of org.apache.kafka.clients.producer.KafkaProducer is constructed using config values provided in resources/config.edn.

A producer can be configured for each of the stream-routes in config.edn. Please see the example below

  :stream-router {:default {:application-id "test"

                            :bootstrap-servers    "localhost:9092"

                            :stream-threads-count [1 :int]

                            :origin-topic         "topic"

                            :channels             {:channel-1 {:worker-count [10 :int]

                                                               :retry {:count [5 :int]

                                                                       :enabled [true :bool]}}}

                            :producer             {:bootstrap-servers "localhost:9092"

                                                   :acks "all"

                                                   :retries-config  5

                                                   :max-in-flight-requests-per-connection 5

                                                   :enable-idempotence  false

                                                   :value-serializer  "org.apache.kafka.common.serialization.StringSerializer"

                                                   :key-serializer    "org.apache.kafka.common.serialization.StringSerializer"}}

Please see the documentation for send for publishing data via Kafka producers

These are the KafkaProducer configs currenlty supported in Ziggurat.

• bootstrap.servers
• acks
• retries
• key.serializer
• value.serializer
• max.in.flight.requests.per.connection
• enable.idempotencecd

Please see producer configs for a complete list of all producer configs available in Kafka.

*implements-serializer?* clj

(*implements-serializer?* serializer-class)

-send clj

(-send stream-config-key topic key value)

(-send stream-config-key topic partition key value)

explain-str clj

implements-serializer? clj

kafka-producers clj

producer-properties clj

(producer-properties config-map)

producer-properties-map clj

(producer-properties-map)

ProducerConfigSchema clj

property->fn clj

(property->fn field-name)

send clj

(send stream-config-key topic key value)

(send stream-config-key topic partition key value)

A wrapper around org.apache.kafka.clients.producer.KafkaProducer#send which enables the users of Ziggurat to produce data to a Kafka topic using a Kafka producer associated with a Kafka stream config key.

E.g. For publishing data to producer defined for the stream router config with defined agains key :default, use send like this.

(send :default "test-topic" "key" "value") (send :default "test-topic" 1 "key" "value")

valid-configs? clj