无法使用SSL在Kafka中设置inter.broker.listener.name

Question

我试图用SSL (TLS)在节点间以及节点和客户端之间配置Kafka节点，但是遇到了配置问题。Kafka版本是2.3.0。我的相关设置是:

      - KAFKA_BROKER_ID=1
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_LISTENERS=LISTENER_INTERNAL://kafka1:9092,LISTENER_EXTERNAL://kafka1:29092
      - KAFKA_ADVERTISED_LISTENERS=LISTENER_INTERNAL://kafka1:9092,LISTENER_EXTERNAL://localhost:29091
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=LISTENER_INTERNAL:SSL,LISTENER_EXTERNAL:SSL
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181
      - KAFKA_AUTO_CREATE_TOPICS_ENABLE=false
      - KAFKA_SSL_TRUSTSTORE_LOCATION=/var/private/ssl/server.truststore.jks
      - KAFKA_SSL_TRUSTSTORE_PASSWORD=changeit
      - KAFKA_SSL_KEYSTORE_LOCATION=/var/private/ssl/server.keystore.jks
      - KAFKA_SSL_KEYSTORE_PASSWORD=changeit
      - KAFKA_SSL_KEY_PASSWORD=changeit
      - KAFKA_SECURITY_INTER_BROKER_PROTOCOL=SSL
      - KAFKA_SSL_CLIENT_AUTH=required

仅供参考，为了简单起见，我从实例化Kafka容器的docker-compose文件中复制了设置。env vars将1:1映射到server.properties.中的属性。在容器启动期间，这些设置将应用于server.properties文件。

当我开始使用此配置时，我收到以下错误消息：

java.lang.IllegalArgumentException：要求失败：inter.broker.listener.name必须是advertised.listeners.中定义的侦听器名称基于当前配置的侦听器的有效选项LISTENER_INTERNAL，LISTENER_EXTERNAL

当我设置inter.broker.listener时。name属性设置为INTERNAL_LISTENER、SSL、null或空字符串，我收到以下错误消息：

org . Apache . Kafka . common . config . config exception:只能设置inter.broker.listener.name和security.inter.broker.protocol中的一个。

在这个问题上，我花了几个小时。我已经将我的设置与网络上应该演示具有SSL配置的Kafka的几个示例进行了比较。

知道吗？

Answer 1

我终于明白了如何拥有多个SSL侦听器。我将在这里记录这一点，以防其他人遇到同样的问题，因为多个SSL侦听器的工作示例似乎很少甚至不存在。下面是我的工作配置(只显示从docker-compose传递过来的相关属性):

ALLOW_PLAINTEXT_LISTENER=no
KAFKA_LISTENERS=ISSL://kafka1:9092,OSSL://kafka1:29092
KAFKA_ADVERTISED_LISTENERS=ISSL://kafka1:9092,OSSL://localhost:29092
KAFKA_INTER_BROKER_LISTENER_NAME=ISSL
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=ISSL:SSL,OSSL:SSL
KAFKA_SSL_CLIENT_AUTH=required

这样做的关键是不要配置KAFKA _ SECURITY _ INTER _ BROKER _ PROTOCOL，因为它与KAFKA _ INTER _ BROKER _ LISTENER _ NAME密钥是互斥的。

在多个听众的情况下，似乎需要KAFKA_LISTENER_SECURITY_PROTOCOL_MAP和KAFKA_INTER_BROKER_LISTENER_NAME的结合。