Kafka Connect: Stop commits on terminated coordinator
This PR aims to improve the handling of coordinator shutdown during sink rebalances, to better ensure only one coordinator is active at a given time. The sink was designed with the assumption that only one coordinator and its tasks are active, to achieve exactly once semantics.
When a rebalance is triggered and close() is called on the sink task, here we immediately flag the coordinator as terminated, shut down the commit executor, and await termination from the main sink task thread (instead of from the coordinator thread). Any new table commits attempted by this coordinator will check the terminated flag and fail if set. This doesn't solve all edge cases but should be an improvement to fencing out zombie coordinators.
This PR also updates some exceptions thrown from RuntimeException to ConnectException to be a bit more specific.
@fqaiser94 @singhpk234 @ajantha-bhat , it would be great to get your feedback on this if you have a chance.
FYI, there is a discussion on how we can improve fencing further at https://github.com/apache/iceberg/issues/6514
Thanks for the reviews @danielcweeks , @singhpk234 , and @nk1506 !