KafkaSinkCluster - handle broker configured connection timeouts

[rukai]

pulled out of #1728

• sending requests will reset the timeout counter, in the case where the connection has timed out when we send the request there should be no pending requests so we can just open it (with a sanity check on number of pending requests)
  • oh wait we cant do this because a message might still pass through on connection closed... Maybe we need a ConnectionError::OtherSideClosedCleanly? Maybe we can use the read disconnect? Need to do testing on kafka itself to determine whether it sends TCP FIN or TCP RST when a connection times out.
• receiving requests will not reset the timeout counter, but we can just avoid checking for requests when a connection has no pending requests.