laskoviymishka commented on code in PR #17376:
URL: https://github.com/apache/iceberg/pull/17376#discussion_r3739164389
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -197,8 +210,15 @@ public void save(Collection<SinkRecord> sinkRecords) {
private void processControlEvents() {
if (coordinatorThread != null && coordinatorThread.isTerminated()) {
- throw new NotRunningException(
- String.format("Coordinator unexpectedly terminated on committer %s",
taskId));
+ if (isProducerFenced(coordinatorThread.exception())) {
+ // Lost the coordinator race (fenced by a newer coordinator). Clear it
so
+ // commit thread pool are released.
+ LOG.warn("Committer {} coordinator was fenced by a newer coordinator;
clearing it", taskId);
+ stopCoordinator();
Review Comment:
After the fence is detected we clear the coordinator and return normally —
but `startCoordinator()` is only ever called from `open()`, which only fires on
a Connect rebalance.
On a stable assignment (no rebalance coming) that leaves the task as a pure
worker with no coordinator: it keeps writing parquet and buffering, but no
`StartCommit` is ever broadcast, so no Iceberg commit happens and nothing
surfaces to Connect's task monitoring. The connector reads as "running" while
it has silently stopped committing, and if control-topic or source retention
expires while the buffer is held that's a data-loss window.
I get why you moved off `throw NotRunningException` (Connect won't
auto-restart a FAILED task), but the replacement is an unbounded, invisible
stall. I'd want either an active re-election trigger (e.g.
`context.requestCommit()` / a health hook) or, at minimum, an ERROR-level "no
coordinator active, commits suspended" signal plus an operator-guide note
quantifying the stall window. How were you picturing recovery here on a cluster
that isn't rebalancing?
##########
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:
The new coordinator transactional.id is a different string from the old
`transactionalPrefix + "coordinator" + transactionalSuffix`, and Kafka fencing
only kicks in when a new producer reuses the *same* id it's replacing.
So during a rolling upgrade an old coordinator (old id) and a new one (new
id) don't fence each other — both stay live for up to
`COORDINATOR_STOP_TIMEOUT_MS`, both broadcast `StartCommit` with different
UUIDs, and for a multi-table connector each can win a commit on a different
table in the same window. `SnapshotAncestryValidator` guards a single table
against the same base snapshot, but it won't catch two coordinators committing
*different* tables, so cross-table consistency can quietly break.
Which did you intend here — a migration that fences the old id explicitly
(init + abort on the old format before switching), or a hard "this upgrade
requires a full connector restart, not a rolling one"? Either is defensible,
but if it's the latter I'd want it enforced with a startup check and
documented, not left implicit. 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";
+
this.producer = clientFactory.createProducer(transactionalId);
- this.consumer = clientFactory.createConsumer(consumerGroupId);
+ this.consumer =
+ "coordinator".equalsIgnoreCase(name)
Review Comment:
The base class is branching on subclass name strings to pick both the
transactional.id format and the offset-reset policy, and the two checks are
asymmetric — the id branch keys off `"worker"` (else → coord id) while the
consumer branch keys off `"coordinator"` (else → latest). Any future subclass,
or a rename/typo, silently inherits the coordinator id with a `latest` reset,
both wrong.
I'd lift these out of the base class — pass `transactionalId` and
`autoOffsetReset` as explicit constructor params computed by each subclass
before `super(...)`, so the dispatch is a real contract instead of a string
match. Cheap to do now, and it removes the base-knows-subclass coupling.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -197,8 +210,15 @@ public void save(Collection<SinkRecord> sinkRecords) {
private void processControlEvents() {
if (coordinatorThread != null && coordinatorThread.isTerminated()) {
- throw new NotRunningException(
- String.format("Coordinator unexpectedly terminated on committer %s",
taskId));
+ if (isProducerFenced(coordinatorThread.exception())) {
Review Comment:
`coordinatorThread.exception()` can be null — `terminate()` sets `terminated
= true` but never calls `exception.set(...)`, so a coordinator that stops
without recording a throwable lands here with a null cause,
`isProducerFenced(null)` returns false, and we fall into the `else` and
hard-fail the task with `NotRunningException`.
So this branch treats "terminated but no recorded exception" as fatal, and
it can't distinguish a clean/requested stop from a genuine crash. Can you
confirm whether a requested terminate can ever reach this check with the field
still set? If it can, this crashes the task after a clean stop; if it can't
today, it's one refactor away. Either way I'd add an explicit null guard and
decide which side null falls on.
While you're here — the new `NotRunningException(String, Throwable)` ctor is
added but this throw site still uses the one-arg form, so the coordinator's
actual exception is dropped from the failure. Passing
`coordinatorThread.exception()` as the cause is presumably what the new ctor
was for.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -195,6 +196,14 @@ private void commit(boolean partialCommit) {
}
}
+ @VisibleForTesting
+ static boolean isRetryable(RuntimeException exception) {
Review Comment:
The interaction between this allowlist and the single
`consecutiveCommitFailures` counter is under-specified, and it bites from two
directions.
Kafka-side retryable errors (`RebalanceInProgressException`, consumer
`CommitFailedException`, any `RetriableException`) return true here, so they
retry — but note these are thrown from `commitConsumerOffsets()` *after* every
`commitToTable()` has already succeeded, and they share one budget with Iceberg
OCC conflicts. With `max-consecutive-failures = 1`, a single
`RebalanceInProgressException` terminates the coordinator even though every
table committed fine; conversely a permanent rebalance loop can retry unbounded
with no circuit breaker.
What semantics did you intend for a purely Kafka-side failure where the
Iceberg commit already landed — reset the counter (it's not an Iceberg-progress
failure), or a separate budget? I'd split the two rather than have OCC
conflicts and offset-commit hiccups share one threshold.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -109,6 +118,7 @@ boolean containsFirstPartition(
"Committer {} contains the first partition {}, this task is the
leader",
taskId,
firstTopicPartition);
+ leaderTopicPartition.set(firstTopicPartition);
Review Comment:
`leaderTopicPartition` is only ever set, never cleared. If a task was leader
(owned P0), loses P0, then `open()` runs again without P0,
`containsFirstPartition` returns false and leaves the stale P0 in place — so a
later `close()` seeing P0 fires `stopCoordinator()` for a coordinator this
epoch never started.
The null guard keeps it harmless, but the "lost leader partition" log then
lies. I'd `leaderTopicPartition.set(null)` in the `else` branch (or in
`stopCoordinator()`) so the field tracks actual leadership.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -284,6 +284,62 @@ public void testCommitBoundedRetryWithMultipleThreads() {
.hasMessageContaining("concurrent update");
}
+ @Test
+ public void testIsRetryableClassifiesCommitAndTransientKafkaErrors() {
+ // Iceberg optimistic-concurrency failure -> retry
+ assertThat(Coordinator.isRetryable(new
CommitFailedException("occ"))).isTrue();
Review Comment:
Now that `CommitFailedException` is retryable, the existing
`testCommitFailedExceptionPropagates` passes for a different reason than its
name — it only propagates because the mock sets `max-consecutive-failures = 1`,
exhausting the budget on the first try, not because the exception is
non-retryable.
I'd rename it to something like
`testCommitFailedExceptionPropagatesAfterThreshold` and assert the retryable
path was taken first, so the test validates the new invariant rather than
passing by coincidence.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -232,9 +255,48 @@ private void stopWorker() {
}
private void stopCoordinator() {
- if (coordinatorThread != null) {
- coordinatorThread.terminate();
- coordinatorThread = null;
+ CoordinatorThread thread = coordinatorThread;
+ if (thread == null) {
+ return;
+ }
+ coordinatorThread = null;
+
+ try {
+ if (!thread.isTerminated()) {
+ thread.terminate();
+ }
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Committer {}: error signalling coordinator termination, continuing
shutdown", taskId, e);
+ }
+
+ try {
+ thread.join(COORDINATOR_STOP_TIMEOUT_MS);
Review Comment:
This `join(60s)` runs on the Connect task thread — `close()` is called on
the poll thread, and `open()` (the next leader election) only runs after
`close()` returns. So a stuck coordinator blocks the task for up to 60s during
a rebalance, and the consumer can't poll in that window; if
`max.poll.interval.ms` is under 60s Connect fences this consumer and triggers
another rebalance, which can cascade.
It's also stacked: `CoordinatorThread.terminate()` internally does
`awaitTermination(1, MINUTES)`, so worst case is ~120s total, which can blow
past Connect's task stop deadline (5s default) and get the task
force-interrupted mid-shutdown.
I'd bound the outer join well under `max.poll.interval.ms` (say 5–10s), keep
the background-shutdown fallback for the rare stuck case, and make sure the
outer timeout is shorter than the inner `awaitTermination` so it can actually
detect the inner hang. Was 60s picked deliberately, or carried over?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -284,6 +284,62 @@ public void testCommitBoundedRetryWithMultipleThreads() {
.hasMessageContaining("concurrent update");
}
+ @Test
+ public void testIsRetryableClassifiesCommitAndTransientKafkaErrors() {
Review Comment:
These classification and dedup tests are good, but the behavior this PR is
actually about — `processControlEvents()` clearing a fenced coordinator instead
of failing the task — has no test.
I'd add one that starts a `CoordinatorThread`, injects a
`ProducerFencedException` (or `InvalidProducerEpochException`) as the thread's
exception and marks it terminated, calls `processControlEvents()`, and asserts
`coordinatorThread == null` with no `NotRunningException` thrown — and the
mirror case (terminated with a non-fencing exception) asserting it *does*
throw. That locks both sides of the branch, including the null-exception edge
flagged in `processControlEvents`.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -232,9 +255,48 @@ private void stopWorker() {
}
private void stopCoordinator() {
- if (coordinatorThread != null) {
- coordinatorThread.terminate();
- coordinatorThread = null;
+ CoordinatorThread thread = coordinatorThread;
+ if (thread == null) {
+ return;
+ }
+ coordinatorThread = null;
Review Comment:
We null `coordinatorThread` before the join completes, which opens a window.
If the join is interrupted we hit the early `return` with the field already
null but the old thread still alive and still holding its producer/consumer — a
following `open()` → `startCoordinator()` then spins up a second coordinator on
the same transactional.id, overlapping the one we never confirmed dead.
The early return also skips the `thread.isAlive()` warning below, so
operators lose the "coordinator still running in the background" signal
precisely on the interrupted-shutdown path where it matters most.
I'd only null the field after the join resolves (use the local `thread` ref
plus a boolean to block a second `startCoordinator()` during the wait), and
move the `isAlive()` check so the interrupt path still logs it.
##########
kafka-connect/kafka-connect-runtime/src/integration/java/org/apache/iceberg/connect/IntegrationTestBase.java:
##########
@@ -225,7 +225,7 @@ protected void runTest(
flush();
Awaitility.await()
- .atMost(Duration.ofSeconds(30))
+ .atMost(Duration.ofSeconds(60))
Review Comment:
Doubling this (and the matching one in `TestIntegrationDynamicTable`) as
"reduce CI flakiness" without a root cause makes future latency regressions
invisible. Three changes in this PR add commit-path latency — the bounded join
in `stopCoordinator()`, and especially the coordinator consumer now reading
from `earliest`, which replays control-topic history on `Channel.start()` and
gets worse as the topic grows in production.
If `earliest` replay is the culprit that's a real production concern, not
just a test-timing one. Can you measure which path slowed down and address that
(e.g. ensure the `-coord` group has committed offsets before the wait), then
restore or justify the timeout?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -232,9 +255,48 @@ private void stopWorker() {
}
private void stopCoordinator() {
- if (coordinatorThread != null) {
- coordinatorThread.terminate();
- coordinatorThread = null;
+ CoordinatorThread thread = coordinatorThread;
+ if (thread == null) {
+ return;
+ }
+ coordinatorThread = null;
+
+ try {
+ if (!thread.isTerminated()) {
+ thread.terminate();
+ }
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Committer {}: error signalling coordinator termination, continuing
shutdown", taskId, e);
+ }
+
+ try {
+ thread.join(COORDINATOR_STOP_TIMEOUT_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn(
+ "Committer {}: interrupted while waiting for the coordinator thread
to stop", taskId, e);
+ return;
+ }
+
+ if (thread.isAlive()) {
+ LOG.warn(
+ "Committer {}: coordinator thread did not stop within {} ms; it will
keep shutting down "
+ + "in the background (a newer coordinator fences its producer)",
+ taskId,
+ COORDINATOR_STOP_TIMEOUT_MS);
+ }
+ }
+
+ private static boolean isProducerFenced(Throwable cause) {
+ Throwable current = cause;
+ for (int depth = 0; current != null && depth < 20; depth++, current =
current.getCause()) {
+ if (current instanceof ProducerFencedException
+ || current instanceof InvalidProducerEpochException
+ || current instanceof UnknownProducerIdException) {
Review Comment:
`UnknownProducerIdException` isn't only a fencing signal — it also fires
when the broker's transactional state for this id expires
(`transactional.id.expiration.ms`) or on a broker restart, where the right
response is re-init, not "a newer coordinator won."
At minimum the log shouldn't assert "fenced by a newer coordinator" for this
case — something like "coordinator producer was invalidated (fenced or its
transactional state expired)" is honest. The open question is whether you'd
rather classify it separately as recoverable and re-init the producer instead
of clearing the coordinator — that's more work but avoids the silent-stall path
above for the expiry case. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -119,21 +129,27 @@ protected void send(List<Event> events,
Map<TopicPartition, Offset> sourceOffset
protected void consumeAvailable(Duration pollDuration) {
ConsumerRecords<String, byte[]> records = consumer.poll(pollDuration);
while (!records.isEmpty()) {
- records.forEach(
- record -> {
- // the consumer stores the offsets that corresponds to the next
record to consume,
- // so increment the record offset by one
- controlTopicOffsets.put(record.partition(), record.offset() + 1);
-
- Event event = AvroUtil.decode(record.value());
-
- if (event.groupId().equals(connectGroupId)) {
- LOG.debug("Received event of type: {}", event.type().name());
- if (receive(new Envelope(event, record.partition(),
record.offset()))) {
- LOG.info("Handled event of type: {}", event.type().name());
- }
- }
- });
+ for (ConsumerRecord<String, byte[]> record : records) {
+ if (record.offset() <
controlTopicOffsets.getOrDefault(record.partition(), 0L)) {
Review Comment:
Worth being precise in the comment about what this filter does, since the PR
frames it as dedup. `controlTopicOffsets` starts empty on every new
coordinator, so this only guards against a fetch-position regression *within* a
single coordinator session (e.g. an eager rebalance rewinding the position) —
it isn't cross-session idempotency.
The actual cross-session dedup comes from the `committedOffsets` comparison
in `commitToTable` plus `SnapshotAncestryValidator`. A one-line comment saying
"intra-session guard against position regression, not cross-session dedup"
would keep the next reader from over-trusting it.
--
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]
