cronsumer

Kafka C[r]onsumer

Description

Kafka Cronsumer is mainly used for retry/exception strategy management. It works based on cron expression and consumes messages in a timely manner with the power of auto pause and concurrency.

For details check our blog post

If you need a whole consumer lifecycle with exception management, check Kafka Konsumer

How Kafka Cronsumer Works

When to use it?

• Iteration-based back-off strategies are applicable
• Messages could be processed in an eventually consistent state
• Max retry exceeded messages could be ignored and send to dead letter topic
• To increase consumer resiliency
• To increase consumer performance with concurrency

When to avoid?

• Messages should be processed in order
• Messages should be certainly processed (we discard messages if max retry is exceeded)
• Messages should be committed (we use auto-commit interval for increasing performance)
• Messages with TTL (Time to Live)

Guide

Installation

go get github.com/Trendyol/kafka-cronsumer@latest

Examples

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. Don't forget its cron based :)

Single Consumer

func main() {
  // ...
  var consumeFn kafka.ConsumeFn = func (message kafka.Message) error {
    fmt.Printf("consumer > Message received: %s\n", string(message.Value))
    return nil
  }

  c := cronsumer.New(kafkaConfig, consumeFn)
  c.Run()
}

Single Consumer With Dead Letter

func main() {
  // ...
  var consumeFn kafka.ConsumeFn = func (message kafka.Message) error {
    fmt.Printf("consumer > Message received: %s\n", string(message.Value))
    return errors.New("error occurred")
  }

  c := cronsumer.New(kafkaConfig, consumeFn)
  c.Run()
}

Multiple Consumers

func main() {
  // ...
  var firstConsumerFn kafka.ConsumeFn = func (message kafka.Message) error {
    fmt.Printf("First consumer > Message received: %s\n", string(message.Value))
    return nil
  }
  first := cronsumer.New(firstCfg, firstConsumerFn)
  first.Start()

  var secondConsumerFn kafka.ConsumeFn = func (message kafka.Message) error {
    fmt.Printf("Second consumer > Message received: %s\n", string(message.Value))
    return nil
  }
  second := cronsumer.New(secondCfg, secondConsumerFn)
  second.Start()
  // ...    
}

Single Consumer With Metric collector

func main() {
  // ...
  var consumeFn kafka.ConsumeFn = func(message kafka.Message) error {
    return errors.New("err occurred")
  }
  
  c := cronsumer.New(config, consumeFn)
  StartAPI(*config, c.GetMetricCollectors()...)
  c.Start()
  // ...    
}

func StartAPI(cfg kafka.Config, metricCollectors ...prometheus.Collector) {
  // ...
  f := fiber.New(
    fiber.Config{},
  )
  
  metricMiddleware, err := NewMetricMiddleware(cfg, f, metricCollectors...)
  
  f.Use(metricMiddleware)
  // ...
}

Configurations

Exposed Metrics

Contribute

Use issues for everything

• For a small change, just send a PR.
• For bigger changes open an issue for discussion before sending a PR.
• PR should have:
  • Test case
  • Documentation
  • Example (If it makes sense)
• You can also contribute by:
  • Reporting issues
  • Suggesting new features or enhancements
  • Improve/fix documentation

Please adhere to this project's code of conduct.

Maintainers

• @Abdulsametileri
• @emreodabas

Code of Conduct

Contributor Code of Conduct. By participating in this project you agree to abide by its terms.

Libraries Used For This Project

• segmentio/kafka-go
• robfig/cron
• uber-go/zap

Additional References

• Kcat
• jq
• golangci-lint
• Kafka Console Producer