Deploying Kafka on minikube
Deploy Kafka on minikube without IP hack for local development.
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.
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:
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
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 :