Ensure exactly-once on connector task(w/ coordinator) rebalancing, as like Apache version

[okayhooni]

re-opening PR #279 with additional defense logic authored by @bryanck

Context

I found that duplicated records occurred on the CDC sink with this Iceberg sink connector after using spot nodes and activating the node consolidation feature of Karpenter. Although it happens very rarely, when it does occur, it tends to happen consecutively. In a related issue inquiry, @bryanck informed me that in the Iceberg version of the connector, safeguard logic has been added to ensure that no more than one coordinator task is running simultaneously during the connector rebalancing process.

Commit Contents

• cherry-pick this safeguard logic from Apache version to the Tabular version as well.
• the safeguard logic is designed to prevent more than one coordinator task from running simultaneously during connector task rebalancing, such as in cases where spot instances are terminated.

Related Links

• Defense logic on Iceberg version: https://github.com/apache/iceberg/blob/4dbc7f578eee7ceb9def35ebfa1a4cc236fb598f/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java#L303-L310
• Related Slack thread: https://apache-iceberg.slack.com/archives/C03SEK9PW0Z/p1721977110612699
• Reference commit: bryanck@e2d4c51
  (The reference commit was not based on the latest branch of Tabular version, so the method names of interface are slightly different like stop() versus close().)

cc/ @fqtab

[okayhooni]

@fqaiser94 ( @fqtab )

Could you review this commit and related discussions on the #kafka-connect channel of apache-iceberg Slack community..?

• https://apache-iceberg.slack.com/archives/C03SEK9PW0Z/p1721977110612699