KAFKA + KSQLDB + .NET #1
Tags: kafka, ksqldb, microservices, docker
Published: 2021-11-29 20:13:06

Hi, I'm [Ricardo Medeiros](https://github.com/jjackbauer), .NET back end developer @vaivoa, and today I'm going to walk you through using ksqlDB to query messages produced in kafka by a .NET/C# producer. For this example, I will be deploying my enviroment as containers, described in a docker compose file, to ensure easy reproducibility of my results.

The source code used in this example is avaliable [here](https://github.com/jjackbauer/ksqlDBDemo).

## Services 
First, let's talk about the docker compose environment services. the file is avaliable [here](https://github.com/jjackbauer/ksqlDBDemo/blob/main/docker-compose.yml).

### .NET API Producer
Automaticaly generated .NET api with docker compose service
```
ksqldbdemo:
    container_name: ksqldbdemo
    image: ${DOCKER_REGISTRY-}ksqldbdemo
    build:
      context: .
      dockerfile: Dockerfile
```

This producer service needs the .NET generated dockerfile shown below:
```
FROM mcr.microsoft.com/dotnet/aspnet:5.0 AS base
WORKDIR /app
EXPOSE 80
EXPOSE 443

FROM mcr.microsoft.com/dotnet/sdk:5.0 AS build
WORKDIR /src
COPY ["ksqlDBDemo.csproj", "."]
RUN dotnet restore "ksqlDBDemo.csproj"
COPY . .
WORKDIR "/src/"
RUN dotnet build "ksqlDBDemo.csproj" -c Release -o /app/build

FROM build AS publish
RUN dotnet publish "ksqlDBDemo.csproj" -c Release -o /app/publish

FROM base AS final
WORKDIR /app
COPY --from=publish /app/publish .
ENTRYPOINT ["dotnet", "ksqlDBDemo.dll"]
```
### ZooKeeper
Despite not been necessary since Kafka 2.8, ZooKeeper coordinates kafka tasks, defining controllers, cluster membership, topic configuration and more. In this tutorial, it's used the confluent inc. ZooKeeper image, due to it's use in the reference material. It makes Kafka more reliable, but adds complexity into the system.
```
zookeeper:
    image: confluentinc/cp-zookeeper:7.0.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
```

### Kafka
Kafka is an event streaming plataform capable of handling trillions of events a day. Kafka is based on the abstraction of an distributed commit log. Initialiy developed at LinkedIn in 2011 to work as a message queue, but it has evolved into a full-fledge event streanming platfmorm. Listed as broker in the services, is the core of this tutorial. It's configuration is tricky, but using it as follows worked well in this scenario.
```
 broker:
    image: confluentinc/cp-kafka:7.0.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
```
### ksqlDB
ksqlDB is a database built to allow distributed stream process applications. Made to work seamsly with kafka, it has a server that runs outside of kafka, with a REST API and a CLI application that can be run separatly and it's used in this tutorial.
#### ksqlDB Server
In this example, it's used the confluent inc image of the ksqlDB server, once more, due to it's widespread usage.


```
ksqldb-server:
    image: confluentinc/ksqldb-server:0.22.0
    hostname: ksqldb-server
    container_name: ksqldb-server
    depends_on:
      - broker
    ports:
      - "8088:8088"
    environment:
      KSQL_LISTENERS: http://0.0.0.0:8088
      KSQL_BOOTSTRAP_SERVERS: broker:29092
      KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
      KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"
```
#### ksqlDB CLI
The same goes for the ksqlDB CLI service, that also use the confluent inc image.
```
ksqldb-cli:
    image: confluentinc/ksqldb-cli:0.22.0
    container_name: ksqldb-cli
    depends_on:
      - broker
      - ksqldb-server
    entrypoint: /bin/sh
    tty: true
```
### Kafdrop
Kafdrop is a Web UI for viewing kafka topics and browsing consumer groups. It makes kafka more accessible.
```
kafdrop:
    container_name: kafdrop
    image: obsidiandynamics/kafdrop:latest
    depends_on:
      - broker
    ports:
      - 19000:9000
    environment:
      KAFKA_BROKERCONNECT: broker:29092
```
## Tutorial
Now it's the time that you have been waiting, let's make it work!
### Enviroment
For this tutorial, you'll need a [docker desktop](https://docs.docker.com/get-docker/) installation, either it's on a Linux distribution or on Windows with WSL and [git](https://git-scm.com/downloads).
### Cloning the project
A Visual Studio project is avaliable [here](https://github.com/jjackbauer/ksqlDBDemo), it has docker support and already deploys all the services needed for this demo in the IDE. However, you will be fine if you don't want or can't use Visual Studio. Just  clone it, running the following comand on the terminal and directory of your preference: 

 ```
 $ git clone https://github.com/jjackbauer/ksqlDBDemo.git
 ```

Use the following command to move to the project folder:

 ```
 $ cd /ksqlDBDemo
 ```
And, in the project folder, that contains the docker-compose.yml run the following command to deploy the services:

```
$ docker compose up -d
```

after this command, make sure that all services are running. Sometimes services fall, but it is okay. In order to see if everything is running ok, it's possible to see the services running in docker desktop, as shown bellow:

![Docker Desktop](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/3uakexrf3p7atec8q7k9.PNG)

Or you can execute the following command:

```
$ docker ps
```

Which should output something like this:

```
CONTAINER ID   IMAGE                               COMMAND                  CREATED       STATUS       PORTS
  NAMES
b42ce9954fd9   ksqldbdemo_ksqldbdemo               "dotnet ksqlDBDemo.d…"   2 hours ago   Up 2 hours   0.0.0.0:9009->80/tcp, 0.0.0.0:52351->443/tcp   ksqldbdemo
0a0186712553   confluentinc/ksqldb-cli:0.22.0      "/bin/sh"                2 hours ago   Up 2 hours
  ksqldb-cli
76519de6946e   obsidiandynamics/kafdrop:latest     "/kafdrop.sh"            2 hours ago   Up 2 hours   0.0.0.0:19000->9000/tcp
  kafdrop
11c3a306ee01   confluentinc/ksqldb-server:0.22.0   "/usr/bin/docker/run"    2 hours ago   Up 2 hours   0.0.0.0:8088->8088/tcp
  ksqldb-server
07cef9d69267   confluentinc/cp-kafka:7.0.0         "/etc/confluent/dock…"   2 hours ago   Up 2 hours   9092/tcp, 0.0.0.0:29092->29092/tcp
  broker
3fa1b9a60954   confluentinc/cp-zookeeper:7.0.0     "/etc/confluent/dock…"   2 hours ago   Up 2 hours   2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp     zookeeper
```

### WEB API
Now, with all services up and running, we can access the WEB API Swagger to populate our Kafka topics. The code is very simple and it's avaliable in the [repository](https://github.com/jjackbauer/ksqlDBDemo).

The WEB API swagger is deployed at [http://localhost:9009/swagger/index.html](http://localhost:9009/swagger/index.html). As shown in the image bellow, it has two endpoints and they create events that could be created by indepent microservices. One for creating an event that creates a userName in the system and another that takes an Id and generates a three digit code.


![Swagger Geral](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/99l7s38ffu0tx1gon62r.PNG)

Then you can create an User with the user name of your choise, as shown:
![Request Create user](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/mxlhjsmfeei7wnym834g.PNG)

And it will have an assigned unique Id, as demonstrated:

![Response create user](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/3g06gff8025nxw6gze9v.PNG)

Now, you can get a three digit code for your user Id as displayed:

![Get Code Request](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/6yoezrw53wibte9baiaj.PNG)

And a random code is generated for the selectd, as we can observe in the image that follows:


![Get Code Response](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/azp6yri8cy4x4day7vyu.PNG)

### Kafdrop
We can use the kafdrop UI the check if everything is okay. Kafdrop is deployed at [http://localhost:19000/](http://localhost:19000/).
There, you will find all the brokers and topics avaliable. It should look like this:


![Kafdrop](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/y9v29cx12cqw5cnp19sr.PNG)

### KSQL CLI
After all that, you'll be able to create your streams of data and query it using ksqlDB. On your preferential terminal, use the command: 

```
docker exec -it ksqldb-cli ksql http://ksqldb-server:8088
```
#### Creating streams
And then you are in the ksql CLI and are free to create your streams and queries. First, let's create a stream for each one of our topics:

```
CREATE STREAM stream_user (Name VARCHAR, Id VARCHAR)
  WITH (kafka_topic='demo-user', value_format='json', partitions=1);
```
```
CREATE STREAM stream_code (Id VARCHAR, code INT)
  WITH (kafka_topic='demo-code', value_format='json', partitions=1);
```

#### Create a materialized view
You can join the client data with the most recent randomized code. to achieve this, you must create a materialized view table, that joins both streams as seen in the ksqldb script that follows:

```
CREATE TABLE currentCodeView AS
>   SELECT user.Name,
>   LATEST_BY_OFFSET(code.code) AS CurrentCode
>   FROM stream_code code INNER JOIN stream_user user
>   WITHIN 7 DAYS ON code.Id = user.Id
>   GROUP BY user.Name
>EMIT CHANGES;
```

#### Making a push query
After that, we can query this materialized view:

```
SELECT * FROM currentCodeView 
  EMIT CHANGES;
```
This push query keep on running until you hit cntrl+c to cancel it.

## Conclusions
In this tutorial it's demonstrated that in a kafka + ksqlDB enviroment, you can make SQL queries and also join on data that comes from different events, which is one of most complexities envolved with microsservices systems. And it is what ksqlDB solves by enabling SQL operations over Kafka topics.
It's my goal to explore the possibilites allowed by this ecosystem and I hope to bring more knowledge on this topic in another articles here. Any sugestions, comments or corrections, fell free to reach me out at [LinkedIn](https://www.linkedin.com/in/rmedio/).

{% user jjackbauer %}





## References
[ksqlDB Quickstart](https://ksqldb.io/quickstart.html?_ga=2.218008467.482211024.1638022122-847939024.1633623088&_gac=1.142412294.1634140787.EAIaIQobChMIjOL6pt_H8wIVmcWaCh1KbwgwEAEYASAAEgLBFvD_BwE)
[ksqlDB Overview](https://docs.confluent.io/platform/current/ksqldb/index.html#ksql-home)
[Kafka .NET Client] (https://docs.confluent.io/clients-confluent-kafka-dotnet/current/overview.html)
[ksqlDB Documentation - Data Types Overview](https://docs.ksqldb.io/en/latest/reference/sql/data-types/)
[KSQL and ksqlDB](https://docs.ksqldb.io/en/latest/operate-and-deploy/ksql-vs-ksqldb/)
[Welcome to Apache ZooKeeper](https://zookeeper.apache.org/)
[What is ZooKeeper & How Does it Support Kafka?](https://dattell.com/data-architecture-blog/what-is-zookeeper-how-does-it-support-kafka/)
[What is Apache Kafka®?](https://www.confluent.io/what-is-apache-kafka/?utm_medium=sem&utm_source=google&utm_campaign=ch.sem_br.nonbrand_tp.prs_tgt.kafka_mt.xct_rgn.latam_lng.eng_dv.all_con.kafka-general&utm_term=apache%20kafka&creative=&device=c&placement=&gcli
d=Cj0KCQiA7oyNBhDiARIsADtGRZYDVaYjkPkoJQHNrz_xBodIq2P8ztwb8g3OTiRG_wMHXyzof1nqKEMaAoT_EALw_wcB)
[ksqlDB - The database purpose-built for stream processing applications](https://ksqldb.io/)
[An overview of ksqlDB](https://ksqldb.io/overview.html)
[CREATE TABLE AS SELECT](https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/create-table-as-select/)
[How to join a stream and a stream](https://kafka-tutorials.confluent.io/join-a-stream-to-a-stream/ksql.html)
[Time and Windows in ksqlDB Queries](https://docs.ksqldb.io/en/latest/concepts/time-and-windows-in-ksqldb-queries/)
[Time operations](https://docs.ksqldb.io/en/latest/reference/sql/time/)