如何在kubernetes中创作简单的Kafka

Question

apiVersion: v1
kind: Service
metadata:
  name: kafka-service
  labels:
    name: kafka
spec:
  ports:
  - port: 9092
    targetPort: 9092
    protocol: TCP
  selector:
    name: kafka
---
apiVersion: v1
kind: Service
metadata:
  name: zookeeper-service
  labels:
    name: zookeeper
spec:
  ports:
  - name: client
    port: 2181
    protocol: TCP
  - name: follower
    port: 2888
    protocol: TCP
  - name: leader
    port: 3888
    protocol: TCP
  selector:
    name: zookeeper
  type: LoadBalancer
---
apiVersion: v1
kind: Pod
metadata:
  name: zookeeper
  labels:
    name: zookeeper 
spec:
  containers:
    - name: zookeeper
      image: zookeeper:3.7.0
---
apiVersion: v1
kind: Pod
metadata:
  name: kafka
  labels:
    name: kafka 
spec:
  containers:
    - name: kafka
      image: wurstmeister/kafka:2.13-2.6.0
      imagePullPolicy: "IfNotPresent"
      env:
        - name: KAFKA_ADVERTISED_PORT
          value: "666"
        - name: KAFKA_ADVERTISED_HOST_NAME
          value: localhost
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: zookeeper-service:2181
        - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
          value: INSIDE:PLAINTEXT
        - name: KAFKA_ADVERTISED_LISTENERS
          value: INSIDE://:666
        - name: KAFKA_LISTENERS
          value: INSIDE://:666
        - name: KAFKA_INTER_BROKER_LISTENER_NAME
          value: INSIDE
      ports:
        - containerPort: 9092

简单的java代码：

@Service
public class Consumer {
    @KafkaListener(topics = "new-topic",groupId = "test")
    public void consumeMessage(String message){
        System.out.println("************************");
        System.out.println(message);
        System.out.println("************************");
    }
}

如何创建kafka broker而不需要扩展测试？

Answer 1

这篇文章对我很有帮助：https://www.confluent.io/blog/kafka-listeners-complinated/

我将yaml文件更改为：

apiVersion: v1
kind: Pod
metadata:
  name: kafka-service
  labels:
    name: kafka-service 
spec:
  containers:
    - name: kafka-service
      image: wurstmeister/kafka:2.13-2.6.0
      imagePullPolicy: "IfNotPresent"
      env:
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: zookeeper-service:2181
        - name: KAFKA_BROKER_ID
          value: "1"

        - name: KAFKA_LISTENERS
          value: IN://:9092,OUT://:9093
        - name: KAFKA_ADVERTISED_LISTENERS
          value: IN://localhost:9092,OUT://kafka-service:9093
        - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
          value: IN:PLAINTEXT,OUT:PLAINTEXT
        - name: KAFKA_INTER_BROKER_LISTENER_NAME
          value: IN

这意味着kubernetes network内的端口9093和端口9092从主机打开。之后一切正常工作。