Kafka Konsumer
Description
Kafka Konsumer provides an easy implementation of Kafka consumer with a built-in retry/exception manager (kafka-cronsumer).
Migration Guide
V2 Release Notes
- Added ability for manipulating kafka message headers.
- Added transactional retry feature. Set false if you want to use exception/retry strategy to only failed messages.
- Enable manuel commit at both single and batch consuming modes.
- Enabling consumer resume/pause functionality. Please refer to its example and
- Bumped kafka-cronsumer to the latest version:
- Reduce memory allocation.
- Increase TP on changing internal concurrency structure.
How to migrate from v1 to v2?
You can get latest version via go get github.com/Trendyol/kafka-konsumer/v2@latest
- You need to change import path from
github.com/Trendyol/kafka-konsumertogithub.com/Trendyol/kafka-konsumer/v2
- You need to change your consume function with pointer signature.
- We moved messageGroupDuration from
batchConfiguration.messageGroupDurationto root level. Because this field is used
Installation
go get github.com/Trendyol/kafka-konsumer/v2@latestExamples
You can find a number of ready-to-run examples at this directory.
After running docker-compose up command, you can run any application you want.
Simple Consumer
func main() { consumerCfg := &kafka.ConsumerConfig{ Reader: kafka.ReaderConfig{ Brokers: []string{"localhost:29092"}, Topic: "standart-topic", GroupID: "standart-cg", }, ConsumeFn: consumeFn, RetryEnabled: false, } consumer, _ := kafka.NewConsumer(consumerCfg) defer consumer.Stop() consumer.Consume() } func consumeFn(message kafka.Message) error { fmt.Printf("Message From %s with value %s", message.Topic, string(message.Value)) return nil }
Simple Consumer With Retry/Exception Option
func main() { consumerCfg := &kafka.ConsumerConfig{ Reader: kafka.ReaderConfig{ Brokers: []string{"localhost:29092"}, Topic: "standart-topic", GroupID: "standart-cg", }, RetryEnabled: true, RetryConfiguration: kafka.RetryConfiguration{ Topic: "retry-topic", StartTimeCron: "/1 *", WorkDuration: 50 * time.Second, MaxRetry: 3, // Applied to retry-topic and dead-letter-topic producers. ProducerCompression: kafka.Gzip, }, ConsumeFn: consumeFn, } consumer, _ := kafka.NewConsumer(consumerCfg) defer consumer.Stop() consumer.Consume() } func consumeFn(message kafka.Message) error { fmt.Printf("Message From %s with value %s", message.Topic, string(message.Value)) return nil }
With Batch Option
func main() { consumerCfg := &kafka.ConsumerConfig{ Reader: kafka.ReaderConfig{ Brokers: []string{"localhost:29092"}, Topic: "standart-topic", GroupID: "standart-cg", }, LogLevel: kafka.LogLevelDebug, RetryEnabled: true, RetryConfiguration: kafka.RetryConfiguration{ Brokers: []string{"localhost:29092"}, Topic: "retry-topic", StartTimeCron: "/1 *", WorkDuration: 50 * time.Second, MaxRetry: 3, }, MessageGroupDuration: time.Second, BatchConfiguration: kafka.BatchConfiguration{ MessageGroupLimit: 1000, BatchConsumeFn: batchConsumeFn, }, } consumer, _ := kafka.NewConsumer(consumerCfg) defer consumer.Stop() consumer.Consume() } func batchConsumeFn(messages []kafka.Message) error { fmt.Printf("%d\n comes first %s", len(messages), messages[0].Value) return nil }
With Disabling Transactional Retry
func main() { consumerCfg := &kafka.ConsumerConfig{ Reader: kafka.ReaderConfig{ Brokers: []string{"localhost:29092"}, Topic: "standart-topic", GroupID: "standart-cg", }, LogLevel: kafka.LogLevelDebug, RetryEnabled: true, TransactionalRetry: kafka.NewBoolPtr(false), RetryConfiguration: kafka.RetryConfiguration{ Brokers: []string{"localhost:29092"}, Topic: "retry-topic", StartTimeCron: "/1 *", WorkDuration: 50 * time.Second, MaxRetry: 3, }, MessageGroupDuration: time.Second, BatchConfiguration: kafka.BatchConfiguration{ MessageGroupLimit: 1000, BatchConsumeFn: batchConsumeFn, }, } consumer, _ := kafka.NewConsumer(consumerCfg) defer consumer.Stop() consumer.Consume() } func batchConsumeFn(messages []kafka.Message) error { // you can add custom error handling here & flag messages for i := range messages { if i%2 == 0 { messages[i].IsFailed = true } } // you must return err here to retry failed messages return errors.New("err") }
With Producer Interceptor
Please refer to Producer Interceptor Example
With Distributed Tracing Support
Please refer to Tracing Example
With Pause & Resume Consumer
Please refer to Pause Resume Example
With Grafana & Prometheus
In this example, we are demonstrating how to create Grafana dashboard and how to define alerts in Prometheus. You can
see the example by going to the with-grafana folder in the examples folder
and running the infrastructure with docker compose up and then the application.
With SASL-PLAINTEXT Authentication
Under the examples - with-sasl-plaintext folder, you can find an example
of a consumer integration with SASL/PLAIN mechanism. To try the example, you can run the command docker compose up
under the specified folder and then start the application.
With Send Direct To Dead Letter
This feature lets you send a message directly to a dead-letter topic by setting message.SendDirectToDeadLetter = true inside your ConsumeFn (or selectively for items in BatchConsumeFn). When this flag is set and your ConsumeFn returns an error, the message is produced to the configured dead-letter topic with an x-error-message header.
- Dead letter topic resolution order: if
consumer.DeadLetterTopicis set, it is used; otherwiseretryConfiguration.deadLetterTopicis used. - You can set
message.ErrDescriptionto override the error header value written asx-error-message. - The flag is also honored while consuming from the retry topic: the message is produced to the dead-letter topic once and is not retried again. If no dead-letter topic is configured, the message follows the normal retry flow.
- Set
DeadLetterProducerBatchBytesto split direct dead-letter ProduceBatch calls by approximate uncompressed message bytes. The default0keeps the current single-batch behavior. Configure this below the Kafka/topic max message size to leave room for Kafka protocol and request overhead. A single message larger than the limit is still sent alone and can still fail with KafkaMessage Size Too Large; adjust DLQ topic/broker limits or payload handling for that case.
Configurations
| config | description | default |
|--------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------------------|
| reader | Describes all segmentio kafka reader configurations | |
| consumeFn | Kafka consumer function, if retry enabled it, is also used to consume retriable messages | |
| skipMessageByHeaderFn | Function to filter messages based on headers, return true if you want to skip the message | nil |
| logLevel | Describes log level; valid options are debug, info, warn, and error | info |
| concurrency | Number of goroutines used at listeners | 1 |
| retryEnabled | Retry/Exception consumer is working or not | false |
| transactionalRetry | Set false if you want to use exception/retry strategy to only failed messages | true |
| deadLetterTopic | Dead-letter topic name to produce messages when message.SendDirectToDeadLetter is true. If empty, retryConfiguration.deadLetterTopic will be used. | "" |
| deadLetterProducerBatchBytes | Optional byte limit for direct dead-letter producer batches, based on approximate uncompressed message size. Zero disables chunking and preserves current behavior. Set below Kafka/topic max message size to leave protocol overhead room. Works with producer compression, but does not replace proper Kafka topic/broker message-size limits. | 0 |
| commitInterval | indicates the interval at which offsets are committed to the broker. | 1s |
| rack | see doc | |
| clientId | see doc | |
| messageGroupDuration | Maximum time to wait for a batch | 1s |
| metricPrefix | MetricPrefix is used for prometheus fq name prefix. If not provided, default metric prefix value is kafka_konsumer. Currently, there are two exposed prometheus metrics. processed_messages_total and unprocessed_messages_total So, if default metric prefix used, metrics names are kafka_konsumer_processed_messages_total_current and kafka_konsumer_unprocessed_messages_total_current. | kafka_konsumer |
| dial.Timeout | see doc | no timeout |
| dial.KeepAlive | see doc | not enabled |
| transport.DialTimeout | see doc | 5s |
| transport.IdleTimeout | see doc | 30s |
| transport.MetadataTTL | see doc | 6s |
| transport.MetadataTopics | see doc | all topics in cluster |
| distributedTracingEnabled | indicates open telemetry support on/off for consume and produce operations. | false |
| distributedTracingConfiguration.TracerProvider | see doc | otel.GetTracerProvider() |
| distributedTracingConfiguration.Propagator | see doc | otel.GetTextMapPropagator() |
| retryConfiguration.clientId | see doc | |
| retryConfiguration.startTimeCron | Cron expression when retry consumer (kafka-cronsumer) starts to work at | |
| retryConfiguration.metricPrefix | MetricPrefix is used for prometheus fq name prefix. If not provided, default metric prefix value is kafka_cronsumer. Currently, there are two exposed prometheus metrics. retried_messages_total_current and discarded_messages_total_current. So, if default metric prefix used, metrics names are kafka_cronsumer_retried_messages_total_current and kafka_cronsumer_discarded_messages_total_current | kafka_cronsumer |
| retryConfiguration.workDuration | Work duration exception consumer actively consuming messages | |
| retryConfiguration.queueCapacity | see doc | 100 |
| retryConfiguration.producerBatchSize | see doc | 100 |
| retryConfiguration.producerBatchBytes | Optional byte limit for retry producer batches, based on approximate uncompressed message size. Zero disables retry producer chunking and preserves current behavior. Requires kafka-cronsumer v1.6.11+. | 0 |
| retryConfiguration.producerBatchTimeout | see doc | 100 |
| retryConfiguration.producerRequiredAcks | see doc. Applied to the retry (kafka-cronsumer) and dead-letter producers (RequireNone / RequireOne / RequireAll). When unset, defaults to RequireNone. | RequireNone (0) |
| retryConfiguration.producerCompression | see doc. Applied to both retry-topic and dead-letter-topic producers. Useful when produced retry or dead-letter messages are large and brokers enforce a maximum message size. Supported values include kafka.Gzip, kafka.Snappy, kafka.Lz4, and kafka.Zstd. | uncompressed |
| retryConfiguration.topic | Retry/Exception topic names | |
| retryConfiguration.startOffset | Same as ReaderConfig.StartOffset: kafka.FirstOffset, kafka.LastOffset, or 0. Zero defaults to FirstOffset. Does not follow the main reader start offset. | 0 (FirstOffset) |
| retryConfiguration.deadLetterTopic | Dead-letter topic name used by the retry/exception pipeline; also used as a fallback for direct dead-lettering when top-level deadLetterTopic is empty. | |
| retryConfiguration.brokers | Retry topic brokers urls | |
| retryConfiguration.maxRetry | Maximum retry value for attempting to retry a message | 3 |
| retryConfiguration.concurrency | Number of goroutines used at listeners | 1 |
| retryConfiguration.tls.rootCAPath | see doc | "" |
| retryConfiguration.tls.intermediateCAPath | Same with rootCA, if you want to specify two rootca you can use it with rootCAPath | "" |
| retryConfiguration.sasl.authType | SCRAM or PLAIN | |
| retryConfiguration.sasl.username | SCRAM OR PLAIN username | |
| retryConfiguration.sasl.password | SCRAM OR PLAIN password | |
| retryConfiguration.backOffStrategyName | Backoff strategy used by the retry consumer. Valid values (case-sensitive): fixed, linear, exponential. If empty, defaults to fixed. An invalid value causes a startup error. | fixed |
| retryConfiguration.skipMessageByHeaderFn | Function to filter messages based on headers, return true if you want to skip the message | nil |
| ~~retryConfiguration.verifyTopicOnStartup~~ | it checks existence of the given retry topic on the kafka cluster. This field now functions as true and will be removed in future Cronsumer versions. | true |
| batchConfiguration.messageGroupLimit | Maximum number of messages in a batch | 100 |
| batchConfiguration.messageGroupByteSizeLimit | Maximum number of bytes in a batch; a single message larger than this limit is processed in its own batch | |
| batchConfiguration.batchConsumeFn | Kafka batch consumer function, if retry enabled it, is also used to consume retriable messages | |
| batchConfiguration.preBatchFn | This function enable for transforming messages before batch consuming starts. If retry enabled, it is also called for each retried message (one message at a time, concurrently with the main flow), so it must be idempotent and safe for concurrent use | |
| batchConfiguration.balancer | see doc | leastBytes |
| tls.rootCAPath | see doc | "" |
| tls.intermediateCAPath | Same with rootCA, if you want to specify two rootca you can use it with rootCAPath | "" |
| sasl.authType | SCRAM or PLAIN | |
| sasl.username | SCRAM OR PLAIN username | |
| sasl.password | SCRAM OR PLAIN password | |
| logger | If you want to custom logger | info |
| apiEnabled | Enabled metrics | false |
| apiConfiguration.port | Set API port | 8090 |
| apiConfiguration.healtCheckPath | Set Health check path | healthcheck |
| metricConfiguration.path | Set metric endpoint path | /metrics |
Monitoring
Kafka Konsumer offers an API that handles exposing several metrics.
Exposed Metrics
| Metric Name | Description | Value Type | |------------------------------------------------------------------|------------------------------------------------|------------| | kafka_konsumer_processed_messages_total_current | Total number of processed messages. | Counter | | kafka_konsumer_unprocessed_messages_total_current | Total number of unprocessed messages. | Counter | | kafka_konsumer_error_count_during_fetching_message_total_current | Total number of error during fetching message. | Counter |