We've ran into a similar issue as this ticket: https://issues.apache.org/jira/projects/KAFKA/issues/KAFKA-7755
The fix for
KAFKA-7755 doesn't work for this case as the hostname is not updated when resolving the addresses.
`ClusterConnectionStates::connecting` method makes an assumption that broker ID will always map to same hostname. In our case, when a broker is terminated in AWS, it is replaced by a different instance under the same broker ID.
In this case, the consumer fails to connect to the right host when the broker ID returns to the cluster. For example, we see the following line in DEBUG logs:
It tries to connect to the new broker instance using the wrong (old) IP address.