如果经纪人死亡，Kafka 将无法恢复

Question

我们有一个带有三个代理（节点 ID 0、1、2）的 kafka 集群和一个带有三个节点的 zookeeper 设置。

我们在这个有 20 个分区和复制因子 2 的集群上创建了一个主题“test”。我们使用 Java 生产者 API 向这个主题发送消息。其中一位 kafka 代理间歇性地宕机，之后无法恢复。为了模拟这种情况，我们手动杀死了其中一个经纪人。根据 kafka arch，它应该会自我恢复，但这并没有发生。当我在控制台上描述该主题时，我看到 ISR 的数量减少到少数分区的一个，因为其中一个代理被杀死。现在，每当我们尝试通过生产者 API（Java 客户端或控制台生产者）推送消息时，我们都会遇到 SocketTimeoutException.. 快速查看日志说，“无法获取元数据”

 WARN [2015-07-01 22:55:07,590] [ReplicaFetcherThread-0-3][] kafka.server.ReplicaFetcherThread - [ReplicaFetcherThread-0-3],
 Error in fetch Name: FetchRequest; Version: 0; CorrelationId: 23711; ClientId: ReplicaFetcherThread-0-3; 
 ReplicaId: 0; MaxWait: 500 ms; MinBytes: 1 bytes; RequestInfo: [zuluDelta,2] -> PartitionFetchInfo(11409,1048576),[zuluDelta,14] -> PartitionFetchInfo(11483,1048576). 
 Possible cause: java.nio.channels.ClosedChannelException


[2015-07-01 23:37:40,426] WARN Fetching topic metadata with correlation id 0 for topics [Set(test)] from broker [id:1,host:abc-0042.yy.xxx.com,port:9092] failed (kafka.client.ClientUtils$)
java.net.SocketTimeoutException
at sun.nio.ch.SocketAdaptor$SocketInputStream.read(SocketAdaptor.java:201)
at sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:86)
at java.nio.channels.Channels$ReadableByteChannelImpl.read(Channels.java:221)
at kafka.utils.Utils$.read(Utils.scala:380)
at kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54)
at kafka.network.Receive$class.readCompletely(Transmission.scala:56)
at kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29)
at kafka.network.BlockingChannel.receive(BlockingChannel.scala:111)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
at kafka.producer.SyncProducer.send(SyncProducer.scala:113)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:93)
at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60)

任何线索将不胜感激...

Answer 1

从您的错误无法获取元数据来看，这主要可能是因为您可能已将bootstrap.servers生产者中的设置为已死亡的代理。

理想情况下，bootstrap.servers 列表中必须有多个代理，因为如果其中一个代理失败（或无法访问），则另一个代理可以为您提供元数据。

仅供参考：元数据是有关特定主题的信息，它告诉您该主题有多少个分区、它们的领导者经纪人、追随者经纪人等。

因此，当为分区生成密钥时，其相应的领导者代理将成为消息将发送到的对象。

从你的问题来看，你的 ISR 集只有一个经纪人。您可以尝试bootstrap.server将此经纪人设置为。