kafka consumer template

Author: Pythonista7

.gitignore

@@ -0,0 +1,23 @@
+# If you prefer the allow list template instead of the deny list, see community template:
+# https://github.com/github/gitignore/blob/main/community/Golang/Go.AllowList.gitignore
+#
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/
+
+# Go workspace file
+go.work
+
+go.sum

README.md

@@ -0,0 +1,48 @@
+# Simple Kafka Consumer in Go
+
+This is a template repository for a simple Kafka consumer in Go. It uses the [confluent pkg](https://github.com/confluentinc/confluent-kafka-go).
+
+## Instructions
+
+1. Clone this repository into your `$GOPATH/src` directory.
+
+2. Make sure your imports are correct and run `go mod tidy` to update the `go.sum` file.
+
+3. Start up a local kafka instance using the below command. This docker compose file will automatically create a topic called `test-topic-1`.Kafka-REST is also included to make it easy to interact with the kafka instance vua HTTP.
+
+    ```bash
+    docker-compose up -d 
+    ```
+
+    You can check that the topic was created by running the following command:
+
+    ```bash
+    curl --request GET --url http://localhost:38082/topics
+    ```
+
+4. Run `go run main.go` to start the consumer.
+
+    *Note: If you are using a different kafka instance, you will need to update the properties in the `config/config.go` file.*
+
+5. Send a message to the topic using the following command:
+
+    ```bash
+    curl --request POST \
+    --url http://localhost:38082/topics/test-topic-1 \
+    --header 'content-type: application/vnd.kafka.json.v2+json' \
+    --data '{
+        "records": [
+            {
+                "value": {
+                    "taskType":"taskType1",
+                    "data":{
+                        "id":1,
+                        "name": "John Doe"
+                    }
+                }
+            }
+        ]
+    }'
+    ```
+
+    You should see the message printed to the console.

config/config.go

@@ -0,0 +1,26 @@
+package config
+
+import (
+	"github.com/ilyakaznacheev/cleanenv"
+	"github.com/sirupsen/logrus"
+)
+
+type Configuration struct {
+	AppEnv            string `env:"APP_ENV" env-description:"app environment" env-default:"staging"`
+	LogFormat         string `env:"LOG_FORMAT" env-description:"format of the go-consumer log" env-default:"text"`
+	NoOfThreads       int    `env:"NO_OF_THREADS" env-description:"Total number of threads/go-routines to be spawned" env-default:"4"`
+	ChannelBufferSize int    `env:"CHANNEL_BUFFER_SIZE" env-description:"total size of channel buffer" env-default:"5"`
+
+	KafkaBroker string `env:"KAFKA_BROKER" env-description:"kafka broker address host:port" env-default:"localhost:9092"`
+	KafkaGroup  string `env:"KAFKA_GROUP" env-description:"kafka consumer group" env-default:"consumer-0"`
+	KafkaTopic  string `env:"KAFKA_TOPIC" env-description:"kafka topic" env-default:"test-topic-1"`
+}
+
+var Config Configuration
+
+func init() {
+	err := cleanenv.ReadEnv(&Config)
+	if err != nil {
+		logrus.Errorf("Something went wrong while loading configurations from env | Error: %s", err.Error())
+	}
+}

consumer/consumer.go

@@ -0,0 +1,37 @@
+package consumer
+
+import (
+	"github.com/pythonista7/go-kafka-consumer/utils"
+	"github.com/pythonista7/go-kafka-consumer/worker"
+	"github.com/sirupsen/logrus"
+)
+
+// Main Consumer
+func ConsumeQueue(
+	log *logrus.Logger,
+	consumeQueue <-chan utils.KafkaPayload,
+) error {
+	for payload := range consumeQueue {
+		log.Debugf("Processing payload ...")
+
+		switch payload.TaskType {
+		case "taskType1":
+			{
+				err := worker.PerformTask1(log, payload.Data)
+				if err != nil {
+					log.Errorf("Error performing task1 : %s", err.Error())
+					continue
+				}
+			}
+			/*
+				Add more cases as necessary
+			*/
+		default:
+			{
+				log.Errorf("Invalid task type: %s", payload.TaskType)
+			}
+		}
+	}
+
+	return nil
+}

docker-compose.yml

@@ -0,0 +1,72 @@
+---
+version: '2'
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:latest
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  kafka:
+    # "`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-
+    # An important note about accessing Kafka from clients on other machines: 
+    # -----------------------------------------------------------------------
+    #
+    # The config used here exposes port 9092 for _external_ connections to the broker
+    # i.e. those from _outside_ the docker network. This could be from the host machine
+    # running docker, or maybe further afield if you've got a more complicated setup. 
+    # If the latter is true, you will need to change the value 'localhost' in 
+    # KAFKA_ADVERTISED_LISTENERS to one that is resolvable to the docker host from those 
+    # remote clients
+    #
+    # For connections _internal_ to the docker network, such as from other services
+    # and components, use kafka:29092.
+    #
+    # See https://rmoff.net/2018/08/02/kafka-listeners-explained/ for details
+    # "`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-'"`-._,-
+    #
+    image: confluentinc/cp-kafka:latest
+    depends_on:
+      - zookeeper
+
+    # reachable on 9092 from the host and on 29092 from inside docker compose
+    ports:
+      - 9092:9092
+    expose:
+      - '29092'
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+
+  init-kafka:
+    image: confluentinc/cp-kafka:latest
+    depends_on:
+      - kafka
+    entrypoint: [ '/bin/sh', '-c' ]
+    command: |
+      "
+      # blocks until kafka is reachable
+      kafka-topics --bootstrap-server kafka:29092 --list
+
+      echo -e 'Creating kafka topics'
+      kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic test-topic-1 --replication-factor 1 --partitions 1
+
+      echo -e 'Successfully created the following topics:'
+      kafka-topics --bootstrap-server kafka:29092 --list
+      "
+
+  kafka-rest:
+    image: confluentinc/cp-kafka-rest:4.1.1
+    hostname: kafka-rest
+    ports:
+      - "38082:38082"
+    depends_on:
+      - kafka
+    environment:
+      KAFKA_REST_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_REST_HOST_NAME: kafka-rest
+      KAFKA_REST_LISTENERS: http://kafka-rest:38082

go.mod

@@ -0,0 +1,10 @@
+module github.com/pythonista7/go-kafka-consumer
+
+go 1.15
+
+require (
+	github.com/confluentinc/confluent-kafka-go v1.9.2 // indirect
+	github.com/ilyakaznacheev/cleanenv v1.3.0
+	github.com/sirupsen/logrus v1.9.0
+	gopkg.in/confluentinc/confluent-kafka-go.v1 v1.8.2
+)

main.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/pythonista7/go-kafka-consumer/config"
+	"github.com/pythonista7/go-kafka-consumer/consumer"
+	"github.com/pythonista7/go-kafka-consumer/utils"
+	"github.com/sirupsen/logrus"
+	"gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
+)
+
+func main() {
+
+	log := logrus.New()
+	log.Out = os.Stdout
+	log.SetReportCaller(true)
+
+	if config.Config.AppEnv != "production" {
+		log.SetLevel(logrus.TraceLevel)
+	}
+
+	if config.Config.LogFormat == "json" {
+		log.SetFormatter(&logrus.JSONFormatter{})
+	}
+
+	topics := strings.Split(config.Config.KafkaTopic, ",")
+	fmt.Printf("Broker : %s\n", config.Config.KafkaBroker)
+	sigchan := make(chan os.Signal, 1)
+	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)
+
+	c, err := kafka.NewConsumer(&kafka.ConfigMap{
+		"allow.auto.create.topics":        true,
+		"bootstrap.servers":               config.Config.KafkaBroker,
+		"group.id":                        config.Config.KafkaGroup,
+		"session.timeout.ms":              6000,
+		"go.events.channel.enable":        true,
+		"go.application.rebalance.enable": true,
+		// Enable generation of PartitionEOF when the
+		// end of a partition is reached.
+		"enable.partition.eof": true,
+		"auto.offset.reset":    "earliest"})
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
+		os.Exit(1)
+	}
+
+	log.Infof("Created Consumer %s\n", c)
+
+	//define channel
+	consume := make(chan utils.KafkaPayload, config.Config.ChannelBufferSize)
+
+	// spawn go routine in block state for the channel above
+	for i := 0; i < config.Config.NoOfThreads; i++ {
+		go consumer.ConsumeQueue(log, consume)
+	}
+
+	// Note: you can also use the non-channel based approach, using the poll API as shown here:
+	// https://github.com/confluentinc/confluent-kafka-go/blob/master/examples/consumer_example/consumer_example.go
+
+	err = c.SubscribeTopics(topics, nil)
+	if err != nil {
+		log.Errorf("Error while subscribing topics: %v | Error: %s", topics, err.Error())
+	}
+
+	run := true
+
+	for run == true {
+
+		select {
+		case sig := <-sigchan:
+			log.Infof("Caught signal %s: terminating\n", sig.String())
+			run = false
+
+		case ev := <-c.Events():
+			switch e := ev.(type) {
+			case kafka.AssignedPartitions:
+				log.Infof("AssignedPartitions : %v\n", e)
+				fmt.Fprintf(os.Stderr, "%% %v\n", e)
+				c.Assign(e.Partitions)
+
+			case kafka.RevokedPartitions:
+				log.Errorf("RevokedPartitions : %v\n", e)
+				fmt.Fprintf(os.Stderr, "%% %v\n", e)
+				c.Unassign()
+
+			case *kafka.Message:
+				log.Debugf("Buffer occupency %d / %d", len(consume), config.Config.ChannelBufferSize)
+				log.Debugf("Picked up kafka message at : %v", time.Now())
+				log.Infoln("Processing Kafka Message ...")
+
+				var payload utils.KafkaPayload
+				err = json.Unmarshal(e.Value, &payload)
+				consume <- payload
+
+			case kafka.PartitionEOF:
+				log.Infof("Reached %v\n", e)
+
+			case kafka.Error:
+				// Errors should generally be considered as informational, the client will try to automatically recover
+				fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)
+				log.Errorf("Error: %v\n", e)
+			}
+		}
+	}
+	fmt.Printf("Closing consumer\n")
+	c.Close()
+}

utils/types.go

@@ -0,0 +1,6 @@
+package utils
+
+type KafkaPayload struct {
+	TaskType string                 `json:"taskType"`
+	Data     map[string]interface{} `json:"data"`
+}

utils/utils.go

@@ -0,0 +1,5 @@
+package utils
+
+/*
+	Add generic/frequently used functions here.
+*/

worker/task1Worker.go

@@ -0,0 +1,33 @@
+package worker
+
+import (
+	"encoding/json"
+
+	"github.com/sirupsen/logrus"
+)
+
+// implement custom definition
+type Task1Payload struct {
+	Id   int    `json:"id"`
+	Name string `json:"name"`
+}
+
+func PerformTask1(log *logrus.Logger, payload map[string]interface{}) error {
+	// typecast payload to custom definition
+	var task1Payload Task1Payload
+
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		log.Errorf("Error decoding JSON : %v", err.Error())
+	}
+
+	err = json.Unmarshal(raw, &task1Payload)
+	if err != nil {
+		log.Errorf("Error converting JSON to struct : %v", err.Error())
+	}
+
+	// implement your custom logic
+	log.Infof("Performing task 1 for id : %d , name: %s", task1Payload.Id, task1Payload.Name)
+
+	return nil
+}