dafka-consumer: A Java repository from sarialalem1

Dockerized kafka consumer

Overview

Dafka-consumer is a dockerized Kafka consumer used to abstract consuming messages from a kafka topic.

Using dafka-consumer, consuming messages is as simple as getting a POST request to your service, with the body of the request being the kafka message.

Motivation

Why use this over just a Kafka client?

Abstracts away the messaging layer, could be replaced with RabbitMQ or any other consumer.
Separates configuration, everything that's related to Kafka is encapsulated in Dafka and not the service itself.
When testing your service you only test your service's logic and not the messaging layer implementation details.

Usage & Examples

docker-compose

version: '3.9'

services:
    consumer:
        image: osskit/dafka-consumer
        ports:
            - 4001:4001
        environment:
            - KAFKA_BROKER=kafka:9092
            - TARGET_BASE_URL=http://target:2000/
            - TOPICS_ROUTES=foo:consume,bar:consume,^([^.]+).bar:consume
            - DEAD_LETTER_TOPIC=dead-letter
            - GROUP_ID=consumer_1
            - MONITORING_SERVER_PORT=4001

In joint with dafka-producer:

version: '3.9'

services:
    consumer:
        image: osskit/dafka-consumer
        ports:
            - 4001:4001
        environment:
            - KAFKA_BROKER=kafka:9092
            - TARGET_BASE_URL=http://target:2000/
            - TOPICS_ROUTES=foo:consume,bar:consume,^([^.]+).bar:consume
            - DEAD_LETTER_TOPIC=dead-letter # optional
            - GROUP_ID=consumer_1

    producer:
        image: osskit/dafka-producer
        ports:
            - 6000:6000
        environment:
            - PORT=6000
            - KAFKA_BROKER=kafka:9092

Kubernetes

You can use the provided Helm Chart, this gives you a Deployment separated from your service's Pod.

It's also possible to use this as a Sidecar.

Parameters

Container images are configured using parameters passed at runtime.

Parameter	Default Values	Description
`KAFKA_BROKER`	`required`	URL for the Kafka Broker
`TARGET_BASE_URL`	`required`	The target's HTTP POST endpoint
`GROUP_ID`	`required`	A unique id for the consumer group
`TOPICS_ROUTES`	`required`	A map between topics and their endpoint routes (e.g `topic:/consume`)
`TARGET_TIMEOUT_MS`	`298000`	Timeout for the target's response. Must be lower then RETRY_POLICY_MAX_DURATION_MS
`RETRY_POLICY_MAX_DURATION_MS`	`299000`	Maximum duration of all retry attempts. Must be lower then KAFKA_POLL_INTERVAL_MS
`KAFKA_POLL_INTERVAL_MS`	`300000` (5 min)	The maximum delay between invocations of poll() when using consumer group management. See max.poll.interval.ms for more details.
`POLL_TIMEOUT`	`1000`	Description of POLL_TIMEOUT
`MAX_POLL_RECORDS`	`50`	Number of records to process in a single batch
`SESSION_TIMEOUT`	`10000`	Description of SESSION_TIMEOUT
`RETRY_PROCESS_WHEN_STATUS_CODE_MATCH`	`5[0-9][0-9]`	Retry to process the record if the returning status code matches the regex
`PRODUCE_TO_DEAD_LETTER_TOPIC_WHEN_STATUS_CODE_MATCH`	`^(?!2\\d\\d$)\\d{3}$`	Produce to dead letter topic when matching status code regex
`RETRY_POLICY_EXPONENTIAL_BACKOFF`	`50,5000,10`	A list that represents the `[delay, maxDelay, delayFactor]` in retrying message processing
`DEAD_LETTER_TOPIC`	`null`	Dead letter topic name
`MONITORING_SERVER_PORT`	`0`	Consumer monitoring and healthcheck service port
`TARGET_HEALTHCHECK`	`null`	Target's healthcheck endpoint to verify it's alive
`USE_SASL_AUTH=false`	`false`	use SASL authentication
`SASL_USERNAME`	`required` if `USE_SASL_AUTH=true`	SASL username to authenticate
`SASL_PASSWORD`	`required` if `USE_SASL_AUTH=true`	SASL password to authenticate
`TRUSTSTORE_FILE_PATH`	`null`	Truststore certificate file path
`TRUSTSTORE_PASSWORD`	`required` if `TRUSTSTORE_FILE_PATH != null`	Truststore's password
`USE_PROMETHEUS`	`false`	Export metrics to Prometheus
`PROMETHEUS_BUCKETS`	`0.003,0.03,0.1,0.3,1.5,10`	A list of Prometheus buckets to use

License

MIT License