KafkaSinkCluster - routing error handling

[rukai]

The kafka cluster has a lot of state that shotover must keep track of for routing purposes.
This includes things like which brokers are holding which partitions.
Currently shotover populates its internal records of this state but it has no way to invalidate these records.
This PR implements this missing invalidation.

To handle changes to the cluster we need to handle routing errors as these indicate that the cluster has changed.

This PR handles the 3 kinds of routing errors as follows:

• NOT_CONTROLLER set controller_broker to BrokerId(-1)
• NOT_COORDINATOR remove group from group_to_coordinator_broker
• NOT_LEADER_OR_FOLLOWER
  • remove topic from topic_by_name and topic_by_id
  • alternatively, if supported on this api version, immediately update topic_by_name/topic_by_id as per KIP-951
    • if the produce response is NOT_LEADER_OR_FOLLOWER and includes a newer leader epoch then we can update the topic entry with the provided broker ids

This PR ensures that every request type that we perform routing also has a response handler that invalidates the routing state when we get a routing error.

[codspeed-hq]

CodSpeed Performance Report

Merging #1659 will not alter performance

_{Comparing rukai:routing_error_handling (aa66e59) with main (caa3453)}

Summary

✅ 38 untouched benchmarks