kumarpritam863 opened a new pull request, #17376:
URL: https://github.com/apache/iceberg/pull/17376
## What
Reworks how the Kafka Connect sink elects its single commit `Coordinator`
and adds a per-connector JMX metric for the number of active coordinators.
### 1. Decouple coordinator election from source-partition assignment
Previously the coordinator was elected inside the source-topic rebalance
callbacks (`open`/`close`) via `Admin.describeConsumerGroups(connectGroupId)`,
picking the task that owned the
globally-lowest `(topic, partition)` from a point-in-time group snapshot.
This coupling caused several issues:
- **Split-brain / orphaned coordinator** — under incremental cooperative
rebalancing the lowest partition could be transiently seen by two members, so
two tasks each elected themselves
(or a losing task failed to de-elect).
- **Zero-coordinator windows** and **hot-path Admin dependency**
(`describeConsumerGroups` on every rebalance, needing `DESCRIBE` ACL and able
to throw on the rebalance path).
This PR replaces that with an election that needs **no Admin call**:
- The election key is partition 0 of the lexicographically-smallest topic
in the task's own consumer `subscription()` (connector-wide, so every task
converges on the same key; works for
both a literal `topics` list and `topics.regex`).
- Leadership is reconciled **level-triggered on the Connect task thread in
`save()`** — the task that owns the election key runs the coordinator. Keeping
start/stop off the rebalance
callbacks avoids stalling the group rebalance (coordinator teardown can
block up to the termination timeout) and is self-correcting across
missed/reordered cooperative-rebalance
callbacks.
assignments.
- Adds a `Committer.configure(...)` lifecycle hook (default no-op on the
interface) for one-time init; the source consumer is resolved once via
reflection and cached off the per-`save()`
hot path, resolving lazily so a resolution failure degrades gracefully
instead of failing task start.
Exactly-once safety is unchanged — it continues to rest on the Iceberg
offsets compare-and-swap, not on single-leadership.
### 2. Per-connector active-coordinator metric
Adds `CoordinatorMetrics`, a JMX gauge exposing `ActiveCoordinatorCount`
on `org.apache.iceberg.connect:type=CoordinatorMetrics,connector="<name>"`:
- One MBean per connector name, shared across that connector's tasks in
the worker JVM; the cluster-wide total is the sum across workers.
- **Ref-counted lifecycle**: registered when a connector's first task
attaches, unregistered when its last task detaches, so a deleted connector
leaves no lingering MBean.
Register/unregister transitions are serialized to avoid the
attach-during-teardown race; counter reads/updates are lock-free.
- Metric failures never affect the connector (registration errors are
logged and swallowed).
Operationally: expected steady state is `1` per connector; a sustained
value `> 1` flags a split-brain (safe under OCC, but wasteful) and `0` past a
commit interval flags a
coordinator-absent gap.
## Compatibility
- No config or wire-format changes.
- No change to exactly-once guarantees.
- Deprecated `start()`/`stop()` behavior is unchanged.
## Testing
- `TestCommitterImpl` — election key resolution and level-triggered
start/stop of the coordinator.
- `TestCoordinatorMetrics` — inc/dec symmetry, shared instance per
connector, per-connector MBean registration/readback, ref-counted teardown,
re-attach after full detach, and
unbalanced-detach safety.
- `TestCoordinator` updated for the explicit partition-count constructor.
Address the following:
1. https://github.com/apache/iceberg/issues/16282#issuecomment-5082228154
2. https://github.com/apache/iceberg/issues/13593#issuecomment-4796667015
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
