Kafka transport check not working as expected

[visionding]

如题，Kafka重启之后，Consumer接收不到消息，但是Producer发送正常，重启之后可正常接收消息。

运行环境：
Cent OS 7.5
Kafka 2.1
CAP 2.6
ASP.NET Core 2.2
Docker 18.06

[visionding]

这是Kafka重启时生产者和消费者的日志
Producer:

{"log":"%3|1574300663.612|ERROR|rdkafka#producer-3| [thrd:192.168.1.222:9092/bootstrap]: 1/1 brokers are down\r\n","stream":"stdout","time":"2019-11-21T01:44:23.613186283Z"}
{"log":"%3|1574300663.612|ERROR|rdkafka#producer-6| [thrd:192.168.1.222:9092/bootstrap]: 1/1 brokers are down\r\n","stream":"stdout","time":"2019-11-21T01:44:23.613211999Z"}
{"log":"%3|1574300663.681|ERROR|rdkafka#producer-2| [thrd:192.168.1.222:9092/bootstrap]: 1/1 brokers are down\r\n","stream":"stdout","time":"2019-11-21T01:44:23.681647714Z"}
{"log":"%3|1574300663.681|ERROR|rdkafka#producer-7| [thrd:192.168.1.222:9092/bootstrap]: 1/1 brokers are down\r\n","stream":"stdout","time":"2019-11-21T01:44:23.681684169Z"}
{"log":"%3|1574300663.683|ERROR|rdkafka#producer-5| [thrd:192.168.1.222:9092/bootstrap]: 1/1 brokers are down\r\n","stream":"stdout","time":"2019-11-21T01:44:23.684381545Z"}
{"log":"%3|1574300663.683|ERROR|rdkafka#producer-4| [thrd:192.168.1.222:9092/bootstrap]: 1/1 brokers are down\r\n","stream":"stdout","time":"2019-11-21T01:44:23.684420976Z"}
{"log":"%3|1574300663.833|FAIL|rdkafka#producer-2| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.833364026Z"}
{"log":"%3|1574300663.833|ERROR|rdkafka#producer-2| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.833388212Z"}
{"log":"%3|1574300663.904|FAIL|rdkafka#producer-6| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 9ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.904577402Z"}
{"log":"%3|1574300663.904|ERROR|rdkafka#producer-6| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 9ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.904606673Z"}
{"log":"%3|1574300663.985|FAIL|rdkafka#producer-5| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.98599147Z"}
{"log":"%3|1574300663.985|ERROR|rdkafka#producer-5| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.986034613Z"}
{"log":"\u001b[41m\u001b[1m\u001b[37mcrit\u001b[39m\u001b[22m\u001b[49m: DotNetCore.CAP.ConsumerRegister[0]\r\n","stream":"stdout","time":"2019-11-21T01:44:23.99579267Z"}
{"log":"      Kafka server connection error. --\u003e An error occurred during connect kafka --\u003e 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.99580923Z"}
{"log":"\u001b[41m\u001b[1m\u001b[37mcrit\u001b[39m\u001b[22m\u001b[49m: DotNetCore.CAP.ConsumerRegister[0]\r\n","stream":"stdout","time":"2019-11-21T01:44:23.995815902Z"}
{"log":"      Kafka server connection error. --\u003e An error occurred during connect kafka --\u003e 2/2 brokers are down\r\n","stream":"stdout","time":"2019-11-21T01:44:23.995819441Z"}
{"log":"\u001b[41m\u001b[1m\u001b[37mcrit\u001b[39m\u001b[22m\u001b[49m: DotNetCore.CAP.ConsumerRegister[0]\r\n","stream":"stdout","time":"2019-11-21T01:44:23.995824214Z"}
{"log":"      Kafka server connection error. --\u003e An error occurred during connect kafka --\u003e GroupCoordinator: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.995830141Z"}
{"log":"%3|1574300664.445|FAIL|rdkafka#producer-4| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:24.445851083Z"}
{"log":"%3|1574300664.445|ERROR|rdkafka#producer-4| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:24.445886161Z"}
{"log":"%3|1574300664.607|FAIL|rdkafka#producer-3| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:24.607189617Z"}
{"log":"%3|1574300664.607|ERROR|rdkafka#producer-3| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:24.607224755Z"}
{"log":"%3|1574300664.640|FAIL|rdkafka#producer-7| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:24.641077676Z"}
{"log":"%3|1574300664.640|ERROR|rdkafka#producer-7| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:24.641099579Z"}

