While at work, monitoring data streams became a hassle and more tedious as time goes then we opted to use real time data streaming service. and to achieve this goal we used apache Kafka as the distributed data streaming services which is fully managed , eliminates the need for us to handle infrastructure management, scaling, or maintenance tasks.

https://github.com/A-LPHARM/K8s-apache-kafka-strimzi

Kafka is a stream processing system used for messaging, website activity tracking, metrics collection and monitoring, logging, event sourcing, commit logs, and real-time analytics. It’s a good fit for large scale message processing applications since it is more robust, reliable, and fault-tolerant compared to traditional message queues.

Open-source Kafka or managed distributions are ubiquitous in modern software development environments. Kafka is used by developers and data engineers at companies such as Uber, Square, Strave, Shopify, and Spotify.

What is strimzi?

Strimzi simplifies the process in deploying and managing Apache Kafka in a Kubernetes cluster. As a Cloud Native Computing Foundation Sandbox project, it provides container images and Operators for running Kafka on Kubernetes environment.

Strimzi Operators are fundamental to this project. These Operators are built with purpose with special operational knowledge to effectively manage Kafka. The Operators are involved in Deploying and running Kafka clusters and components, Configuring and securing access to Kafka, Upgrading and managing Kafka and even taking care of managing topics and users

Summary of Kafka Components:

1. Broker:
- Orchestrates the storage and transmission of messages.
- Uses Apache ZooKeeper for storing configuration data and cluster coordination.

2. ZooKeeper Cluster:
- Consists of replicated ZooKeeper instances.
- Used by brokers for configuration storage and cluster coordination.

3. Topic:
- Provides a destination for storing data.
- Split into one or more partitions.

4. Kafka Cluster:
- Group of broker instances working together.

5. Partition:
- Splits a single topic log into multiple logs.
- Each partition can reside on a separate node in the Kafka cluster.

6. Kafka Connect Cluster:
- Facilitates external data connections.

7. Kafka MirrorMaker Cluster:
- Replicates data between two Kafka clusters.
- Operates within or across data centers.

8. Kafka Exporter:
- Extracts additional Kafka metrics data for monitoring.
- Provides metrics related to offsets, consumer groups, consumer lag, and topics.

9. Kafka Bridge:
- Enables HTTP-based requests to the Kafka cluster.

10. Use Cases of Kafka:
- Kafka is commonly used for real-time event streaming.
- It serves as a distributed messaging system for large-scale data processing.
- Kafka supports log aggregation and data integration in various applications.

11. Kafka Uses:
- Data streaming and real-time analytics.
- Messaging and communication between microservices.
- Handling large-scale data processing and analytics.

Pre-requisites:

Docker
Kubernetes
kubectl command
AWS cloud services

https://github.com/A-LPHARM/K8s-apache-kafka-strimzi

Step 1: Download and Extract Debezium MySQL Connector Archive

Begin by downloading and extracting the Debezium MySQL Connector archive using the following command:

sudo curl https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.0.0.Final/debezium-connector-mysql-1.0.0.Final-plugin.tar.gz | tar xvz

Next, build your Docker image for Kafka Connect and the connector using the provided Dockerfile:

FROM strimzi/kafka:0.20.1-kafka-2.5.0
USER root:root
RUN mkdir -p /opt/kafka/plugins/debezium
COPY ./debezium-connector-mysql/ /opt/kafka/plugins/debezium/
USER 1001

Create Kafka Connector Using Docker Image

Use the Docker image to create your Kafka connector. Additionally, prepare a Dockerfile that incorporates the connector files into the Strimzi Kafka Connect image.

docker build . -t henriksin1/connect-debezium
docker push henriksin1/connect-debezium

Alternatively, utilize the specified Docker repository for the push operation.

deploy your kubenetes cluster using either EKS or Kops

eksctl create cluster — name test-cluster — version 1.21 — region us-east-1 — nodegroup-name linux-node — node-type t2.medium— nodes 2

kops create cluster — name henry-new-kops.k8s.local  \
 — cloud=aws — networking calico \
 — zones us-east-1a,us-east-1b,us-east-1c \
 — master-size t2.medium — node-size t2.medium \
 — node-count=3 — master-count=1

kops update cluster — name henry-new-kops.k8s.local — yes — admin

STEP 2

create namespace for all the deployments

kubectl apply -f 0-namespace.yaml

STEP 3

deploy strimzi operators from the operator lifecycle manager this simplifies the process to monitor the Kafka clusters

each operator manages Kafka and performs a separate function.

this installation deploys the custom resources definition CRD is an extension of the Kubernetes API that is not necessarily available in a default Kubernetes installation. It represents a customization of a particular Kubernetes installation.

basically we are customizing our resources for the operators to run effortlessly

to obtain the latest releases

curl -L https://github.com/operator-framework/operator-lifecycle-manager/releases/download/v0.26.0/install.sh -o install.sh
chmod +x install.sh
./install.sh v0.26.0

then you run

kubectl create -f https://operatorhub.io/install/strimzi-kafka-operator.yaml

step 4

deploy your secrets file this allows the connectors and Kafka have access into the data base and run the topics

cat <<EOF> debezium-mysql-credentials.properties 
mysql_username: debezium 
mysql_password: dbz 
EOF
kubectl -n kafka create secret generic debezium-secret — from-file=secrets.properties

step 5

deploy the role-back access control

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
 name: kafka-connector-configuration-role
 namespace: kafka
rules:
- apiGroups: [“”]
  resources: [“secrets”]
  resourceNames: [“debezium-secret”]
  verbs: [“get”]

kubectl apply -f rbac-debezium-role.yaml

step 6

deploy the service account to combine with the RBAC and secrets

