I am running into an endless loop of the following 4 statements in kafka debug logs when I try to read information from a topic via kafka-console-consumer from one of the broker machines.
[2021-07-06 08:15:14,499] DEBUG [Consumer clientId=consumer-test-consumer-group-1-1, groupId=test-consumer-group-1] Sending FindCoordinator request to broker kafka01-broker:9094 (id: 5 rack: us-west-2b) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2021-07-06 08:15:14,504] DEBUG [Consumer clientId=consumer-test-consumer-group-1-1, groupId=test-consumer-group-1] Received FindCoordinator response ClientResponse(receivedTimeMs=1625559314504, latencyMs=5, disconnected=false, requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=3, clientId=consumer-test-consumer-group-1-1, correlationId=32), responseBody=FindCoordinatorResponseData(throttleTimeMs=0, errorCode=0, errorMessage='NONE', nodeId=3, host='kafka02-broker', port=9094)) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2021-07-06 08:15:14,504] INFO [Consumer clientId=consumer-test-consumer-group-1-1, groupId=test-consumer-group-1] Discovered group coordinator kafka02-broker:9094 (id: 2147483644 rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2021-07-06 08:15:14,504] INFO [Consumer clientId=consumer-test-consumer-group-1-1, groupId=test-consumer-group-1] Group coordinator kafka02-broker:9094 (id: 2147483644 rack: null) is unavailable or invalid, will attempt rediscovery (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2021-07-06 08:15:14,504] DEBUG [Consumer clientId=consumer-test-consumer-group-1-1, groupId=test-consumer-group-1] Sending FindCoordinator request to broker kafka01-broker:9094 (id: 5 rack: us-west-2b) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2021-07-06 08:15:14,507] DEBUG [Consumer clientId=consumer-test-consumer-group-1-1, groupId=test-consumer-group-1] Received FindCoordinator response ClientResponse(receivedTimeMs=1625559314504, latencyMs=5, disconnected=false, requestHeader=RequestHeader(apiKey=FIND_COORDINATOR, apiVersion=3, clientId=consumer-test-consumer-group-1-1, correlationId=32), responseBody=FindCoordinatorResponseData(throttleTimeMs=0, errorCode=0, errorMessage='NONE', nodeId=3, host='kafka02-broker', port=9094)) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2021-07-06 08:15:14,507] INFO [Consumer clientId=consumer-test-consumer-group-1-1, groupId=test-consumer-group-1] Discovered group coordinator kafka02-broker:9094 (id: 2147483644 rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2021-07-06 08:15:14,507] INFO [Consumer clientId=consumer-test-consumer-group-1-1, groupId=test-consumer-group-1] Group coordinator kafka02-broker:9094 (id: 2147483644 rack: null) is unavailable or invalid, will attempt rediscovery (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
and so on
Looking at kafka source code, I could see that the below code returns true and hence the endless loop
AbstractCoordinator.java
client.isUnavailable(coordinator) in
protected synchronized boolean ensureCoordinatorReady(final Timer timer) {
...
} else if (coordinator != null && client.isUnavailable(coordinator)) {
// we found the coordinator, but the connection has failed, so mark
// it dead and backoff before retrying discovery
markCoordinatorUnknown();
timer.sleep(rebalanceConfig.retryBackoffMs);
}
kafka version: 2.12-2.5.0
If I try connecting to the kafka02-broker:9094 from the kafka command utils, it seems to work fine, I am wondering why client.isUnavailable(coordinator) would return true.