---
apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-config
  namespace: mid
#/kafka/config/server.properties
data:
  start-kafka.sh: |
    #!/bin/bash
    echo 44m6ypTXR4iJCJ9Boc9MbQ > /kafka/uuid/uuid.txt
    cat>/kafka/config/server.properties<<EOF
    process.roles=broker,controller
    node.id=${HOSTNAME##*-}
    controller.quorum.voters=0@kafka-0.kafka-headless:9093,1@kafka-1.kafka-headless:9093,2@kafka-2.kafka-headless:9093
    # 这里kafka启用了两个监听器，k8s集群内直接使用INSIDE访问，因为podip会变而域名不会改变所以重启kafka的时候也不会导致logstash连接不上
    listeners=INSIDE://0.0.0.0:9091,CONTROLLER://0.0.0.0:9093,OUTSIDE://0.0.0.0:9092
    inter.broker.listener.name=INSIDE
    advertised.listeners=INSIDE://${HOSTNAME}.kafka-headless:9091,OUTSIDE://${HOST_IP}:9092
    #advertised.listeners=PLAINTEXT://${HOST_IP}:9092
    controller.listener.names=CONTROLLER
    listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
    num.network.threads=16
    num.io.threads=16
    #socket.send.buffer.bytes=102400
    #socket.receive.buffer.bytes=102400
    socket.request.max.bytes=209715200
    log.dirs=/kafka/combined-logs
    num.partitions=3
    num.recovery.threads.per.data.dir=1
    offsets.topic.replication.factor=3
    transaction.state.log.replication.factor=3
    transaction.state.log.min.isr=1
    log.retention.hours=48
    #log.segment.bytes=1073741824
    #log.retention.check.interval.ms=300000
    delete.topic.enable=true
    host.name=${HOST_IP}
    EOF
    /kafka/bin/kafka-storage.sh format -t `cat /kafka/uuid/uuid.txt` -c $KAFKA_HOME/config/server.properties
    /kafka/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties