KafkaSinkCluster - set KafkaNode::state

[rukai]

This PR:

• sets the KafkaNode::state field to NodeState::Down whenever a connection error is encountered.
  • This sets us up to make use of that fields value as part of our routing logic in a follow up PR.
• recreates lost connections where it is safe to do so.
  • send - never recreate lost connections on send, since we cant tell if the sent message was received or not. Possibly in the future we could recreate lost connections in certain conditions. But its not really important.
  • receive - never recreate lost connections on receive, we only check for received messages if they are pending, and if there were pending responses we would not be able to recreate the connection.
  • control_send_receive - we can recreate connections as needed. Since control connections are used for metadata requests generated by shotover instead of arbitrary requests coming from the client, we can just recreate connections without fear of idempotency issues.

To enable this a Connections::handle_connection_error method is added. This method is called anytime a connection error occurs.

It creates a new connection to check if the node is broken. If the connection fails the node is marked as down and Err(..) is returned, otherwise the old connection is replaced with the new connection and Ok(()) is returned.

It is then up to the caller (send/receive/control_send_receive) to decide how to handle this success or failure.

• send ignores the success and fails anyway.
• receive will hit the Succesfully reopened outgoing connection but previous outgoing connection had pending requests. error since it always has a pending request (thats why its receiving!)
• control_send_receive has its own extensive logic that calls handle_connection_error first to check if the original node is bad or just the connection, then it proceeds to try new nodes without going through handle_connection_error. It can recover from issues at any point since control_send_receive is used for internal shotover requests.

[codspeed-hq[bot]]

CodSpeed Performance Report

Merging #1715 will degrade performances by 11.47%

_{Comparing rukai:kafka_sink_cluster_set_state (3ee31a0) with main (307bb10)}

Summary

❌ 1 regressions ✅ 38 untouched benchmarks

:warning: _Please fix the performance issues or acknowledge them on CodSpeed._

Benchmarks breakdown

	Benchmark	main	rukai:kafka_sink_cluster_set_state	Change
❌	encode_system.local_result_v5_no_compression	92.9 µs	104.9 µs	-11.47%

[justinweng-instaclustr]

What's the performance regression about?

[rukai]

its for a cassandra benchmark, definitely just noise