kafka-metamorphosis.producer — org.clojars.caioclavico/kafka-metamorphosis 0.2.0

Kafka producer functions with idiomatic Clojure interface

close! clj

(close! producer)

(close! producer timeout-ms)

Close the producer and release resources. Optionally specify timeout in milliseconds.

create clj

(create config)

Create a Kafka producer with the given configuration map.

Example config: {:bootstrap-servers "localhost:9092" :key-serializer "org.apache.kafka.common.serialization.StringSerializer" :value-serializer "org.apache.kafka.common.serialization.StringSerializer" :acks "all" :retries 3}

flush! clj

(flush! producer)

Flush any buffered records to the Kafka cluster. This method makes all buffered records immediately available to send.

send! clj

(send! producer topic value)

(send! producer topic key value)

Send a message to a Kafka topic synchronously. Returns RecordMetadata on success.

Usage: (send! producer "my-topic" "key" "value") (send! producer "my-topic" nil "value-only") ; key can be nil

send-async! clj

(send-async! producer topic value)

(send-async! producer topic key value)

(send-async! producer topic key value callback-fn)

Send a message to a Kafka topic asynchronously with optional callback. Returns a Future.

Callback function receives [metadata exception] where:

• metadata: RecordMetadata if successful, nil if error
• exception: Exception if error occurred, nil if successful

Usage: (send-async! producer "my-topic" "key" "value") (send-async! producer "my-topic" "key" "value" (fn [metadata exception] (if exception (println "Error:" (.getMessage exception)) (println "Sent to partition:" (.partition metadata)))))