kafka SASL / SCRAM Error de autenticación

Traté de agregar seguridad a mi clúster kafka, seguí la documentación:

• https://kafka.apache.org/documentation/#security_sasl_scram
• https://docs.confluent.io/current/kafka/authentication_sasl/authentication_sasl_scram.html#

Agrego el usuario usando esto:

kafka-configs.sh --zookeeper zookeeper1:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin-secret],SCRAM-SHA-512=[password=admin-secret]' --entity-type users --entity-name admin

Modifico el server.properties:

broker.id=1
listeners=SASL_PLAINTEXT://kafka1:9092
advertised.listeners=SASL_PLAINTEXT://kafka1:9092
sasl.enabled.mechanisms=SCRAM-SHA-256
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
default.replication.factor=3
min.insync.replicas=2
log.dirs=/var/lib/kafka
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181/kafka
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

Creó el archivo jaas:

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin-secret"
};

Creó el archivo kafka_opts.sh en /etc/profile.d:

export KAFKA_OPTS=-Djava.security.auth.login.config=/opt/kafka_2.12-2.5.0/config/kafka_server_jaas.conf

Pero cuando inicio kafka arroja el siguiente error:

[2020-05-04 10:54:08,782] INFO [Controller id=1, targetBrokerId=1] Failed authentication with kafka1/kafka1 (Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256) (org.apache.kafka.common.network.Selector)

Utilizo en lugar de kafka1, kafka2, kafka3, zookeeper1, zookeeper2 y zookeeper3 respectivamente la ip de cada servidor, ¿puede alguien ayudarme con mi problema?

Mi principal problema fue esta configuración:

zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181/kafka

Esta configuración en server.properties era necesaria para tener un orden en la forma en que zookeeper creaba la información de kafka, pero eso afecta la forma en que necesito ejecutar el comando kafka-configs.sh, así que explicaré los pasos que necesitaba seguir.

1. Primero modifique el guardián del zoológico.

He descargado zookeeper del sitio oficial https://zookeeper.apache.org/releases.html

Modifiqué el archivo zoo.cfg y agregué la configuración para la seguridad:

tickTime=2000
dataDir=/var/lib/zookeeper/
clientPort=2181
initLimit=5
syncLimit=2
server.1=zookeeper1:2888:3888
server.2=zookeeper2:2888:3888
server.3=zookeeper3:2888:3888
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl

Creo el archivo jaas para zookeeper:

Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    user_admin="admin_secret";
};

Creo el archivo java.env en / conf / y agregué lo siguiente:

SERVER_JVMFLAGS="-Djava.security.auth.login.config=/opt/apache-zookeeper-3.6.0-bin/conf/zookeeper_jaas.conf"

Con estos archivos, le está diciendo al guardián del zoológico que use el archivo jaas para permitir que kafka se autentique ante el guardián del zoológico, para validar que guardián del zoológico está tomando el archivo que solo necesita ejecutar:

zkServer.sh print-cmd

responderá:

/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /opt/apache-zookeeper-3.6.0-bin/bin/../conf/zoo.cfg
"java"  -Dzookeeper.log.dir="/opt/apache-zookeeper-3.6.0-bin/bin/../logs" ........-Djava.security.auth.login.config=/opt/apache-zookeeper-3.6.0-bin/conf/zookeeper_jaas.conf....... "/opt/apache-zookeeper-3.6.0-bin/bin/../conf/zoo.cfg" > "/opt/apache-zookeeper-3.6.0-bin/bin/../logs/zookeeper.out" 2>&1 < /dev/null

1. Modificar kafka

He descargado kafka del sitio oficial https://www.apache.org/dyn/closer.cgi?path=/kafka/2.5.0/kafka_2.12-2.5.0.tgz

Modifiqué / agregué la siguiente configuración en el archivo server.properties:

listeners=SASL_PLAINTEXT://kafka1:9092
advertised.listeners=SASL_PLAINTEXT://kafka1:9092
sasl.enabled.mechanisms=SCRAM-SHA-256
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:admin

Creé el archivo jaas para kafka:

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin_secret";
};
Client {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   username="admin"
   password="admin_secret";
};

Una cosa importante que debe comprender, la parte del Cliente debe ser la misma que el archivo jaas en zookeeper y la parte de KafkaServer es para la comunicación entre agentes.

También necesito decirle a kafka que use el archivo jaas, esto se puede hacer configurando la variable KAFKA_OPTS:

export KAFKA_OPTS=-Djava.security.auth.login.config=/opt/kafka_2.12-2.5.0/config/kafka_server_jaas.conf

1. Cree el administrador de usuario para los brokers de kafka

Ejecute el siguiente comando:

kafka-configs.sh --zookeeper zookeeper:2181/kafka --alter --add-config 'SCRAM-SHA-256=[password=admin_secret]' --entity-type users --entity-name admin

Como mencioné antes, mi error fue que no estaba agregando la parte / kafka a la ip del guardián del zoológico (tenga en cuenta que todo lo que use el guardián del zoológico deberá agregar la parte / kafka al final de la ip), ahora si inicia zookeeper y kafka todo va a funcionar muy bien.