Awesome

Ziggurat

Description

Ziggurat is a framework built to simplify stream processing on Kafka. It can be used to create a full-fledged Clojure app that reads and processes messages from Kafka. Ziggurat abstracts the following features:

Reading messages from Kafka
Retrying failed messages via RabbitMQ
Setting up an HTTP server

Refer to concepts to understand the concepts referred to in this document.

Important Concepts and Usage Docs

Ziggurat HTTP Server
Toggle Streams on a Running Actor
Middlewares in Ziggurat
Consuming and Publishing Messages to Kafka
Connecting RabbitMQ and using channels
Configuration and Config Description
To read about all concepts, please refer the Concepts file

Dev Setup

For Mac Users Only

Install Clojure: brew install clojure
Install Leiningen: brew install leiningen
Run docker-compose: docker-compose up. This starts:

Kafka on localhost:9092
ZooKeeper on localhost:2181
RabbitMQ on localhost:5672

Run tests: make test

Running a Cluster Setup Locally

Run make setup-cluster. This clears up the volume and starts:
- 3 Kafka brokers on localhost:9091, localhost:9092, and localhost:9093
- Zookeeper on localhost:2181
- RabbitMQ on localhost:5672

Running Tests via a Cluster

Run make test-cluster. This uses config.test.cluster.edn instead of config.test.edn.

Usage

Add this to your project.clj:

[tech.gojek/ziggurat "4.11.1"]


_Please refer [clojars](https://clojars.org/tech.gojek/ziggurat) for the latest stable version_

To start a stream (a thread that reads messages from Kafka), add this to your core namespace.

```clojure
(require '[ziggurat.init :as ziggurat])

(defn start-fn []
    ;; your logic that runs at startup goes here
)

(defn stop-fn []
    ;; your logic that runs at shutdown goes here
)

(defn main-fn
  [{:keys [message metadata] :as message-payload}]
    (println message)
    :success)

(def handler-fn
    (-> main-fn
      (middleware/protobuf->hash ProtoClass :stream-id)))
;; Here ProtoClass refers to the fully qualified name of the Java class which the code is used to de-serialize the message.

(ziggurat/main start-fn stop-fn {:stream-id {:handler-fn handler-fn}})

NOTE: this example assumes that the message is serialized in Protobuf format

Please refer the Middleware section for understanding handler-fn here.

The main-fn is the function that will be applied to every message that is read from the Kafka stream.
The main-fn will take map as an argument that takes 2 keys i.e
- message - It is the byte[] array received from kafka.
- metadata
  - topic - It is the topic from where kafka message is consumed.
  - timestamp - It is ingestion timestamp in kafka.
  - partition - The partition from message is consumed.
  - rabbitmq-retry-count - The number of retries done by rabbitmq for given message.
The main-fn returns a keyword which can be any of the below words
- :success - The message was successfully processed and the stream should continue to the next message
- :retry - The message failed to be processed and it should be retried via RabbitMQ.
- :dead-letter - The message is not retried and is directly pushed to the dead letter queue
- :skip - The message should be skipped without reporting its failure or retrying the message. Same as :success except that a different metric is published to track skipped messages
The start-fn is run at the application startup and can be used to initialize connection to databases, http clients, thread-pools, etc.
The stop-fn is run at shutdown and facilitates graceful shutdown, for example, releasing db connections, shutting down http servers etc.

Multiple stream routes

Ziggurat enables reading from multiple streams and applying same/different functions to the messages. :stream-id is a unique identifier per stream which needs to be included in config.edn file
All configs, queues and metrics will be namespaced under this id.

(ziggurat/main start-fn stop-fn {:stream-id-1 {:handler-fn handler-fn-1}
                                         :stream-id-2 {:handler-fn handler-fn-2}})

(require '[ziggurat.init :as ziggurat])

(defn start-fn []
    ;; your logic that runs at startup goes here
)

(defn stop-fn []
    ;; your logic that runs at shutdown goes here
)


(defn api-handler [_request]
  {:status  200
   :headers {"Content-Type" "application/json"}
   :body    (get-resource)})

(def routes [["v1/resources" {:get api-handler}]])

(defn main-fn
  [{:keys [message metadata] :as message-payload}]
    (println message)
    :success)

(def handler-fn
    (-> main-fn
      (middleware/protobuf->hash ProtoClass :stream-id)))

(ziggurat/main start-fn stop-fn {:stream-id {:handler-fn handler-fn}} routes)