Mac 上的 Kafka 启动控制台消费者时没有对节点 -1 进行中的请求

Question

我是 Kafka 新手，尝试安装并运行控制台消费者，但出现错误java.lang.IllegalStateException: There are no in-flight requests for node -1

我尝试的环境如下

Kafka Version kafka_2.13-2.6.0
MacOS java11 Fails
MacOS java 1.8 Fails
Windows 10 Java11 Success

以下是我正在执行的详细步骤。相同的步骤适用于 Windows。

步骤 1 下载卡夫卡

我刚刚从https://www.apache.org/dyn/closer.cgi?path=/kafka/2.6.0/kafka_2.13-2.6.0.tgz下载了 Kafka

STEP 2 启动zookeeper服务，工作正常。

我用以下命令启动动物园管理员

bin % zookeeper-server-start.sh ../config/zookeeper.properties

这是动物园管理员日志

[2020-08-30 14:28:52,234] INFO Server environment:java.io.tmpdir=/var/folders/q7/khp8p9k14rzfs_zl52m57hlr0000gn/T/ (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:java.compiler=<NA> (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:os.name=Mac OS X (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:os.arch=x86_64 (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:os.version=10.15.6 (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:user.name=jigarnaik (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:user.home=/Users/jigarnaik (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:user.dir=/Users/jigarnaik/Documents/kafka_2.13-2.6.0/bin (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:os.memory.free=496MB (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:os.memory.max=512MB (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,234] INFO Server environment:os.memory.total=512MB (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,235] INFO minSessionTimeout set to 6000 (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,235] INFO maxSessionTimeout set to 60000 (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,236] INFO Created server with tickTime 3000 minSessionTimeout 6000 maxSessionTimeout 60000 datadir /tmp/zookeeper/version-2 snapdir /tmp/zookeeper/version-2 (org.apache.zookeeper.server.ZooKeeperServer)
[2020-08-30 14:28:52,243] INFO Using org.apache.zookeeper.server.NIOServerCnxnFactory as server connection factory (org.apache.zookeeper.server.ServerCnxnFactory)
[2020-08-30 14:28:52,247] INFO Configuring NIO connection handler with 10s sessionless connection timeout, 2 selector thread(s), 32 worker threads, and 64 kB direct buffers. (org.apache.zookeeper.server.NIOServerCnxnFactory)
[2020-08-30 14:28:52,254] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)
[2020-08-30 14:28:52,268] INFO zookeeper.snapshotSizeFactor = 0.33 (org.apache.zookeeper.server.ZKDatabase)
[2020-08-30 14:28:52,271] INFO Snapshotting: 0x0 to /tmp/zookeeper/version-2/snapshot.0 (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2020-08-30 14:28:52,273] INFO Snapshotting: 0x0 to /tmp/zookeeper/version-2/snapshot.0 (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2020-08-30 14:28:52,286] INFO Using checkIntervalMs=60000 maxPerMinute=10000 (org.apache.zookeeper.server.ContainerManager)
load: 1.58  cmd: java 17764 waiting 0.68u 0.11s

第 3 步启动 Kafka - 日志对我来说看起来很好。

之后我使用下面的命令启动 Kafka

bin % kafka-server-start.sh ../config/server.properties

Kafka 启动看起来不错，日志尾部如下

[2020-08-30 14:33:41,708] INFO [TransactionCoordinator id=0] Starting up. (kafka.coordinator.transaction.TransactionCoordinator)
[2020-08-30 14:33:41,709] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
[2020-08-30 14:33:41,709] INFO [TransactionCoordinator id=0] Startup complete. (kafka.coordinator.transaction.TransactionCoordinator)
[2020-08-30 14:33:41,728] INFO [ExpirationReaper-0-AlterAcls]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-08-30 14:33:41,757] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2020-08-30 14:33:41,770] INFO [SocketServer brokerId=0] Starting socket server acceptors and processors (kafka.network.SocketServer)
[2020-08-30 14:33:41,773] INFO [SocketServer brokerId=0] Started data-plane acceptor and processor(s) for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2020-08-30 14:33:41,773] INFO [SocketServer brokerId=0] Started socket server acceptors and processors (kafka.network.SocketServer)
[2020-08-30 14:33:41,775] INFO Kafka version: 2.6.0 (org.apache.kafka.common.utils.AppInfoParser)
[2020-08-30 14:33:41,775] INFO Kafka commitId: 62abe01bee039651 (org.apache.kafka.common.utils.AppInfoParser)
[2020-08-30 14:33:41,775] INFO Kafka startTimeMs: 1598769221773 (org.apache.kafka.common.utils.AppInfoParser)
[2020-08-30 14:33:41,776] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
[2020-08-30 14:33:41,808] INFO [ReplicaFetcherManager on broker 0] Removed fetcher for partitions Set(word-count-output-0, word-count-input-0) (kafka.server.ReplicaFetcherManager)
[2020-08-30 14:33:41,815] INFO [Partition word-count-output-0 broker=0] Log loaded for partition word-count-output-0 with initial high watermark 0 (kafka.cluster.Partition)
[2020-08-30 14:33:41,821] INFO [Partition word-count-input-0 broker=0] Log loaded for partition word-count-input-0 with initial high watermark 0 (kafka.cluster.Partition)

STEP 4 创建主题，主题创建成功。之后我正在创建主题

bin % kafka-topics.sh \
    --create \
    --zookeeper localhost:2181 \
    --replication-factor 1 \
    --partitions 1 \
    --topic my-topic
Created topic my-topic.

列出主题

bin % kafka-topics.sh \
    --list \
    --zookeeper localhost:2181
my-topic
word-count-input
word-count-output

步骤 5 运行控制台生成器，失败并出现以下错误。

但是当我尝试使用下面的命令启动控制台生成器时，我遇到了异常。java.lang.IllegalStateException: There are no in-flight requests for node -1当我在 Windows 上运行时，同样的步骤工作正常，但我仅在 Mac 上遇到问题。

bin % kafka-console-producer.sh \
    --topic word-count-input \
    --bootstrap-server localhost:9092
>[2020-08-30 14:35:17,902] ERROR [Producer clientId=console-producer] Uncaught error in kafka producer I/O thread:  (org.apache.kafka.clients.producer.internals.Sender)
java.nio.BufferUnderflowException
    at java.nio.Buffer.nextGetIndex(Buffer.java:509)
    at java.nio.HeapByteBuffer.getInt(HeapByteBuffer.java:364)
    at org.apache.kafka.common.protocol.ByteBufferAccessor.readInt(ByteBufferAccessor.java:43)
    at org.apache.kafka.common.message.ResponseHeaderData.read(ResponseHeaderData.java:102)
    at org.apache.kafka.common.message.ResponseHeaderData.<init>(ResponseHeaderData.java:70)
    at org.apache.kafka.common.requests.ResponseHeader.parse(ResponseHeader.java:66)
    at org.apache.kafka.clients.NetworkClient.parseStructMaybeUpdateThrottleTimeMetrics(NetworkClient.java:717)
    at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:834)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:553)
    at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:325)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240)
    at java.lang.Thread.run(Thread.java:748)
[2020-08-30 14:35:17,904] ERROR [Producer clientId=console-producer] Uncaught error in kafka producer I/O thread:  (org.apache.kafka.clients.producer.internals.Sender)
java.lang.IllegalStateException: There are no in-flight requests for node -1
    at org.apache.kafka.clients.InFlightRequests.requestQueue(InFlightRequests.java:62)
    at org.apache.kafka.clients.InFlightRequests.completeNext(InFlightRequests.java:70)
    at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:833)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:553)
    at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:325)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240)
    at java.lang.Thread.run(Thread.java:748)
[2020-08-30 14:35:17,938] ERROR Uncaught exception in thread 'kafka-producer-network-thread | console-producer': (org.apache.kafka.common.utils.KafkaThread)
java.lang.OutOfMemoryError: Java heap space
    at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
    at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
    at org.apache.kafka.common.memory.MemoryPool$1.tryAllocate(MemoryPool.java:30)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:113)
    at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:447)
    at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:397)
    at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:678)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:580)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:485)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:544)
    at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:325)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240)
    at java.lang.Thread.run(Thread.java:748)

Answer 1

最后我发现了问题，该端口被本地运行的 sonarqube 使用，但令人惊讶的是，当我启动 kafka 时，没有报告错误。

Answer 2

老实说，我不知道防火墙/防病毒软件是如何发挥作用的，但不知何故它与这个问题有关。

就我而言，消费者只能消费一次消息，但当BufferUnderflowException我再次启动消费者时，它会抛出异常。

关闭防火墙/防病毒对我来说不是选项，因此我在 kafka-serveradvertised.listeners=PLAINTEXT://localhost:9092中添加/启用了配置server.properties，然后它按预期工作。

卡夫卡版本：2.6.0