kafka-metamorphosis.core — org.clojars.caioclavico/kafka-metamorphosis 0.4.0

Kafka Metamorphosis - A Clojure wrapper for Apache Kafka

'When Gregor Samsa woke up one morning from unsettling dreams, he found himself changed in his bed into a monstrous vermin.' — Franz Kafka, The Metamorphosis

Just as Gregor Samsa underwent his transformation, this library metamorphoses the complex Java Kafka APIs into elegant Clojure functions.

-main clj

(-main & args)

Entry point for the Kafka Metamorphosis library. Supports different commands and demonstrates library usage.

admin-config clj

(admin-config)

(admin-config bootstrap-servers)

Create an admin client configuration.

Usage: (admin-config) ; Local broker (admin-config "kafka:9092") ; Custom broker

build-config clj

(build-config overrides serializers-or-deserializers)

(build-config bootstrap-servers overrides serializers-or-deserializers)

Generic configuration builder that can be used for producers, consumers, and admin clients.

Usage: (build-config {:group-id "my-group"} string-deserializers) ; Consumer config (build-config {:acks "all"} json-serializers) ; Producer config
(build-config {} {}) ; Admin config (build-config "kafka:9092" {:group-id "group"} string-deserializers) ; Custom broker

consume-json-messages! clj

(consume-json-messages! group-id topics)

(consume-json-messages! group-id topics opts)

Consume JSON messages from topics with minimal setup. Returns a lazy sequence of messages with values automatically parsed from JSON.

Usage: (consume-json-messages "my-group" ["topic1" "topic2"]) (consume-json-messages "my-group" ["topic1"] {:max-messages 10}) (consume-json-messages "my-group" ["topic1"] {:timeout-ms 5000 :max-messages 100})

consume-messages! clj

(consume-messages! group-id topics)

(consume-messages! group-id topics opts)

Consume messages from topics with minimal setup. Returns a lazy sequence of messages.

Usage: (consume-messages "my-group" ["topic1" "topic2"]) (consume-messages "my-group" ["topic1"] {:max-messages 10}) (consume-messages "my-group" ["topic1"] {:timeout-ms 5000 :max-messages 100})

consumer-config clj

(consumer-config group-id)

(consumer-config group-id overrides)

(consumer-config bootstrap-servers group-id overrides)

(consumer-config bootstrap-servers group-id overrides deserializers)

Create a consumer configuration with sensible defaults.

Usage: (consumer-config "my-group") ; String deserializers, local broker (consumer-config "my-group" {:auto-offset-reset "latest"}) ; Override settings (consumer-config "kafka:9092" "my-group" {}) ; Custom broker + group (consumer-config "my-group" {} string-deserializers) ; Custom deserializers (consumer-config "kafka:9092" "my-group" {} json-deserializers) ; JSON deserializers

default-bootstrap-servers clj

Default Kafka broker address for local development

health-check clj

(health-check)

(health-check bootstrap-servers)

Check if Kafka is accessible and return cluster information.

Usage: (health-check) (health-check "kafka:9092")

json-consumer-config clj

(json-consumer-config group-id)

(json-consumer-config group-id overrides)

(json-consumer-config bootstrap-servers group-id overrides)

Create a consumer configuration for JSON values with sensible defaults. Uses simple JSON mode (no schema registry required).

Usage: (json-consumer-config "my-group") ; String keys, JSON values, local broker (json-consumer-config "my-group" {:auto-offset-reset "latest"}) ; Override settings (json-consumer-config "kafka:9092" "my-group" {}) ; Custom broker + group

json-producer-config clj

(json-producer-config)

(json-producer-config overrides)

(json-producer-config bootstrap-servers overrides)

Create a producer configuration for JSON values with sensible defaults. Uses simple JSON mode (no schema registry required).

Usage: (json-producer-config) ; String keys, JSON values, local broker (json-producer-config {:acks "1"}) ; Override specific settings (json-producer-config "kafka:9092" {:acks "all"}) ; Custom broker + settings

local-config clj

(local-config)

(local-config bootstrap-servers)

Create a base configuration for local Kafka development.

Usage: (local-config) ; Use default localhost:9092 (local-config "kafka:9092") ; Custom broker address

producer-config clj

(producer-config)

(producer-config overrides)

(producer-config bootstrap-servers overrides)

(producer-config bootstrap-servers overrides serializers)

Create a producer configuration with sensible defaults.

Usage: (producer-config) ; String serializers, local broker (producer-config {:acks "1"}) ; Override specific settings (producer-config "kafka:9092" {:acks "all"}) ; Custom broker + settings (producer-config {} string-serializers) ; Custom serializers (producer-config "kafka:9092" {} json-serializers) ; JSON serializers

send-json-message! clj

(send-json-message! topic value)

(send-json-message! topic key value)

(send-json-message! topic key value producer-opts)

Send a JSON message to a topic with minimal setup. The value will be automatically converted to JSON.

Usage: (send-json-message! "my-topic" {:name "John" :age 30}) (send-json-message! "my-topic" "user-123" {:name "John" :age 30}) (send-json-message! "my-topic" "user-123" {:name "John"} {:acks "1"})

send-message! clj

(send-message! topic value)

(send-message! topic key value)

(send-message! topic key value producer-opts)

Send a message to a topic with minimal setup.

Usage: (send-message! "my-topic" "Hello, World!") (send-message! "my-topic" "key" "Hello, World!") (send-message! "my-topic" "key" "Hello, World!" {:acks "1"})