Microservices in Go: Events and Background jobs using Kafka
Tags: go, kafka
Published: 2021-09-03 00:36:03

Post originally published at [mariocarrion.com](https://mariocarrion.com/2021/06/03/golang-microservices-events-streaming-kafka.html).

## What is Kafka?

[Kafka](https://kafka.apache.org/) is an _event streaming platform_, it allows us to **publish**, **subscribe**, **store** and **process** events. An **Event** indicates something that happened, if we use our [To Do Microservice](https://github.com/MarioCarrion/todo-api-microservice-example) as example, we could define events to indicate a `TaskCreated` or `TaskUpdated` when a Task is created or when a Task is updated, respectively.

## How does Kafka work?

In Kafka there's a concept called _Topic_, _Topics_ store events published by a _Publisher_; those _Topics_ can be named for to differentiate between them and they can be _Partitioned_, what this means is that they could be separated into multiple Kafka brokers, or instances in charge of storing the topics, defining various partitions allows scaling Kafka to allow multiple publishers and consumers to write and read data.

When an event is published to a _Topic_ it will be stored in order it was received, this allows the _Consumers_ to read those events in the same order they were published:

![Kafka Topics](https://mariocarrion.com/img/2021/06/03/1.png?src=devto2)

_Consumers_ use _Group IDs_ to identify themselves when reading events that way multiple processes can consume the same topic in a way the events are still consumed in order:

![Kafka Topics](https://mariocarrion.com/img/2021/06/03/2.png?src=devto)

Because of those _Group IDs_, multiple processes can read the same data at a different pace of speed without affecting other consumers:

![Kafka Topics](https://mariocarrion.com/img/2021/06/03/3.png?src=devto)

## Publisher implementation using a Repository

> The code used for this post is [available on Github](https://github.com/MarioCarrion/todo-api-microservice-example/tree/8f7d667a2136c66ad2f893c7802edf46cafd6f35).

For communicating with Kafka we are going to use the official package [`confluentinc/confluent-kafka-go`](https://github.com/confluentinc/confluent-kafka-go) and similar to other [Repositories](https://mariocarrion.com/2021/04/04/golang-microservices-repository-pattern.html) we implemented previously a new package will be defined called `kafka`.

This package will implement the corresponding [`Task`](https://github.com/MarioCarrion/todo-api-microservice-example/blob/8f7d667a2136c66ad2f893c7802edf46cafd6f35/internal/kafka/task.go) in charge of publishing the events that represent the actions executed when Tasks change to indicate creation, deletion and updates.

The code looks like this:

```go
func (t *Task) Created(ctx context.Context, task internal.Task) error {
	return t.publish(ctx, "Task.Created", "tasks.event.created", task)
}

func (t *Task) Deleted(ctx context.Context, id string) error {
	return t.publish(ctx, "Task.Deleted", "tasks.event.deleted", internal.Task{ID: id})
}

func (t *Task) Updated(ctx context.Context, task internal.Task) error {
	return t.publish(ctx, "Task.Updated", "tasks.event.updated", task)
}
```

Those exported methods use an unexported method called `publish` that interacts directly with the Kafka Producer, the event to be sent will be encoded using `encoding/json`, the idea is to define a JSON payload that then our Consumers can use for determine what event to use, more in the section below covering _Consumers_:

```go
func (t *Task) publish(ctx context.Context, spanName, routingKey string, e interface{}) error {
	// XXX: Excluding OpenTelemetry and error checking for simplicity
	var b bytes.Buffer

	evt := event{
		Type:  msgType,
		Value: task,
	}

	_ = json.NewEncoder(&b).Encode(evt)

	_ = t.producer.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{
			Topic:     &t.topicName,
			Partition: kafka.PartitionAny,
		},
		Value: b.Bytes(),
	}, nil)

	return nil
}
```

To connect this Kafka repository with the PostgreSQL repository we will update each method in `service.Task` to call the corresponding publish-related method, [for example](https://github.com/MarioCarrion/todo-api-microservice-example/blob/8f7d667a2136c66ad2f893c7802edf46cafd6f35/internal/service/task.go#L62-L76):

```go
func (t *Task) Create(ctx context.Context, description string, priority internal.Priority, dates internal.Dates) (internal.Task, error) {
	// XXX: Excluding OpenTelemetry and error checking for simplicity
	task, _ := t.repo.Create(ctx, description, priority, dates)

	// XXX: Transactions will be revisited in future episodes.
	_ = t.msgBroker.Created(ctx, task) // XXX: Ignoring errors on purpose

	return task, nil
}
```

Refer to the [original code](https://github.com/MarioCarrion/todo-api-microservice-example/blob/8f7d667a2136c66ad2f893c7802edf46cafd6f35/internal/service/task.go) to see how the other methods were updated to do something similar.

For consuming the events we will define a new program that will be using the same topic to read the data we published, and then it will update the Elasticsearch values using that repository.

## Consumer implementation

Like I mentioned above the [new program](https://github.com/MarioCarrion/todo-api-microservice-example/blob/8f7d667a2136c66ad2f893c7802edf46cafd6f35/cmd/elasticsearch-indexer-kafka/main.go) will consume the Kafka events and depending on the event type it will call the corresponding [Elasticsearch](https://mariocarrion.com/2021/05/24/golang-microservices-searching-with-elasticsearch.html) method to reindex the values; this program also supports [Graceful shutdown](https://mariocarrion.com/2021/05/21/golang-microservices-graceful-shutdown.html).

The program uses the [`Consumer`](https://pkg.go.dev/github.com/confluentinc/confluent-kafka-go/kafka#Consumer) to poll the values that represent the events [to read](https://github.com/MarioCarrion/todo-api-microservice-example/blob/8f7d667a2136c66ad2f893c7802edf46cafd6f35/cmd/elasticsearch-indexer-kafka/main.go#L142-L190), the simplified code looks like:


```go
for run {
	msg, ok := s.kafka.Consumer.Poll(150).(*kafka.Message)
	if !ok {
		continue
	}

	var evt struct {
		Type  string
		Value internaldomain.Task
	}

	_ = json.NewDecoder(bytes.NewReader(msg.Value)).Decode(&evt)

	switch evt.Type {
	case "tasks.event.updated", "tasks.event.created":
		// call Elasticsearch to index record
	case "tasks.event.deleted":
		// call Elasticsearch to delete record
	}
}
```

When building your final program consider implementing a _Server_-like type, with the goal of separating the different received events into their corresponding types, that way your `switch` could be replaced with a map of functions pointing to the _"handlers"_ of each type being consumed.

## Conclusion

_Kafka_ is a powerful platform for dealing with events, it could be used as a message broker to distribute information across multiple services but it also supports storing and replaying events as well as analysis of those events they are received.

## Recommended Reading

If you're looking to do something similar in RabbitMQ and Redis, I recommend reading the following links:

* [Microservices in Go: Events and Background jobs using RabbitMQ](https://mariocarrion.com/2021/05/28/golang-microservices-events-background-jobs-rabbitmq.html)
* [Microservices in Go: Using Pub/Sub with Redis](https://mariocarrion.com/2021/06/10/golang-microservices-pub-sub-redis.html)

---

> This post includes icons made by [itim2101](https://www.flaticon.com/authors/itim2101) from [Flaticon](https://www.flaticon.com/)