kubernetes kafka 연동 (with springboot)

kubernetes yaml 파일 작성시 띄어쓰기 조금만 잘못되도 object가 올라가지 않는다.

아래 사이트에서 yaml 파일 검증을 했다.

 

https://codebeautify.org/yaml-validator

Best YAML Validator Online

 

 

deployment는 pod를 생서하는 개념이고 service는 노출하는 개념이다.

 

apiVersion: apps/v1
kind: Deployment
metadata:
  name: zookeeper-deploy
  labels:
    app: zookeeper-1
spec:
  selector:
    matchLabels:
      app: zookeeper-1
  replicas: 2  
  template:
    metadata:
      labels:
        app: zookeeper-1        
    spec:
      containers:
        - name: zoo1
          image: digitalwonderland/zookeeper
          ports:
            - containerPort: 2181
          env:
            - name: ZOOKEEPER_ID
              value: "1"              
            - name: ZOOKEEPER_SERVER_1
              value: zoo1

zookeeper-deployment.yml

 

apiVersion: v1
kind: Service
metadata:
  name: zoo1
  labels:
    app: zookeeper-1
spec:
  ports:
    - name: client
      port: 2181
      protocol: TCP
    - name: follower
      port: 2888
      protocol: TCP
    - name: leader
      port: 3888
      protocol: TCP
  selector: 
      app: zookeeper-1

zookeeper-service.yml

 

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-broker0
spec:
  selector:
    matchLabels:
      app: kafka
      id : "0"
  replicas: 2  
  template:
    metadata:
      labels:
        app: kafka
        id: "0"
    spec:
      containers:
      - name: kafka
        image: wurstmeister/kafka
        ports:
        - containerPort: 9092
        env:
        - name: KAFKA_ADVERTISED_LISTENERS
          value: PLAINTEXT://localhost:9092
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: zoo1:2181
        - name: KAFKA_CREATE_TOPICS
          value: go-kafka
        - name: KAFKA_ADVERTISED_HOST_NAME
          value: localhost
        - name: KAFKA_BROKER_ID
          value: "0"

kafka-deployment.yml

apiVersion: v1
kind: Service
metadata:
    name: kafka-service
    labels:
      name: kafka
spec:
    ports:
    - port: 9092
      name: kafka-port
      protocol: TCP
    selector:
      app: kafka
      id: "0"
    type: LoadBalancer

kafka-service.yml

 

// 실행 순서
$kubectl create -f zookeeper-deployment.yml
$kubectl create -f zookeeper-service.yml
$kubectl create -f kafka-deployment.yml
$kubectl create -f kafka-service.yml

 

---

SpringBoot

build.gradle

 implementation 'org.springframework.kafka:spring-kafka'

 

application.yml

spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
      group-id: test-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      bootstrap-servers: localhost:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

 

KafkaConsumer.java

@Service
public class KafkaConsumer {

    @KafkaListener(topics = "go-kafka", groupId = "test-group")
    public void consume(String message) throws IOException {
        System.out.println(String.format("Consumed message : %s", message));
    }
}

 

KafkaProducer.java

@Service
public class KafkaProducer {
    private static final String TOPIC = "go-kafka";
    private final KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public KafkaProducer(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(String message) {
        System.out.println(String.format("Produce message : %s", message));
        this.kafkaTemplate.send(TOPIC, message);
    }
}

 

 

KafkaController.java

@RestController
@RequestMapping("/kafka")
public class KafkaController {

    private final KafkaProducer producer;

    @Autowired
    KafkaController(KafkaProducer producer) {
        this.producer = producer;
    }

    @GetMapping
    public String sendMessage() {
        this.producer.sendMessage("test");
        return "SUCCESS";
    }
}