我有一个 Kafka 集群,有 3 个经纪人和 3 个动物园管理员。根据Kafka server.log,由于某种未知原因,它间歇性地遇到损坏的管道错误。在损坏的管道错误消失后,Kafka 代理没有重新加入集群,而是决定离开集群并成为自己的领导者(因为它将 ISR 从 3 缩小到 1)。
到目前为止,唯一的解决方法是重新启动代理,它将作为跟随者正常重新加入集群。但我们不能每次出现类似问题时都手动重新启动。
[2019-05-10 10:32:48,344] WARN Failed to send SSL Close message (org.apache.kafka.common.network.SslTransportLayer)
java.io.IOException: Connection reset by peer
at sun.nio.ch.FileDispatcherImpl.write0(Native Method)
at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:47)
at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
at sun.nio.ch.IOUtil.write(IOUtil.java:65)
at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:471)
at org.apache.kafka.common.network.SslTransportLayer.flush(SslTransportLayer.java:209)
at org.apache.kafka.common.network.SslTransportLayer.close(SslTransportLayer.java:172)
at org.apache.kafka.common.utils.Utils.closeAll(Utils.java:718)
at org.apache.kafka.common.network.KafkaChannel.close(KafkaChannel.java:61)
at org.apache.kafka.common.network.Selector.doClose(Selector.java:746)
at org.apache.kafka.common.network.Selector.close(Selector.java:734)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:532)
at org.apache.kafka.common.network.Selector.poll(Selector.java:424)
at kafka.network.Processor.poll(SocketServer.scala:628)
at kafka.network.Processor.run(SocketServer.scala:545)
at java.lang.Thread.run(Thread.java:745)
[2019-05-10 10:32:48,368] WARN Failed to send SSL Close message (org.apache.kafka.common.network.SslTransportLayer)
java.io.IOException: Broken pipe
at sun.nio.ch.FileDispatcherImpl.write0(Native Method)
at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:47)
at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
at sun.nio.ch.IOUtil.write(IOUtil.java:65)
at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:471)
at org.apache.kafka.common.network.SslTransportLayer.flush(SslTransportLayer.java:209)
at org.apache.kafka.common.network.SslTransportLayer.close(SslTransportLayer.java:159)
at org.apache.kafka.common.utils.Utils.closeAll(Utils.java:718)
at org.apache.kafka.common.network.KafkaChannel.close(KafkaChannel.java:61)
at org.apache.kafka.common.network.Selector.doClose(Selector.java:746)
at org.apache.kafka.common.network.Selector.close(Selector.java:734)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:532)
at org.apache.kafka.common.network.Selector.poll(Selector.java:424)
at kafka.network.Processor.poll(SocketServer.scala:628)
at kafka.network.Processor.run(SocketServer.scala:545)
at java.lang.Thread.run(Thread.java:745)
[2019-05-10 10:32:53,422] WARN Failed to send SSL Close message (org.apache.kafka.common.network.SslTransportLayer)
java.io.IOException: Broken pipe
at sun.nio.ch.FileDispatcherImpl.write0(Native Method)
at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:47)
at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93)
at sun.nio.ch.IOUtil.write(IOUtil.java:65)
at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:471)
at org.apache.kafka.common.network.SslTransportLayer.flush(SslTransportLayer.java:209)
at org.apache.kafka.common.network.SslTransportLayer.close(SslTransportLayer.java:159)
at org.apache.kafka.common.utils.Utils.closeAll(Utils.java:718)
at org.apache.kafka.common.network.KafkaChannel.close(KafkaChannel.java:61)
at org.apache.kafka.common.network.Selector.doClose(Selector.java:746)
at org.apache.kafka.common.network.Selector.close(Selector.java:734)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:532)
at org.apache.kafka.common.network.Selector.poll(Selector.java:424)
at kafka.network.Processor.poll(SocketServer.scala:628)
at kafka.network.Processor.run(SocketServer.scala:545)
at java.lang.Thread.run(Thread.java:745)
[2019-05-10 10:32:56,976] INFO [Partition CS_NL_CUSTOMER_ADD-1 broker=4] Shrinking ISR from 4,6 to 4 (kafka.cluster.Partition)
[2019-05-10 10:32:56,994] INFO [Partition CS_NL_CUSTOMER_ADD-1 broker=4] Cached zkVersion [24394] not equal to that in zookeeper, skip updating ISR (kafka.cluster.Partition)
[2019-05-10 10:32:56,994] INFO [Partition _confluent-controlcenter-5-1-0-1-MetricsAggregateStore-changelog-3 broker=4] Shrinking ISR from 4,6 to 4 (kafka.cluster.Partition)
[2019-05-10 10:32:57,023] INFO [Partition _confluent-controlcenter-5-1-0-1-MetricsAggregateStore-changelog-3 broker=4] Cached zkVersion [3724] not equal to that in zookeeper, skip updating ISR (kafka.cluster.Partition)
[2019-05-10 10:32:57,023] INFO [Partition TEST_3_PART-2 broker=4] Shrinking ISR from 4,6 to 4 (kafka.cluster.Partition)
[2019-05-10 10:32:57,033] INFO [Partition TEST_3_PART-2 broker=4] Cached zkVersion [3300] not equal to that in zookeeper, skip updating ISR (kafka.cluster.Partition)
据我对 Kafka 的了解,Kafka 代理不是应该尽可能重新加入集群吗?为什么在出现管道损坏错误后没有发生? 还有一件事是,知道是什么导致了“Broken Pipe”错误吗?是不是网络问题?
证书好像有问题。您需要为代理颁发签名证书并将其添加到客户端的密钥库中。
Confluence 关于如何配置使用 SSL 进行加密和身份验证 的指南应该会让事情变得更加清晰。
更新:解决方案是在部署Kafka的虚拟机上禁用vMotion。
看来每当vMotion执行实时迁移时,Kafka都会受到影响并导致管道损坏错误。根据此链接,Confluence 还建议禁用 vMotion,因为它可能会导致集群中断。
禁用 vMotion 后,我的 Kafka 集群再也不会遇到任何损坏的管道问题。这仍然没有回答为什么遇到断管错误后不会自动重新加入集群,但至少解决了断管错误。