Publish Vonage Events to Kafka with Spring Boot
Tags: spring, kotlin, kafka, vonage
Published: 2020-07-26 22:29:23

[Event-Driven Architecture](https://en.wikipedia.org/wiki/Event-driven_architecture) is the asynchronous, dynamic processing of events to communicate between decoupled services.

For example, we might have a unified communication platform that manages millions of users and conversations. We might bring in these conversations from a multitude of sources and process them in many different ways.

Event-Driven Architecture enables us to distribute processing of these events between various services which might:

- Transfer information to long-term storage like a [data lake](https://en.wikipedia.org/wiki/Data_lake)
- Perform administrative tasks like stopping spammers in real-time.

Events can come from many different channels. In this guide, you will build a [Spring Boot](https://spring.io) application, using Kotlin, that can receive events from the [Vonage Communication APIs](https://developer.nexmo.com/) and store them in [Apache Kafka](https://kafka.apache.org/).

This guide assumes that you have some experience with [Spring Boot](https://spring.io) and the [Vonage Communication APIs](https://developer.nexmo.com).

An example of the code can be found on GitHub: [cr0wst/vonage-events-kafka](https://github.com/cr0wst/vonage-events-kafka)

### What are the Vonage Communication APIs?
The [Vonage Communication APIs](https://developer.nexmo.com/) enable developers to create things like in-app messaging, voice chat, and integration with third-party chat services.

Many of these actions generate [Events](https://developer.nexmo.com/conversation/concepts/event), which can be handled by your application via a pre-configured [Webhook](https://developer.nexmo.com/concepts/guides/webhooks).

I have created a [Demo Application](https://github.com/cr0wst/vonage-conversation-demo) that can generate a random conversation to produce events.

### What is Kafka?
[Apache Kafka](https://kafka.apache.org/) is a horizontally scalable, fault-tolerant tool for building real-time data pipelines and streaming applications.

Kafka allows you to publish and subscribe to [topics](), which serve as a category for records to be published. A topic can have multiple publishers and consumers. Unlike traditional message brokers, it can store historical data indefinitely.

You can read more about Kafka on the [Confluent Docs](https://docs.confluent.io/current/kafka/introduction.html).

This guide will cover the following concepts:

- Connecting a Spring Boot Application to Kafka.
- Creating a controller to serve as a [Webhook](https://developer.nexmo.com/concepts/guides/webhooks) for incoming Vonage Communication events.
- Pipelining these events into a Kafka topic for later processing.

## Download the Demo Application

The [Pre-Setup](https://github.com/cr0wst/vonage-conversation-demo#pre-setup) instructions will guide you through creating a Vonage Application. You can also follow the official documentation for [Creating an Application](https://developer.nexmo.com/application/overview#creating-applications).

Most of this guide assumes that you are already receiving webhook events from Vonage, but you aren't currently sending them to Kafka.

If you don't have an application, I have created a [Demo Application](https://github.com/cr0wst/vonage-conversation-demo), which simulates multiple conversations using the Vonage Conversation API.

## Initialize a New Spring Boot Application

Use your IDE or the [Spring Initializr](https://start.spring.io) to create a new Spring Boot Application with **Spring For Apache Kafka** and **Spring Reactive Web**:

![The Spring Initializr with Kafka and Reactive Web Selected](https://dev-to-uploads.s3.amazonaws.com/i/64eaw10l0v9l2hvp1552.png)

## Create a Webhook

The Vonage Communication APIs send event messages to a pre-configured webhook. 

### Configure the Event URL

The Event URL can be defined using the [Vonage Dashboard](https://dashboard.nexmo.com/applications/)

![The Vonage Dashboard showing an application being edited to have its RTC event URL updated](https://github.com/cr0wst/vonage-conversation-demo/raw/master/capability.png)

> This has to be a publicly-accessible URL. You can use a tool like [ngrok](https://ngrok.com/) to safely tunnel traffic to your application. Check out this post on [Working with Webhooks using Ngrok](https://dev.to/bitsmonkey/working-with-webhooks-using-ngrok-38a1).

I'm going to start a tunnel using Ngrok to listen on port 8080:

```sh
ngrok http 8080 -subdomain=cr0wst
```

### Create a Controller

Create a new class called `EventWebhookController` and add the following code:

```kotlin
package com.smcrow.demo

import org.slf4j.LoggerFactory
import org.springframework.web.bind.annotation.PostMapping
import org.springframework.web.bind.annotation.RequestBody
import org.springframework.web.bind.annotation.RestController

@RestController
class EventWebhookController {
    @PostMapping("/webhooks/events")
    fun incomingEvents(@RequestBody body: String) {
        // Handle the Event Here
        log.info(body)
    }

    companion object {
        private val log = LoggerFactory.getLogger(this::class.java)
    }
}
```

This code registers a route at `/webhooks/events` that accepts `POST` requests and logs the request body.

If you were to start both applications, you would see a stream of Vonage events:

Start the Webhook Application:
```sh
./mvnw spring-boot:run
```

Now, start the demo application:

```sh
cd ~/Code/vonage-conversation-demo
./mvnw spring-boot:run
```

The top terminal shows the events from Vonage, and the bottom is the application generating random conversations and messages:

![A horizontally split terminal showing the webhook application on top and the automatic conversation app running on the bottom](https://dev-to-uploads.s3.amazonaws.com/i/k5auvixd7wpij8jsis4z.png)

Now that your application is receiving events from Vonage, it's time to set up Kafka.

## Setting up Kafka

If you don't already have access to a Kafka cluster, there are a few solutions.

The simplest way is to sign up for a service like [Confluent Cloud](https://www.confluent.io/confluent-cloud). I will be using Confluent for the remainder of this guide and the [Confluent Cloud CLI](https://docs.confluent.io/current/cloud/cli/index.html)

### Create a Cluster

Create a new cluster on Confluent Cloud with the following command:

```sh
ccloud kafka cluster create demo --cloud aws --region us-east-2
```

The output has an ID number and an Endpoint, record these for later. For this guide, I will be using the ID `lkc-03wkq` and endpoint `pkc-ep9mm.us-east-2.aws.confluent.cloud:9092`

### Create a Topic

Create the `vonage.webhook.events` topic inside of your Kafka cluster for storing the Vonage webhook events:

```sh
ccloud kafka topic create --cluster lkc-03wkq --partitions 1 vonage.webhook.events
```

## Connecting a Spring Boot Application to Kafka

To connect your Spring Boot Application to Confluent Cloud, you'll need to create an API Key and Secret. Run the following command, replacing the resource with your ID from a previous step.

```sh
ccloud api-key create --resource=lsrc-7qz91
```

You will also need your bootstrap server address, which is the endpoint from a previous step.

Using the bootstrap server address, API Key and Secret, add the following properties to `application.properties`, replacing the appropriate variables.

```properties
# Kafka
spring.kafka.properties.ssl.endpoint.identification.algorithm=https
spring.kafka.properties.sasl.mechanism=PLAIN
spring.kafka.bootstrap-servers={{ BOOTSTRAP_SERVER_ADDRESS }}
spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule   required username="{{ CLUSTER_API_KEY }}"   password="{{ CLUSTER_API_SECRET }}";
spring.kafka.properties.security.protocol=SASL_SSL

# Producer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
```

For example:

```properties
# Kafka
spring.kafka.properties.ssl.endpoint.identification.algorithm=https
spring.kafka.properties.sasl.mechanism=PLAIN
spring.kafka.bootstrap-servers=pkc-ep9mm.us-east-2.aws.confluent.cloud:9092
spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule   required username="my-api-key"   password="my-api-secret";
spring.kafka.properties.security.protocol=SASL_SSL

# Producer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
```

## Produce Messages from Spring Boot

The [Spring Kafka](https://spring.io/projects/spring-kafka) project uses a `KafkaTemplate` object to produce messages. `KafkaTemplate` is generic with two generic types: one for the Kafka key, and one for the value.

Update the `EventWebhookController` to inject a `KafkaTemplate<String, String>` as a dependency:

```kotlin
package com.smcrow.demo

import org.slf4j.LoggerFactory
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.web.bind.annotation.PostMapping
import org.springframework.web.bind.annotation.RequestBody
import org.springframework.web.bind.annotation.RestController

@RestController
class EventWebhookController(private val kafkaTemplate: KafkaTemplate<String, String>) {
    @PostMapping("/webhooks/events")
    fun incomingEvents(@RequestBody body: String) {
        // Handle the Event Here
        log.info(body)
    }

    companion object {
        private val log = LoggerFactory.getLogger(this::class.java)
    }
}
```

Now, update the `incomingEvents` method to send the event to Kafka:

```kotlin
package com.smcrow.demo

import org.slf4j.LoggerFactory
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.web.bind.annotation.PostMapping
import org.springframework.web.bind.annotation.RequestBody
import org.springframework.web.bind.annotation.RestController

@RestController
class EventWebhookController(private val kafkaTemplate: KafkaTemplate<String, String>) {
    @PostMapping("/webhooks/events")
    fun incomingEvents(@RequestBody body: String) {
        // Handle the Event Here
        kafkaTemplate.send("vonage.webhook.events", body)
    }

    companion object {
        private val log = LoggerFactory.getLogger(this::class.java)
    }
}
```

## Consuming Kafka Messages Using the CLI

You can use the Confluent Cloud CLI to consume events on the `vonage.webhook.events` topic.

First, configure the CLI to use your API key from before, replacing `my-api-key` with your API key and `lkc-03wkq` with the ID from a previous step:

```sh
ccloud api-key use my-api-key --resource lkc-03wkq
```

Now, start consuming messages:

```sh
ccloud kafka topic consume --cluster lkc-03wkq -b vonage.webhook.events
```

> Remember that Kafka is a commit log. The `-b` flag starts consumption from the beginning. Using this flag means you'll get a back-fill of messages that you missed while you weren't connected.

## Start Everything

First, start the Kafka Consumer if you haven't already:

```sh
ccloud kafka topic consume --cluster lkc-03wkq -b vonage.webhook.events
```

Now start the Webhook Application:

```sh
cd ~/Code/vonage-events-kafka-demo
./mvnw spring-boot:start
```

Finally, start the conversation simulator application:

```sh
cd ~/Code/vonage-conversation-demo
./mvnw spring-boot:run
```

The following shows all three running:

![A horizontally split terminal. The top is vertically split showing the two applications running. The bottom is showing the kafka consumer output](https://dev-to-uploads.s3.amazonaws.com/i/ys3xq80ibsr5r7vz2oze.png)

## Conclusion

In this guide, you created a Spring Boot Application that can listen to Vonage Events and publish them to Kafka.

An example of the code can be found on GitHub: [cr0wst/vonage-events-kafka](https://github.com/cr0wst/vonage-events-kafka)

In another guide, I'll show you how to consume these events and store them in a relational database like [Snowflake](https://www.snowflake.com/). 

Here are some links you might find useful:

- [The Spring Kafka Project](https://spring.io/projects/spring-kafka)
- [Confluent Kafka and Spring Boot](https://www.confluent.io/blog/apache-kafka-spring-boot-application)