在多个网络上使用 kafka

Question

我在不同的 VMS 上有一个 kafka 集群。 我希望能够从来自 2 个不同网络（10.X.X.X 和 172.x.X.x）的主题中生成和消费。 我尝试配置不同的解决方案，例如在侦听器或配置中不指定任何内容，但似乎没有任何效果，有人可以建议什么是正确的配置吗？

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# see kafka.server.KafkaConfig for additional details and defaults

############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=3

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://0.0.0.0:9092

# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=PLAINTEXT://172.X.X.X:9092

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

还尝试配置以下选项：

# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners= EXTERNAL://172.X.X:9093,INTERNAL://10.20.4.234:9092

# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners= EXTERNAL://172.X.X.X:9093,INTERNAL://10.X.X.X:9092

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
    listener.security.protocol.map=PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT

我不断收到以下错误：

[2021-03-30 08:30:02,798] ERROR Exiting Kafka due to fatal exception (kafka.Kafka$)
java.lang.IllegalArgumentException: requirement failed: inter.broker.listener.name must be a listener name defined in advertised.listeners. The valid options based on currently configured listeners are EXTERNAL,INTERNAL
        at scala.Predef$.require(Predef.scala:277)
        at kafka.server.KafkaConfig.validateValues(KafkaConfig.scala:1399)
        at kafka.server.KafkaConfig.<init>(KafkaConfig.scala:1374)
        at kafka.server.KafkaConfig.<init>(KafkaConfig.scala:1063)
        at kafka.server.KafkaConfig$.fromProps(KafkaConfig.scala:1043)
        at kafka.server.KafkaServerStartable$.fromProps(KafkaServerStartable.scala:28)
        at kafka.Kafka$.main(Kafka.scala:59)
        at kafka.Kafka.main(Kafka.scala)

Answer 1

最终我以这种方式配置它并且它可以工作：

 The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=EXTERNAL://172.X.X.X:9093,INTERNAL://10.X.X.X:9092

# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
advertised.listeners=EXTERNAL://172.X.X.X:9093,INTERNAL://10.X.X.X:9092

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
listener.security.protocol.map=EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
inter.broker.listener.name=INTERNAL