Major Description
We are running in an issue really similar to KAFKA-4950.
We have a producer running and a MetricsReporter with a background thread which publishes these metrics.
The concurrent exception happens when calling `InFlightRequests.count()` in one thread when a connection or disconnection is happening.
In this case one thread is iterating over the map while another is adding/removing from it thus causing the exception.
We could potentially fix this with a volatile like in KAFKA-4950.