apiVersion: v1
kind: ServiceAccount
metadata:
  name: debezium-connect-cluster-connect
  namespace: kafka

kubectl apply -f serviceaccount.yaml

step 7

deploy the rbac cluster binding resources which binds the serviceaccount to the role

https://github.com/A-LPHARM/K8s-apache-kafka-strimzi/blob/main/4-rbac-cluster-binding.yaml

kubectl apply -f rbac-cluster-binding.yaml

step 8

deploy the kafka cluster we are using the

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  namespace: kafka
spec:
  kafka:
    replicas: 1
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external
        port: 9094
        type: nodeport
        tls: false
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: '3.6'
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 100Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

kubectl apply -f kafka-kluster.yaml

the kafka cluster will deploy all the topic operators and user operators and zookeepers

note confirm the cluster is running

kubectl wait kafka/my-cluster — for=condition=Ready — timeout=300s -n kafka

then you execute commands within the Kafka cluster to create and inspect topics, checking if they are consumed.

To produce a topic

kubectl -n kafka exec my-cluster-kafka-0 -c kafka -i -t -- bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic my-topic

open another terminal to verify whether the topics are being consumed

kubectl -n kafka exec my-cluster-kafka-0 -c kafka -i -t -- bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning

Step 9:

After successfully deploying your Kafka cluster, proceed to deploy the MySQL database cluster. Once the database is deployed, apply the configuration from the ‘6-sql.yaml’ file using the following command:

https://github.com/A-LPHARM/K8s-apache-kafka-strimzi/blob/main/6-sql.yaml

kubectl apply -f 6-sql.yaml

To ensure the database is running and determine the endpoint port, execute:

kubectl describe service mysql -n kafka

Then, in another terminal, we can run the command line client:

exec into the database

kubectl exec -it mysql-6597659cb8-j9rk4 -n kafka -- sh

mysql -h mysql -u root -p
password:
mysql> use inventory;
mysql> show tables;

Utilize the obtained information to connect with your local MySQL Workbench.

once it confirms successful you prepare your kafka-connect manifest file

Step 10

Deploying the debezium-connect wasnt easy after so many logging and troubleshooting to sort out the networking. Deploying a self-managed service does require more effort in terms of setting up and maintaining the cluster

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: debezium-connect-cluster
  namespace: kafka
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.6.0
  image: henriksin1/connect-debezium:v1
  replicas: 1
  bootstrapServers: my-cluster-kafka-bootstrap:9092
  config:
    # config.providers: secret
    # config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    # -1 means it will use the default replication factor configured in the broker
    config.storage.replication.factor: 1
    offset.storage.replication.factor: 1
    status.storage.replication.factor: 1
    config.providers: file
    config.providers.file.class: org.apache.kafka.common.config.provider.FileConfigProvider
  externalConfiguration:
    volumes:
      - name: connector-config
        secret:
          secretName: debezium-secret

kubectl apply -f 7-kafka-connect.yaml

then you confirm all the pods are running

kubectl get all -n kafka

then run

kubectl logs debezium-connect-cluster-connect-0 -n kafka

check if the logs are good

note: this explains the configurations in the Kafka connect

the strimzi.io/use-connector-resources: “true” annotation tells the cluster operator that Kafka connector resources will be used to configure connectors within this Kafka Connect cluster.

The spec.image this image has to be built in your docker and deployed
the config.storage and config settings will use 1 as its replication factor because we created a single-broker Kafka cluster, more replicas we deploy more storage and config setting.
In the external configuration we’re referencing the secret we just created.

this diagram explains the movement of data from the database into the sink connector inside the kafka connect and how topics is sent to the kafka connector through the kafka connect API

then

Step 11.

create your kafka connector file

apiVersion: "kafka.strimzi.io/v1beta2"
kind: "KafkaConnector"
metadata:
  name: "debezium-connector-mysql"
  namespace: kafka
  labels:
    strimzi.io/cluster: debezium-connect-cluster
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  tasksMax: 1
  config:
    database.hostname: 52.90.119.28 #1
    database.port: 30036 #2
    database.user: root  #3
    database.password: debezium  #4
    database.server.id: "184054"
    database.whitelist: "inventory"
    database.server.name: "mysql"   #5
    database.history.kafka.bootstrap.servers: "my-cluster-kafka-bootstrap:9092"
    database.history.kafka.topic: "schema-changes.inventory"
    include.schema.changes: "true"
    database.allowPublicKeyRetrieval: "true"

In metadata.labels, strimzi.io/cluster names the kafkaconnect cluster which this connector will be created in.

The spec.class names the Debezium MySQL connector and spec.tasks must be 1 because that’s all this connector ever uses.

The config object contains the rest of the connector configuration. The explains

the database, hostname uses the ip of the mysql server as IP address for connecting to MySQL

2. The database port uses the ports which is gotten from the mysql database configuration which makes me use the node port

3. the file used for the database.user and database.passwrodis a placeholder which gets replaced with the referenced property from the given file in the secret we created.

4. the database.history.kafka: schema-changes.inventor configured debezium to use the schema-changes.inventory topic to store the database schema history

kubectl apply -f debezium-connector.yaml

the run

kubectl describe kafkaconnector debezium-connector-mysql -n kafka

once it shows you have connection you can run your topics

TIME FOR WORK

kubectl -n kafka exec my-cluster-kafka-0 -c kafka -i -t -- bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

this gives you the consumer results for every topic that’s given from the data base

return to the terminal having the sql data base thats is on hold

then input this command

mysql> SELECT * FROM customers;

in another terminal run

kubectl -n kafka exec my-cluster-kafka-0 -c kafka -i -t -- bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mysql --from-beginning

Deploying a scalable Kafka Architecture with an Sql database on Kubernetes Cluster