KafkaSinkCluster: handle broker closing connection due to idle timeout

[rukai]

Kafka brokers have a config field connections.max.idle.ms, when a connection has been idle for longer than this timeout.
Normally a client can expect idle connections to timeout after some time of being unused.
However when KafkaSinkCluster is sitting in between the client and the broker there can be some unexpected behaviour.

For example:

1. We have a kafka cluster configured with connections.max.idle.ms=5s.
2. A client sends a request every 1s.
3. Shotover routes the first request to node A
4. Shotover routes the next 10 requests to node B
5. Shotover routes another request to node A
   • this request will fail due to the connection hitting idle time out even though the client was sending one request per second, well within the timeout!

With more reasonable timeout values this issue is harder to hit, but we are still still hitting this in our preprod environment with the default idle timeout of 10 minutes.

Integration test

This PR introduces an integration test that runs kafka with a very low connections.max.idle.ms (30s), this reproduces the issue described above.
The new integration test fails when running the shotover main branch.

Fix implementation

This PR then fixes the failing test by adding a check into our connection getting logic to recreate the connection if it has hit an error.
This does not attempt to fix all possible failures due to closed connection, as that would be impossible due to race conditions.
Instead we make a best effort handling which will catch the regular failures we are seeing.

[codspeed-hq]

CodSpeed Performance Report

Merging #1752 will improve performances by 10.65%

_{Comparing rukai:catch_closed_connections_during_send (5722c26) with main (12ff171)}

Summary

⚡ 1 improvements
✅ 38 untouched benchmarks

Benchmarks breakdown

	Benchmark	main	rukai:catch_closed_connections_during_send	Change
⚡	encode_request_metadata	11.6 µs	10.5 µs	+10.65%

[justinweng-instaclustr]

Hey Lucas, the connections.max.idle.ms in description should be milliseconds, right? But it's referred to as seconds
EDIT: Seconds are correctly converted to milliseconds in code. All good!