KAFKA без ZOOKEEPER або добрий день KraFT

Треба настроїти кафку, без ZOOKEEPER та з авторихацією Намучався з останноьою версією Кафки но усе вийшло. Качаємо кафку, розпаковуємо та створюємо каталоги

wget https://downloads.apache.org/kafka/3.5.2/kafka_2.12-3.5.2.tgz
tar -xvf kafka
mkdir -R /opt/kafka/data
mkdir /opt/kafka/logs_metadata
cd kafka_2.12-3.2
mv kafka /opt
mv server.properties server.properties.orig
touch server.properties

Добавляємо конфіг:

#/opt/kafka/config/kraft/server.properties
node.id=1
num.network.threads=3
num.io.threads=8
log.dirs=/opt/kafka/logs
metadata.log.dir=/opt/kafka/logs_metadata


process.roles=broker,controller
listeners=BROKER://192.168.232.130:9092,CONTROLLER://192.168.232.130:9093
advertised.listeners=BROKER://192.168.232.130:9092
listener.security.protocol.map=BROKER:SASL_PLAINTEXT,CONTROLLER:SASL_PLAINTEXT
controller.quorum.voters=1@192.168.232.130:9093

inter.broker.listener.name=BROKER
controller.listener.names=CONTROLLER

sasl.enabled.mechanisms=PLAIN
sasl.mechanism.controller.protocol=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

#authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:admin

delete.topic.enable=true
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600

num.partitions=3
default.replication.factor=2
min.insync.replicas=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
auto.create.topics.enable=true
unclean.leader.election.enable=false

authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer

listener.name.broker.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="admin" \
    password="admin" \
    user_admin="admin";

listener.name.controller.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="admin" \
    password="admin" \
    user_admin="admin";

############################

створюємо jaas.config

#/opt/kafka/config/kraft/jaas.config
KafkaServer {
 org.apache.kafka.common.security.plain.PlainLoginModule required
 username="admin"
 password="admin"
 user_admin="admin"
 user_usera="usera"
 user_userb="userb";
};

Та конфігурацію для локальної перевірки скриптів створюємо admin.config

#/opt/kafka/config/kraft/admin.config
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

Додаємо до автозагрузки

touch /etc/systemd/system/kafka.service

#/etc/systemd/system/kafka.service
[Unit]

Description=kafka Service
After=network-online.target
Requires=network-online.target


[Service]
Type=root
User=root
SyslogIdentifier=kafka
Environment="KAFKA_HEAP_OPTS=-Xms1G -Xmx1G"
Environment="KAFKA_OPTS=-Djava.security.auth.login.config=/opt/kafka/config/kraft/jaas.config"
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh /opt/kafka/config/kraft/server.properties
WorkingDirectory=/opt/kafka

Restart=on-failure

[Install]
WantedBy=multi-user.target

Створюємо Идентифікатор кластера:

KAFKA_CLUSTER_ID="$(/opt/kafka/bin/kafka-storage.sh random-uuid)"
/opt/kafka/bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c /opt/kafka/config/kraft/server.properties

Запускаемо:

systemctl daemon-reload
systemctl enable kafka
systemctl start kafka

Пробуємо створити та потім вивести ТОПІКИ

/opt/kafka/bin/kafka-topics.sh --bootstrap-server 192.168.232.130:9092 --create --topic newtopic --partitions 1 --replication-factor 1 --command-config /opt/kafka/config/kraft/admin.config

/opt/kafka/bin/kafka-topics.sh --bootstrap-server 192.168.232.130:9092 --list --command-config /opt/kafka/config/kraft/admin.config

А тепер надати користувачу доступ до топіка

/opt/kafka/bin/kafka-acls.sh --bootstrap-server 192.168.232.130:9092 --command-config /opt/kafka/config/kraft/admin.config --add --allow-principal User:userсс --operation All --topic newtopic

Для візуального відображення можно скористатись kafka-ui

wget https://github.com/provectus/kafka-ui/releases/download/v0.7.2/kafka-ui-api-v0.7.2.jar

Робимо конфіг

#application-local.yml
auth:
  type: DISABLED

management:
  endpoint:
    info:
      enabled: true
    health:
      enabled: true
  endpoints:
    web:
      exposure:
        include: "info,health,prometheus"

spring:
  jmx:
    enabled: true

logging:
  level:
    root: INFO
    com.provectus: DEBUG
    reactor.netty.http.server.AccessLog: INFO
    org.hibernate.validator: WARN


kafka:
  clusters:
    - name: local
      bootstrapServers: 192.168.232.130:9092
      schemaRegistry: http://192.168.232.130:8085
      schemaRegistryAuth:
        username: admin
        password: admin
      ksqldbServer: http://192.168.232.130:8088
      kafkaConnect:
        - name: first
          address: http://192.168.232.130:8083
      metrics:
        port: 9997
        type: JMX

dynamic.config.enabled: true
  

java -Dspring.config.additional-location=application-local.yml -Djava.security.auth.login.config=/opt/kafka/config/kraft/jaas.config --add-opens java.rmi/javax.rmi.ssl=ALL-UNNAMED -jar kafka-ui-api-v0.7.2.jar 

---

upd: перенес на прод и перестал работать JMX
для этого исправил строку заодно и перенес на порт 9994 
в 
#/etc/systemd/system/kafka.service
вместо 
Environment="KAFKA_OPTS=-Djava.security.auth.login.config=/opt/kafka/config/kraft/jaas.config"
на 
Environment="KAFKA_OPTS=-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9994 -Dcom.sun.management.jmxremote.rmi.port=9994 -Dcom.sun.management.jmxremote.authenticate=false -Djava.security.auth.login.config=/opt/kafka/config/kraft/jaas.config"

Cоответственно и  kafka-ui надо  изменить:
        port: 9997
на 
        port: 9994

а если запускали kafka-ui и чтото меняли в конфиге то еще и в локальном динамическом конфиге.
/etc/kafkaui/dynamic_config.yaml
        port: 9997
на 
        port: 9994

Усе працює й на 3.7.0

 wget https://downloads.apache.org/kafka/3.7.0/kafka_2.12-3.7.0.tgz