У меня есть кластер 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)
Насколько я понимаю, о Кафке, разве не предполагается, что брокер Кафки должен присоединиться к кластеру, когда это возможно? Почему это не происходит после ошибки сломанной трубы?
Еще одна вещь, есть идеи, что вызвало ошибку Broken Pipe? Это проблема сети?