Flume 连接拒绝 Kafka 代理

Question

我正在尝试将消息从 Apache Flume 发送到我的 Apache Kafka 实例。通过 localhost 在本地执行所有操作时，我没有任何问题。当我从机器上的不同虚拟机尝试此操作时，我在 Flume 的调试日志中被拒绝连接。就演示而言，我只是通过 telnet 向我的 Flume 实例发送消息。

172.16.26.1 是虚拟机看到的我的 mac 的 IP

172.16.26.138 是运行 Kafka/Zookeeper 的 VM 的 IP

172.16.26.139是运行Flume的VM的IP

Flume 配置

    #Name the components of this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1

    #Describe/configure the source
    a1.sources.r1.type = netcat 
    a1.sources.r1.port = 44444
    a1.sources.r1.bind = 0.0.0.0 
    a1.sources.r1.host = 172.16.26.1 

    #Describe the sink
    a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
    a1.sinks.k1.kafka.topic = macdemo
    a1.sinks.k1.kafka.bootstrap.servers = 172.16.26.138:9092
    a1.sinks.k1.kafka.flumeBatchSize = 20
    a1.sinks.k1.kafka.producer.acks = 1
    a1.sinks.k1.kafka.producer.linger.ms = 1
    a1.sinks.k1.kafka.producer.compression.type = snappy

    #Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100

    #Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

Kafka配置（关键部分，其他全部默认）

    broker.id=0
    listeners=PLAINTEXT://localhost:9092
    zookeeper.connect=localhost:2181

这是代理启动时从 Flume 打印出来的配置

    compression.type = snappy
    metric.reporters = []
    metadata.max.age.ms = 300000
    metadata.fetch.timeout.ms = 60000
    reconnect.backoff.ms = 50
    sasl.kerberos.ticket.renew.window.factor = 0.8
    bootstrap.servers = [172.16.26.138:9092]
    retry.backoff.ms = 100
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    buffer.memory = 33554432
    timeout.ms = 30000
    key.serializer = class org.apache.kafka.common.serialization.StringSerializer
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    ssl.keystore.type = JKS
    ssl.trustmanager.algorithm = PKIX
    block.on.buffer.full = false
    ssl.key.password = null
    max.block.ms = 60000
    sasl.kerberos.min.time.before.relogin = 60000
    connections.max.idle.ms = 540000
    ssl.truststore.password = null
    max.in.flight.requests.per.connection = 5
    metrics.num.samples = 2
    client.id = 
    ssl.endpoint.identification.algorithm = null
    ssl.protocol = TLS
    request.timeout.ms = 30000
    ssl.provider = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    acks = 1
    batch.size = 16384
    ssl.keystore.location = null
    receive.buffer.bytes = 32768
    ssl.cipher.suites = null
    ssl.truststore.type = JKS
    security.protocol = PLAINTEXT
    retries = 0
    max.request.size = 1048576
    value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
    ssl.truststore.location = null
    ssl.keystore.password = null
    ssl.keymanager.algorithm = SunX509
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    send.buffer.bytes = 131072
    linger.ms = 1

可行的场景

Flume/Kafka/Zookeeper 都在 Mac 本地运行 -> 工作

VM 中的 Flume（在 Mac 上），Kafka/Zookeeper 从 Mac 本地运行 -> 工作

VM 中的 Flume（Mac 上），VM 中的 Kafka/Zookeeper（Mac 上）-> 连接被拒绝

这是我得到的输出错误

    2018-02-26 14:23:47,009 (kafka-producer-network-thread | producer-1) [DEBUG - org.apache.kafka.common.network.Selector.poll(Selector.java:307)]            Connection with /172.16.26.138 disconnected
    java.net.ConnectException: Connection refused
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
    at org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:54)
    at org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:72)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:274)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:128)
    at java.lang.Thread.run(Thread.java:748)
    2018-02-26 14:23:47,010 (kafka-producer-network-thread | producer-1) [DEBUG - org.apache.kafka.clients.NetworkClient.handleDisconnections(NetworkClient.java:454)] Node -1 disconnected.
    2018-02-26 14:23:47,011 (kafka-producer-network-thread | producer-1) [DEBUG - org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.maybeUpdate(NetworkClient.java:608)] Give up sending metadata request since no node is available

我确信我忽略了一些超级简单的事情，但是已经花了几天时间没有得到 Google 博士的运气。关于虚拟机，这些是在 Centos 7 上运行的，我禁用了 firewalld.service 和 SELinux。

Answer 1

我发现的问题与 listeners=PLAINTEXT://localhost:9092 设置有关。将 localhost 更改为 VM 的实际 IP 地址（kafka 的 IP）有效。