Fix NOT_COORDINATOR warnings for Kafka consumer offset commit requests

[justinweng-instaclustr]

This PR fixes the NOT_COORDINATOR warning when Kafka consumers try to commit offsets. It's fixed by routing the OffsetCommit requests to the group coordinator in the KafkaSinkCluster method route_requests().

It also adds a new Kafka integration test case produce_consume_commit_offsets_partitions1 which calls commit_sync(&offsets) to commit the offset after a valid consume, and calls committed to verify the offset has been successfully committed (refs: Kafka Java driver commitSync(), Kafka Java driver committed(), Kafka C++ driver commit(), Kafka C++ driver committed_offsets()).

commit_sync was chosen over commit_async due to the following reasons:

• We need to check if the offset has been committed after sending the offset commit request, and hence need the commit_sync to block until the commit succeeds.

• It may be possible to wrap committed() in the OffsetCommitCallback for the commitAsync() method so that the check on committed offsets can be triggered after commitAsync() completes. But it seems complicated to implement and probably not worth the effort.

Closes #1687

[codspeed-hq[bot]]

CodSpeed Performance Report

Merging #1693 will not alter performance

_{Comparing justinweng-instaclustr:1687-kafka-consumers-seeing-not_coordinator-warnings (9fef022) with main (6ec2022)}

Summary

✅ 39 untouched benchmarks