生产者在Kafka重启完成之后还能成功发出消息

Consumer:

{"log":"\u001b[41m\u001b[1m\u001b[37mcrit\u001b[39m\u001b[22m\u001b[49m: DotNetCore.CAP.ConsumerRegister[0]\r\n","stream":"stdout","time":"2019-11-21T01:44:23.613128611Z"}
{"log":"      Kafka server connection error. --\u003e An error occurred during connect kafka --\u003e GroupCoordinator: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.613159406Z"}
{"log":"\u001b[41m\u001b[1m\u001b[37mcrit\u001b[39m\u001b[22m\u001b[49m: DotNetCore.CAP.ConsumerRegister[0]\r\n","stream":"stdout","time":"2019-11-21T01:44:23.68389106Z"}
{"log":"      Kafka server connection error. --\u003e An error occurred during connect kafka --\u003e 2/2 brokers are down\r\n","stream":"stdout","time":"2019-11-21T01:44:23.683919334Z"}
{"log":"\u001b[41m\u001b[1m\u001b[37mcrit\u001b[39m\u001b[22m\u001b[49m: DotNetCore.CAP.ConsumerRegister[0]\r\n","stream":"stdout","time":"2019-11-21T01:44:23.684367093Z"}
{"log":"      Kafka server connection error. --\u003e An error occurred during connect kafka --\u003e 192.168.1.222:9092/0: Connect to ipv4#192.168.1.222:9092 failed: Connection refused (after 0ms in state CONNECT)\r\n","stream":"stdout","time":"2019-11-21T01:44:23.684380321Z"}

{"log":"%4|1574300678.287|REQTMOUT|rdkafka#consumer-1| [thrd:192.168.1.222:9092/bootstrap]: 192.168.1.222:9092/0: Timed out 0 in-flight, 1 retry-queued, 0 out-queue, 0 partially-sent requests\r\n","stream":"stdout","time":"2019-11-21T01:44:38.288574501Z"}

消费者在Kafka重启完成之后就再也接收不到消息，重启消费者之后才恢复正常

[visionding]

经测试，Kafka重启之后日志中出现

Nov 21 14:27:49 kafka-server-start.sh: [2019-11-21 14:27:49,676] INFO [GroupCoordinator 0]: Member rdkafka-1ea4a1d7-b4c0-48e9-be8d-b65bfe1007b6 in group cap.queue.notification.test.v1 has failed, removing it from the group (kafka.coordinator.group.GroupCoordinator)
Nov 21 14:27:49 kafka-server-start.sh: [2019-11-21 14:27:49,678] INFO [GroupCoordinator 0]: Preparing to rebalance group cap.queue.notification.test.v1 in state PreparingRebalance with old generation 8 (__consumer_offsets-45) (reason: removing member rdkafka-1ea4a1d7-b4c0-48e9-be8d-b65bfe1007b6 on heartbeat expiration) (kafka.coordinator.group.GroupCoordinator)
Nov 21 14:27:49 kafka-server-start.sh: [2019-11-21 14:27:49,682] INFO [GroupCoordinator 0]: Group cap.queue.notification.test.v1 with generation 9 is now empty (__consumer_offsets-45) (kafka.coordinator.group.GroupCoordinator)

类似这种日志，则消费者无法接收到消息

[yang-xiaodong]

Hello, thank you for your feedback.

After checking, we found that the transport check processor in Kafka did not works as expected, it is a bug, we will fix it in v3.0

[yang-xiaodong]

v3.0 nuget package is released!