Add Batch Consumer

Makefile

@@ -52,6 +52,12 @@ test-cov: gocov
 test-xml: test-cov gocov-xml
 	@jq -n '{ Packages: [ inputs.Packages ] | add }' $(shell find . -type f -name 'coverage.json' | sort) | $(GOCOVXML) > coverage.xml
 
+.PHONY: test-html
+
+test-html: test-cov gocov-html
+	@jq -n '{ Packages: [ inputs.Packages ] | add }' $(shell find . -type f -name 'coverage.json' | sort) | $(GOCOVHTML) -t kit -r > coverage.html
+	@open coverage.html
+
 .PHONY: check
 check: fmt vet imports lint
 	@git diff --quiet || test $$(git diff --name-only | grep -v -e 'go.mod$$' -e 'go.sum$$' | wc -l) -eq 0 || ( echo "The following changes (result of code generators and code checks) have been detected:" && git --no-pager diff && false ) # fail if Git working tree is dirty
@@ -84,6 +90,10 @@ GOCOVXML = $(BIN_DIR)/gocov-xml
 gocov-xml:
 	$(call go-get-tool,$(GOCOVXML),github.com/AlekSi/[email protected])
 
+GOCOVHTML = $(BIN_DIR)/gocov-html
+gocov-html:
+	$(call go-get-tool,$(GOCOVHTML),github.com/matm/gocov-html/cmd/[email protected])
+
 MOCKERY = $(BIN_DIR)/mockery
 mockery:
 	$(call go-get-tool,$(MOCKERY),github.com/vektra/mockery/[email protected])

examples/xkafka/README.md

@@ -1,4 +1,4 @@
-Examples of how to use the xkafka package to read and write messages to a Kafka topic using `xkafka.Consumer` and `xkafka.Producer`.
+Examples of how to use the xkafka package to publish and consume messages from a Kafka broker. Simulates different Kafka consumer use cases.
 
 ## Running Kafka
 
@@ -10,18 +10,34 @@ $ docker-compose up -d
 
 ## Scenarios
 
-### Sequential Consumer
+### Basic Consumer
 
-This is the default behavior of the consumer. The consumer will process messages sequentially and commit the offset based on the `auto.commit.interval.ms` configuration.
+The basic consumer reads messages from a Kafka topic and prints them to the console. It simulates a process crash by restarting the consumer after a random number of messages have been consumed.
 
 ```bash
-go run *.go sequential
+go run *.go basic --partitions=2 --consumers=2 --messages=10
 ```
 
 ### Async Consumer
 
-Async mode is enabled by setting `xkafka.Concurrency` to a value greater than 1. The consumer will process messages using a pool of Go routines.
+The async consumer reads messages concurrently using a configurable pool of goroutines.
 
 ```bash
-go run *.go async
+go run *.go basic --partitions=2 --consumers=2 --messages=10 --concurrency=4
+```
+
+### Batch Consumer
+
+The batch consumer reads messages in batches of a configurable size.
+
+```bash
+go run *.go batch --partitions=2 --consumers=2 --messages=20 --batch-size=3
+```
+
+### Async Batch Consumer
+
+The async batch consumer processes batches concurrently using a configurable pool of goroutines.
+
+```bash
+go run *.go batch --partitions=2 --consumers=2 --messages=20 --batch-size=3 --concurrency=4
 ```

examples/xkafka/basic.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"context"
+
+	"github.com/rs/zerolog"
+	"github.com/urfave/cli/v2"
+
+	"github.com/gojekfarm/xrun"
+	"github.com/gojekfarm/xtools/xkafka"
+	logmw "github.com/gojekfarm/xtools/xkafka/middleware/zerolog"
+)
+
+func runBasic(c *cli.Context) error {
+	partitions := c.Int("partitions")
+	pods := c.Int("consumers")
+	count := c.Int("messages")
+	concurrency := c.Int("concurrency")
+
+	ctx, cancel := context.WithCancel(c.Context)
+
+	// create topic
+	topic := createTopic(partitions)
+
+	// generate messages
+	messages := generateMessages(topic, count)
+
+	// publish messages
+	publishMessages(messages)
+
+	tracker := NewTracker(messages, cancel)
+
+	// consumer options
+	opts := []xkafka.ConsumerOption{
+		xkafka.Brokers(brokers),
+		xkafka.Topics{topic},
+		xkafka.ConfigMap{
+			"auto.offset.reset": "earliest",
+		},
+		xkafka.Concurrency(concurrency),
+		xkafka.ErrorHandler(func(err error) error {
+			// return error to stop consumer
+			return err
+		}),
+	}
+
+	// create and run consumers
+	runConsumers(ctx, tracker, pods, opts...)
+
+	tracker.Summary()
+
+	return nil
+}
+
+func basicHandler(tracker *Tracker) xkafka.HandlerFunc {
+	return func(ctx context.Context, msg *xkafka.Message) error {
+		err := tracker.SimulateWork(msg)
+		if err != nil {
+			msg.AckFail(err)
+
+			return err
+		}
+
+		tracker.Ack(msg)
+		msg.AckSuccess()
+
+		tracker.CancelIfDone()
+
+		return nil
+	}
+}
+
+func runConsumers(
+	ctx context.Context,
+	tracker *Tracker,
+	pods int,
+	opts ...xkafka.ConsumerOption,
+) {
+	log := zerolog.Ctx(ctx)
+	handler := basicHandler(tracker)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			components := make([]xrun.Component, 0, pods)
+
+			for i := 0; i < pods; i++ {
+				consumer := createConsumer(handler, opts...)
+				components = append(components, consumer)
+			}
+
+			err := xrun.All(xrun.NoTimeout, components...).Run(ctx)
+			if err != nil {
+				log.Error().Err(err).Msg("Error running consumers")
+			}
+		}
+	}
+}
+
+func createConsumer(handler xkafka.HandlerFunc, opts ...xkafka.ConsumerOption) *xkafka.Consumer {
+	consumer, err := xkafka.NewConsumer(
+		"test-basic-consumer",
+		handler,
+		opts...,
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	consumer.Use(logmw.LoggingMiddleware(zerolog.DebugLevel))
+
+	return consumer
+}