Deploying Kafka on minikube

Deploy Kafka on minikube without IP hack for local development.

Photo by John Schnobrich on Unsplash

Problem: If you are struggling with installing and running Kafka on minikube then you are at the right place.

The solution here is to create a “headless service” for brokers.

Why only headless service and not others?

Let’s first understand what services do in Kubernetes.

Service and PODs

Kubernetes services act as proxies/wrappers/LB to ephemeral PODs. Now when we query the service object we get the address of the proxy and not the actual POD address where the broker is running.

For the Kafka client to connect successfully to the broker we need the broker's address as shown below:

Kafka client broker communication

A Kafka client(Producer/Consumer) wants to connect to the actual broker and not a proxy. So how to get the brokers' address and not the services' address? There can be two approaches here,

1. Use Kubernetes API Server to get the address for Broker (custom)

2. Use OOTB Headless service

In this article, we are going ahead with a headless approach. So what is headless service?

A headless service is a service that does not allocate cluster IP and no load balancing or proxying is done. Unlike other service objects when headless service is queried, it returns IP’s of POD’s(broker) which can be then directly consumed by the Kafka client.

How to create a headless service?

Create a service for the broker and set the service type as clusterIP and clusterIP none as shown below,

clusterIP: None → Important

type: ClusterIP

Let’s check it practically now

Pre-requisites :

Spring Boot version: 2.7.3

kafka version: 3.2.0

minikube version: v1.25.1

Kubectl: v1.22.5

Step 1 : Create a zoo-keeper POD.

zookeeper.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    app: zookeeper-service
  name: zookeeper-service
spec:
  type: NodePort
  ports:
    - name: zookeeper-port
      port: 2181
      nodePort: 30181
      targetPort: 2181
  selector:
    app: zookeeper
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: zookeeper
  name: zookeeper
spec:
  replicas: 1
  selector:
    matchLabels:
      app: zookeeper
  template:
    metadata:
      labels:
        app: zookeeper
    spec:
      containers:
        - image: wurstmeister/zookeeper
          imagePullPolicy: IfNotPresent
          name: zookeeper
          ports:
            - container Port: 2181

Using kubectl apply -f zookeeper.yaml deploy it.

Step 2 : Create a kafka-broker POD.

kafka-broker.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka-service
  name: kafka-service
spec:
  ports:
   - name: kafka-port
     port: 9092
     protocol: TCP
     targetPort: 9092
  selector:
    app: kafka-service
  clusterIP: None
  type: ClusterIP
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: kafka-service
  name: kafka-service
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-service
  template:
    metadata:
      labels:
        app: kafka-service
    spec:
      containers:
      - env:
        - name: KAFKA_BROKER_ID
          value: "1"
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: zookeeper-service:2181
        - name: KAFKA_LISTENERS
          value: LISTENER_INTERNAL://kafka-service:9092,LISTENER_EXTERNAL://localhost:9093
        - name: KAFKA_ADVERTISED_LISTENERS
          value: LISTENER_INTERNAL://kafka-service:9092,LISTENER_EXTERNAL://localhost:9093
        - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
          value: LISTENER_INTERNAL:PLAINTEXT,LISTENER_EXTERNAL:PLAINTEXT
        - name: KAFKA_INTER_BROKER_LISTENER_NAME
          value: LISTENER_INTERNAL
        image: wurstmeister/kafka
        imagePullPolicy: IfNotPresent
        name: kafka-service
        ports:
        - containerPort: 9092

IMP (Solution): As you can see service type is ClusterIP and clusterIP: None

Using Kubectl apply -f kafka-broker.yaml deploy it.

Step 3 : Create a SpringBoot Application using spring starter io name it — order-service,

3.1 Dependency add the following 2 dependencies

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
   <groupId>org.springframework.kafka</groupId>
   <artifactId>spring-kafka</artifactId>
   <version>3.2.0</version>
</dependency>

3.2 Create MessageProducer class

@Component
public class MessageProducer {
private final KafkaTemplate<String, String> kafkaTemplate;
public MessageProducer(KafkaTemplate<String, String> kafkaTemplate) {
   this.kafkaTemplate = kafkaTemplate;
}
public void send(String message) {
   this.kafkaTemplate.send("order-topic", message);
}
}

3.3 Create Rest Endpoint to post message

@SpringBootApplication
@RestController
public class OrderServiceApplication {
@Autowired
Environment env;
@Autowired
MessageProducer messageProducer;
public static void main(String[] args) {
SpringApplication.run(OrderServiceApplication.class, args);
}
@PostMapping("/notify")
public String sendNotifation(@RequestBody String message){
System.out.println("sending message : " + message);
messageProducer.send(message);
System.out.println("Sent message : " + message);
return message;
}
}

3.4 Specify bootstrap server in application.yaml

spring:
  application:
    name: order-service
  kafka:
    bootstrap-servers: kafka-service:9092
    consumer:
      group-id: "myGroup"

Step 4 : Create a SpringBoot Application using spring starter io name it — notification-service,

4.1 : Follow Step 3.1 to add dependencies

4.2 : Create a consumer

@SpringBootApplication
public class NotificationApplication {
public static void main(String[] args) {        SpringApplication.run(NotificationApplication.class, args);
}
@KafkaListener(id = "myGroup", topics = "order-topic")
public void listen(String in) {
  //Logic to consume/process message goes here
  System.out.println(in);
}
}

Step 5 : Test

Check Verify all pods are up and running

kubectl get pods

OR

using k8slens

Producer

First, we need to expose service to our local m/c using the following command,

kubectl port-forward <order-service-pod-name> 8080:8080

Post message

Consumer

kubectl logs -f <notification-service-pod-name>

Conclusion

We have successfully deployed Kafka on minikube and using Spring Producer and Consumer we are able to send and receive messages.

#InvalidReplicationfactor #UnknownHostException #Kafka #Minikube #SpringBoot

References :

Spring for Apache Kafka

Kafka Listeners - Explained | Confluent

Why Stateful Applications in K8s need a headless service