Consumer not connecting to available broker after broker restart

[parthipanj]

B1, B2, B3 - Brokers

Scale down B2, B3 - Now B1 becomes the leader for topics
Scale down B1
Scale up B2, B3
Only after scaling up B1 We can receive the message in that topic
The issue is, without restarting the Kafka client consumer not trying to connect to the next broker if the current broker is crashed.

consumer = KafkaConsumer(
    <<topic>>,
    group_id=<<group-id>>,
    bootstrap_servers=[B1,B2,B3],
    auto_offset_reset='latest'
)

for message in consumer:
# message value and key are raw bytes -- decode if necessary!
# e.g., for unicode: `message.value.decode('utf-8')`
print("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                                      message.offset, message.key,
                                      message.value))

Error:

DEBUG:kafka.client:Initializing connection to node 2 for metadata request
DEBUG:kafka.conn:<BrokerConnection node_id=2 host=<<host>> <disconnected> [IPv4 ('<<host>>', 9094)]>: creating new socket
DEBUG:kafka.conn:<BrokerConnection node_id=2 host=<<host>> <disconnected> [IPv4 ('<<host>>', 9094)]>: setting socket option (6, 1, 1)
INFO:kafka.conn:<BrokerConnection node_id=2 host=<<host>> <connecting> [IPv4 ('<<host>>', 9094)]>: connecting to <<host>> [('<<host>>', 9094) IPv4]
ERROR:kafka.conn:Connect attempt to <BrokerConnection node_id=2 host=<<host>> <connecting> [IPv4 ('<<host>>', 9094)]> returned error 111. Disconnecting.
INFO:kafka.conn:<BrokerConnection node_id=2 host=<<host>> <connecting> [IPv4 ('<<host>>)]>: Closing connection. KafkaConnectionError: 111 ECONNREFUSED
DEBUG:kafka.conn:<BrokerConnection node_id=2 host=<<host>> <connecting> [IPv4 ('<<host>>)]>: reconnect backoff 1.0150467424400873 after 121 failures
WARNING:kafka.client:Node 2 connection failed -- refreshing metadata
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available
DEBUG:kafka.client:Give up sending metadata request since no node is available

[jeffwidman]

As seen in your logs, the client has no known active broker to contact... when you took down B1, even if you immediately turn on B2/B3, the client may not have yet received updated metadata from the cluster, so it hasn't been told about B2/B3... the client has no idea who to contact to find updated info.

I do not remember whether we throw an exception or fallback to bootstrap servers list in this situation.

What were you expecting to happen? Why would this be a bug rather than expected behavior?

[MdStart]

Thanks for the insight @jeffwidman .
I am also facing similar problem- consumer not getting aware if broker gets restarted.
Can you help me to clarify this--
If that is the expected behavior then is there any callback or exception in kafka-python which can be used to check the kafka server health and restart or re-instantiate the consumer instance??
Or any other health-check mechanism in the library available.

[dpkp]

I believe the behavior changed in #1753 -- bootstrap server list is now only used for initial connections. Once we get a single MetadataResponse w/ brokers, we use that broker list and all future reconnects would come from that list. So if a cluster gets into a really bad state and drops to a single broker, and the client picks up the metadata showing only that single broker, and then that broker goes offline, then the client would retry only that single broker and not recover even if other brokers come back first.

But in the general case, if a broker goes down and comes back -- that will get handled in the normal course of metadata updates.

[MdStart]

@dpkp Thanks but metadata failing to update if any broker goes down causing no messages in consumer even though other brokers are available to serve the consumer. This is happening only in multi-broker environment for a consumer however single broker goes down, comes back consumer works as expected.

Any insight or solution for the multi-broker failure handling in consumer is appreciated.

[mariomtzjr]

@Parthi-J I have the same problem, Did you resolve that problem with you consumer? I get this output in my console:

2019-07-08 11:35:57,647: kafka.conn - ERROR: Connect attempt to <BrokerConnection node_id=1 host=172.28.0.3:9092 <connecting> [IPv4 ('172.28.0.3', 9092)]> returned error 60. Disconnecting.
2019-07-08 11:35:57,652: kafka.conn - INFO: <BrokerConnection node_id=1 host=172.28.0.3:9092 <connecting> [IPv4 ('172.28.0.3', 9092)]>: Closing connection. KafkaConnectionError: 60 ETIMEDOUT
2019-07-08 11:35:57,654: kafka.conn - DEBUG: <BrokerConnection node_id=1 host=172.28.0.3:9092 <connecting> [IPv4 ('172.28.0.3', 9092)]>: reconnect backoff 1.148591061811787 after 13 failures
2019-07-08 11:35:57,655: kafka.client - WARNING: Node 1 connection failed -- refreshing metadata
2019-07-08 11:35:57,656: kafka.client - DEBUG: Initializing connection to node 1 for metadata request
2019-07-08 11:35:57,760: kafka.conn - DEBUG: <BrokerConnection node_id=1 host=172.28.0.3:9092 <disconnected> [IPv4 ('172.28.0.3', 9092)]>: creating new socket
2019-07-08 11:35:57,761: kafka.conn - DEBUG: <BrokerConnection node_id=1 host=172.28.0.3:9092 <disconnected> [IPv4 ('172.28.0.3', 9092)]>: setting socket option (6, 1, 1)
2019-07-08 11:35:57,762: kafka.conn - INFO: <BrokerConnection node_id=1 host=172.28.0.3:9092 <connecting> [IPv4 ('172.28.0.3', 9092)]>: connecting to 172.28.0.3:9092 [('172.28.0.3', 9092) IPv4]

[jeffwidman]

@MdStart when you say:

causing no messages in consumer even though other brokers are available to serve the consumer.

This sounds alarming.

Has the leadership of the partitions on the failed broker finished moving to a new broker? You can check the Kafka broker logs to verify.

Can you paste DEBUG logs from the consumer?

Theoretically it should realize there's a connection difficulty, mark that broker as failed, then request who is the new leader of the partition... if there is not yet a new leader, it will cycle for a little bit until the cluster stabilizes then it should start fetching from the new leader. It can take a little bit of time depending on how fast your cluster stabilizes and how fast the client realizes the broker is down.

There are a lot of steps in there, so if this is truly a bug, we really need to see the logs to see at what point it's breaking...

[dpkp]

I'm confused by what is expected here so I'm going to close this issue.