Functional Kafka with Spring Cloud - Part 1
Tags: spring, cloud, kafka, functional
Published: 2022-04-27 20:30:46

I've put this article together to create a working demo of Spring Cloud Kafka which I've been unable to find so far.

Prerequisites:
* Java 11+ (I'm using Java 18)
* Linux based OS (I'm on macOS)
* [Httpie](https://httpie.io/cli)

Setup:
* Spring Cloud 2021.0.1
* Confluent Schema Registry 7.1.0
* Apache Kafka 2.13_3.1.0
* Apache ZooKeeper 3.7.0

This article will first start with setting up a web API publishing events to Kafka as a string with a functional kafka consumer using Spring Cloud Stream.

## Set up the environment
Download Apache ZooKeeper from here:

```
https://zookeeper.apache.org/releases.html#download
```

Decompress it and move it to a working folder (we will use `$ZOOKEEPER_HOME`)

Download Kafka from here:

```
https://kafka.apache.org/downloads
```

Again, decompress the archive and move it to a working folder, this time we will refer to the working folder as `$KAFKA_HOME`

Download and decompress the Schema Registry using the following:

```
$ wget https://packages.confluent.io/archive/7.0/confluent-community-7.0.1.tar.gz
$ tar -xf confluent-community-7.0.1.tar.gz
$ cd confluent-7.0.1
```

We will refer to this applications location at `$CONFLUENT_HOME`

## Configure the environment

First let's set up ZooKeeper!

### ZooKeeper Config
Create the config file by copying the example config:

```
ZOOKEEPER_HOME $> cp conf/zoo_sample.cfg conf/zoo.cfg
```

The only value you may want to edit is the dataDir directory to a location that you are comfortable storing the metadata to:

```
...
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=./data
```

Save the file and we will move on to configuring Kafka.

### Kafka Config

The file we are interested here is:

```
$KAFKA_HOME/config/server.properties
```

There are 2 values we most likely will change:

* `log.dirs=./logs`
* `zookeeper.connect=localhost:2181`

This ensures our logs are in an accessible location as well as allows Kafka to connect to ZooKeeper to persist data.

### Configure Schema Registry

Just one file needs to be checked here: `$CONFLUENT_HOME/etc/schema-registry/schema-registry.properties`

All the default should be fine:

```
listeners=http://0.0.0.0:8081
kafkastore.bootstrap.servers=PLAINTEXT://localhost:9092
kafkastore.topic=_schemas
debug=false
```

## Starting ZooKeeper and Kafka

We won't be using the Schema Registry to begin with since we are just working with `String` values in Kafka so let's get them started:

```
$ZOOKEEPER_HOME > bin/zkServer.sh start conf/zoo.cfg
ZooKeeper JMX enabled by default
Using config: conf/zoo.cfg
Starting zookeeper ... STARTED

$KAFKA_HOME > bin/kafka-server-start.sh config/server.properties
...
[2022-04-25 11:19:17,742] INFO Kafka version: 3.1.0 (org.apache.kafka.common.utils.AppInfoParser)
[2022-04-25 11:19:17,743] INFO Kafka commitId: 37edeed0777bacb3 (org.apache.kafka.common.utils.AppInfoParser)
[2022-04-25 11:19:17,743] INFO Kafka startTimeMs: 1650910757737 (org.apache.kafka.common.utils.AppInfoParser)
[2022-04-25 11:19:17,745] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
```

Okay, let's get programming!!!

## Employee API

This will be a simple API that will havea single `POST` endpoint to create a kafka message.

Just head to https://start.spring.io and create a basic web application:


![Spring Initializer](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/yjeeepytett4r2ewas5b.jpeg)

Click download and expand the `demo.zip` file to a workspace

To get started we are going to need some dependencies to publish to Kafka.

To your `pom.xml` file add the following dependency:

```
<dependency>
	<groupId>org.springframework.kafka</groupId>
	<artifactId>spring-kafka</artifactId>
	<version>2.8.4</version>
</dependency>
```

Next we will create the controller to serve our endpoint.

```
package com.example.demo;

import com.example.demo.model.Employee;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.net.URI;
import java.util.UUID;

@RequestMapping("/employee")
@RestController
public class EmployeeController {

    @Autowired
    protected KafkaTemplate<String, String>  kafka;

    public ResponseEntity<Void> createEmployee(@RequestParam("firstname") String firstname,
                                               @RequestParam("lastname") String lastname) {

        String id = UUID.randomUUID().toString();
        kafka.send("employee", id, String.format("%s, %s", lastname, firstname));

        return ResponseEntity.created(URI.create(String.format("/employee/%s", id))).build();
    }

}
```

Let's quickly walk through the code...

First, we inject the `KafkaTemplate<String, String> template` since we aren't doing anything fancy, this is going to just send the message key as a `java.lang.String` and the body of the message will also be of type `java.lang.String`, nothing fancy.

Our `@PostMapping` is going to simply be the endpoint with 2 query parameters:
* firstname
* lastname

So the URL will have the format:
`http://localhost:8050/employee?firstname=Paula&lastname=Abdul`

And when we call the API we will use Httpie with the following syntax:

```
http POST http://localhost:8070/employee firstname==Paula lastname==Abdul
```

Next we are just generating a random UUID as the message id and send this along with a concatenated string of the lastname and firstname to the `employee` topic:

```
String id = UUID.randomUUID().toString();
kafka.send("employee", id, String.format("%s, %s", lastname, firstname));
```

Before we start the application, lets tweak the configuration. In the `src/main/resources/application.yml` file, let's make it the same as:

```
spring:
  kafka:
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    bootstrap-servers:
      - localhost:9092

server:
  port: 8070
```

Now we can run the application and hit the endpoint:

```
./mvnw spring-boot:run
...
: Tomcat initialized with port(s): 8070 (http)
: Starting service [Tomcat]
: Starting Servlet engine: [Apache Tomcat/9.0.54]
: Initializing Spring embedded WebApplicationContext
: Root WebApplicationContext: initialization completed in 754 ms
: Tomcat started on port(s): 8070 (http) with context path ''
: Started DemoApplication in 1.522 seconds (JVM running for 1.841)
```

Call the endpoint:

```
$ http POST http://localhost:8070/employee firstname==Andrew lastname==Marshall
HTTP/1.1 201 
Connection: keep-alive
Content-Length: 0
Date: Wed, 27 Apr 2022 17:30:39 GMT
Keep-Alive: timeout=60
Location: /employee/830da346-38b9-4d5b-a051-a302c395333e
```

You can track the message in Kafka using the `kafka-console-consumer.sh` command:

```
$ $KAFKA_HOME/bin/kafka-console-consumer  --topic employee --bootstrap-server localhost:9092 --from-beginning
Marshall, Andrew
```
Next article we will setup a functional consumer using Spring Cloud Stream...