laskoviymishka commented on code in PR #17450:
URL: https://github.com/apache/iceberg/pull/17450#discussion_r3711442566
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -192,13 +130,32 @@ public void save(Collection<SinkRecord> sinkRecords) {
startWorker();
worker.save(sinkRecords);
}
+ if (reconcileNeeded) {
+ reconcileLeadership();
+ reconcileNeeded = false;
+ }
processControlEvents();
}
+ private void reconcileLeadership() {
+ Set<String> subscribedTopics =
Sets.newTreeSet(sourceConsumer().subscription());
+ TopicPartition leader = leaderPartition(subscribedTopics);
+ if (leader != null && context.assignment().contains(leader)) {
+ startCoordinator(subscribedTopics);
+ } else {
+ stopCoordinator();
+ }
+ }
+
private void processControlEvents() {
if (coordinatorThread != null && coordinatorThread.isTerminated()) {
- throw new NotRunningException(
- String.format("Coordinator unexpectedly terminated on committer %s",
taskId));
+ if (isProducerFenced(coordinatorThread.exception())) {
+ LOG.warn("Committer {} coordinator was fenced by a newer coordinator;
clearing it", taskId);
+ stopCoordinator();
Review Comment:
When we clear a fenced coordinator here, we don't re-arm `reconcileNeeded`,
so I think we can end up with no coordinator at all.
If the fence happens after the last `reconcileLeadership()` (so
`reconcileNeeded` is already false), we null out `coordinatorThread` and
nothing schedules another reconcile — no future `save()` re-evaluates
leadership until the next rebalance. This task is still the leader by
assignment, but it silently stops producing commits.
Setting `reconcileNeeded = true` right after `stopCoordinator()` in this
branch would let the next `save()` restart it. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -53,20 +54,29 @@ abstract class Channel {
private final Admin admin;
private final Map<Integer, Long> controlTopicOffsets = Maps.newHashMap();
private final String producerId;
+ private final String channelId;
Channel(
String name,
String consumerGroupId,
IcebergSinkConfig config,
KafkaClientFactory clientFactory,
SinkTaskContext context) {
+ this.channelId = config.connectorName() + "-" + config.taskId() + "-" +
name;
this.controlTopic = config.controlTopic();
this.connectGroupId = config.connectGroupId();
this.context = context;
- String transactionalId = config.transactionalPrefix() + name +
config.transactionalSuffix();
+ String transactionalId =
+ "worker".equalsIgnoreCase(name)
+ ? config.transactionalPrefix() + name +
config.transactionalSuffix()
+ : connectGroupId + "-" + config.connectorName() + "-coord";
Review Comment:
I'd hold on this txn-id change — it's the biggest risk in the PR for me.
The format flips from `transactionalPrefix + name + transactionalSuffix` to
`connectGroupId + "-" + connectorName + "-coord"`, so an old coordinator and a
new one carry different transactional ids across a rolling upgrade.
`initTransactions()` only fences the same id, so for the overlap window both
coordinators are live — Iceberg CAS lets one win per table, but the loser can
still advance the control-topic consumer-group offsets, and a later coordinator
replaying from those offsets skips events.
It also drops `transactionalSuffix()` (the worker path keeps it). Operators
set that suffix to isolate producers across multiple Connect clusters, so every
cluster's coordinator now collapses onto the same id and they fence each other.
I'd keep the suffix on both paths and gate the new format behind config so
upgrades have a migration path. (Minor while we're here: `connectGroupId`
already defaults to `connect-<connectorName>`, so this renders as
`connect-myconn-myconn-coord` — the name lands twice.) wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -214,11 +171,21 @@ private void startWorker() {
}
}
- private void startCoordinator() {
+ private void startCoordinator(Set<String> subscribedTopics) {
if (null == this.coordinatorThread) {
- LOG.info("Task {} elected leader, starting commit coordinator", taskId);
+ int topicPartitionCount = 0;
+ for (String topic : subscribedTopics) {
+ List<PartitionInfo> partitions = sourceConsumer().partitionsFor(topic);
+ if (partitions != null) {
Review Comment:
I think there's a data-loss path here when topic metadata isn't cached yet.
`partitionsFor(topic)` returns null while the consumer hasn't fetched
metadata for a topic (broker still loading, topic just created, first
assignment after a cold start), and we silently skip it — so
`topicPartitionCount` can come out 0 and the Coordinator is built expecting 0
partitions. With nothing to wait for, the first `DATA_COMPLETE` looks
commit-ready and we run a full commit while the other tasks' files are still
buffered, and those get dropped on `clearResponses`.
The old code summed partitions from the group members, which was
authoritative. I'd either restore an authoritative count or treat `count == 0`
as unknown and never take the full-commit path in that state — a timeout-based
partial commit is fine there, dropping files isn't. Could we guard that case?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -195,6 +193,14 @@ private void commit(boolean partialCommit) {
}
}
+ @VisibleForTesting
+ static boolean isRetryable(RuntimeException exception) {
+ return exception instanceof CommitFailedException
+ || exception instanceof
org.apache.kafka.clients.consumer.CommitFailedException
+ || exception instanceof
org.apache.kafka.common.errors.RebalanceInProgressException
+ || exception instanceof
org.apache.kafka.common.errors.RetriableException;
Review Comment:
Broadening retry to the whole `RetriableException` hierarchy is where I'd
want to be surgical.
Iceberg's `CommitFailedException`, the Kafka consumer
`CommitFailedException`, and `RebalanceInProgressException` are all safe to
retry. But a `TimeoutException` out of `producer.commitTransaction()` is
ambiguous — per the Kafka docs the broker may already have committed — so
retrying by starting a fresh transaction and committing again risks a double
write. Config-type `RetriableException`s like `NotEnoughReplicas*` will also
retry to the threshold while logging "will retry", which just misleads
operators.
I'd enumerate the exceptions we actually treat as transient for a
transactional commit rather than catch the superclass. Which ones did you
intend to be retryable here?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -192,13 +130,32 @@ public void save(Collection<SinkRecord> sinkRecords) {
startWorker();
worker.save(sinkRecords);
}
+ if (reconcileNeeded) {
+ reconcileLeadership();
+ reconcileNeeded = false;
+ }
processControlEvents();
}
+ private void reconcileLeadership() {
+ Set<String> subscribedTopics =
Sets.newTreeSet(sourceConsumer().subscription());
Review Comment:
This election looks deterministic for a static `topics` list but I think it
breaks for `topics.regex`.
`subscription()` returns only the topics this task instance currently
resolved/was assigned, not the full cluster set. With a static list every task
sees the same set, so `leaderPartition` agrees. With a regex, task A might see
`[event-aaa, event-bbb]` and task B `[event-ccc, ...]`, so they compute
different local-min topics and each can elect itself → two coordinators
overlapping, or zero if nobody holds partition 0 of their local min.
Election needs a global view of the subscribed topics for the regex case.
Could we resolve it from cluster state, or explicitly reject/document regex
here for now?
--
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]
