laskoviymishka commented on code in PR #16084:
URL: https://github.com/apache/iceberg/pull/16084#discussion_r3659271098
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -109,17 +198,55 @@ protected boolean receive(Envelope envelope) {
events.add(readyEvent);
send(events, results.sourceOffsets());
+ }
- return true;
+ @Override
+ protected boolean receive(Envelope envelope) {
+ if (envelope.event().payload().type() == PayloadType.START_COMMIT) {
+ controlEventQueue.offer(envelope);
+ LOG.debug("Worker {} buffered START_COMMIT event", taskId);
+ return true;
+ }
+ return false;
}
@Override
void stop() {
- super.stop();
+ LOG.info("Worker {} stopping.", taskId);
+ terminateBackGroundPolling();
Review Comment:
`terminateBackGroundPolling()` running before `sinkWriter.close()` and
`super.stop()` means a throw here leaks everything downstream.
Walk the shutdown: if `awaitTermination` times out or gets interrupted,
`terminateBackGroundPolling()` throws `ConnectException` — and
`sinkWriter.close()` and `super.stop()` (consumer, producer, admin) never run.
We leak the transactional producer with a possibly-open transaction, which can
then block the next task start, plus any buffered writer file handles. The
timeout path is most likely exactly when we're under a rebalance storm.
I'd wrap this in try/finally so cleanup always runs:
```java
void stop() {
LOG.info("Worker {} stopping.", taskId);
try {
terminateBackGroundPolling();
} finally {
sinkWriter.close();
super.stop();
LOG.info("Worker {} stopped.", taskId);
}
}
```
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -58,19 +79,89 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
- }
- void process() {
- consumeAvailable(Duration.ZERO);
+ this.taskId = config.connectorName() + "-" + config.taskId();
+ this.controlEventQueue = new ConcurrentLinkedQueue<>();
+ this.running = new AtomicBoolean(false);
+ this.pollInterval = Duration.ofMillis(config.controlPollIntervalMs());
+ this.pollingExecutor =
+ Executors.newSingleThreadExecutor(
+ r -> {
+ Thread thread = new Thread(r, "worker-control-poller-" + taskId);
+ thread.setDaemon(true);
+ return thread;
+ });
}
@Override
- protected boolean receive(Envelope envelope) {
- Event event = envelope.event();
- if (event.payload().type() != PayloadType.START_COMMIT) {
- return false;
+ void start() {
+ // Do NOT call super.start() — all consumer access must happen on the
background thread
+ // to satisfy KafkaConsumer's single-thread requirement. The background
thread calls
+ // initializeConsumer() as its first action.
+ running.set(true);
+
+ try {
+ pollingExecutor.execute(this::backgroundPoll);
Review Comment:
There's a startup window where `process()` can run before the background
thread has subscribed.
`start()` sets `running=true`, submits `backgroundPoll`, and returns
immediately — but `backgroundPoll` hasn't necessarily reached
`initializeConsumer()` yet. The next `save()` → `processControlEvents()` →
`worker.process()` on the main thread then finds an empty queue and no error,
so it silently no-ops, and a `START_COMMIT` already sitting on the broker isn't
picked up until a later cycle. Given the whole point of the PR is "zero missed
START_COMMIT at startup," this is the exact gap it's trying to close. A
`CountDownLatch(1)` counted down at the end of `initializeConsumer()` and
awaited (with a timeout) in `start()` would close it.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -58,19 +79,89 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
- }
- void process() {
- consumeAvailable(Duration.ZERO);
+ this.taskId = config.connectorName() + "-" + config.taskId();
+ this.controlEventQueue = new ConcurrentLinkedQueue<>();
+ this.running = new AtomicBoolean(false);
+ this.pollInterval = Duration.ofMillis(config.controlPollIntervalMs());
+ this.pollingExecutor =
+ Executors.newSingleThreadExecutor(
+ r -> {
+ Thread thread = new Thread(r, "worker-control-poller-" + taskId);
+ thread.setDaemon(true);
+ return thread;
+ });
}
@Override
- protected boolean receive(Envelope envelope) {
- Event event = envelope.event();
- if (event.payload().type() != PayloadType.START_COMMIT) {
- return false;
+ void start() {
+ // Do NOT call super.start() — all consumer access must happen on the
background thread
+ // to satisfy KafkaConsumer's single-thread requirement. The background
thread calls
+ // initializeConsumer() as its first action.
+ running.set(true);
+
+ try {
+ pollingExecutor.execute(this::backgroundPoll);
+ } catch (Exception ex) {
+ LOG.error("Worker {} failed to execute the task.", taskId, ex);
+ throw new ConnectException(
+ String.format("Worker %s failed to execute the poll task", taskId));
+ }
+ LOG.info(
+ "Worker {} started with async control event processing (poll interval:
{}ms)",
+ taskId,
+ pollInterval.toMillis());
+ }
+
+ /**
+ * Background polling task that subscribes to the control topic and
continuously polls for events,
+ * buffering them for processing by the main thread. All KafkaConsumer
access is confined to this
+ * thread.
+ */
+ private void backgroundPoll() {
+ LOG.info("Background control topic polling thread started on {}", taskId);
+ try {
+ // Initialize consumer on this thread — KafkaConsumer is NOT thread-safe,
+ // so subscribe + all poll calls must happen on the same thread.
+ initializeConsumer();
+
+ while (running.get() && !Thread.currentThread().isInterrupted()) {
+ try {
+ consumeAvailable(pollInterval);
+ } catch (WakeupException e) {
+ // Expected during shutdown — wakeupConsumer() was called
+ LOG.debug("Consumer wakeup received during shutdown for {}", taskId,
e);
+ break;
+ }
+ }
+ } catch (Exception e) {
Review Comment:
This outer `catch (Exception e)` makes a single bad record fatal to the
whole poller.
Any exception escaping the loop — including an `AvroUtil.decode()` failure
on one poison-pill control-topic record — sets `errorRef`, exits the thread,
and stops consuming. Until the next `put()` observes the error (bounded by
roughly `commit.timeout-ms`), the control topic isn't being read, so a
coordinator `START_COMMIT` in that window gets no response and times out to a
partial commit. In the old synchronous design that decode error surfaced
immediately and the task restarted promptly. I'd have `backgroundPoll` skip a
single undecodable record and continue rather than treating it as terminal —
see the related note on the recovery path in `CommitterImpl`.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -109,17 +198,55 @@ protected boolean receive(Envelope envelope) {
events.add(readyEvent);
send(events, results.sourceOffsets());
+ }
- return true;
+ @Override
+ protected boolean receive(Envelope envelope) {
+ if (envelope.event().payload().type() == PayloadType.START_COMMIT) {
+ controlEventQueue.offer(envelope);
+ LOG.debug("Worker {} buffered START_COMMIT event", taskId);
+ return true;
+ }
+ return false;
}
@Override
void stop() {
- super.stop();
+ LOG.info("Worker {} stopping.", taskId);
+ terminateBackGroundPolling();
sinkWriter.close();
+ super.stop();
+ LOG.info("Worker {} stopped.", taskId);
+ }
+
+ private void terminateBackGroundPolling() {
+ running.set(false);
+ wakeupConsumer();
+ pollingExecutor.shutdownNow();
+ try {
+ if (!pollingExecutor.awaitTermination(1, TimeUnit.MINUTES)) {
+ LOG.warn("Polling thread did not terminate in time on worker {},
forcing shutdown", taskId);
+ throw new ConnectException(
+ String.format(
+ "Background polling thread of worker %s did not terminate
gracefully.", taskId));
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Worker {} got interrupted while waiting for polling thread
shutdown", taskId, e);
+ throw new ConnectException(
+ String.format(
+ "Background polling thread of worker %s interrupted while
closing.", taskId),
+ e);
+ }
+ controlEventQueue.clear();
+ errorRef.set(null);
Review Comment:
Minor, and tied to the error-handling above: `errorRef.set(null)` on
teardown throws away the diagnostic right when it's most useful, and if a
restart path ever gets added this would hide the error that caused the stop.
I'd reset these in `start()` instead, if/when there's a restart path — not a
blocker.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -109,17 +198,55 @@ protected boolean receive(Envelope envelope) {
events.add(readyEvent);
send(events, results.sourceOffsets());
+ }
- return true;
+ @Override
+ protected boolean receive(Envelope envelope) {
+ if (envelope.event().payload().type() == PayloadType.START_COMMIT) {
+ controlEventQueue.offer(envelope);
+ LOG.debug("Worker {} buffered START_COMMIT event", taskId);
+ return true;
+ }
+ return false;
}
@Override
void stop() {
- super.stop();
+ LOG.info("Worker {} stopping.", taskId);
+ terminateBackGroundPolling();
sinkWriter.close();
+ super.stop();
+ LOG.info("Worker {} stopped.", taskId);
+ }
+
+ private void terminateBackGroundPolling() {
+ running.set(false);
+ wakeupConsumer();
+ pollingExecutor.shutdownNow();
+ try {
+ if (!pollingExecutor.awaitTermination(1, TimeUnit.MINUTES)) {
Review Comment:
Throwing out of `stop()` feels risky for a SinkTask.
Kafka Connect calls `SinkTask.stop()` on shutdown and doesn't expect it to
throw; an unchecked exception here propagates through
`CommitterImpl.close()`/`stopWorker()` and can leave the coordinator thread and
producer un-closed. I'd log the timeout at ERROR and continue with cleanup
rather than throwing. Same for the `InterruptedException` branch just below —
and there we're also clearing the interrupt flag without restoring it; I'd add
`Thread.currentThread().interrupt()` before proceeding.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -58,19 +79,89 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
- }
- void process() {
- consumeAvailable(Duration.ZERO);
+ this.taskId = config.connectorName() + "-" + config.taskId();
+ this.controlEventQueue = new ConcurrentLinkedQueue<>();
+ this.running = new AtomicBoolean(false);
+ this.pollInterval = Duration.ofMillis(config.controlPollIntervalMs());
+ this.pollingExecutor =
+ Executors.newSingleThreadExecutor(
+ r -> {
+ Thread thread = new Thread(r, "worker-control-poller-" + taskId);
+ thread.setDaemon(true);
+ return thread;
+ });
}
@Override
- protected boolean receive(Envelope envelope) {
- Event event = envelope.event();
- if (event.payload().type() != PayloadType.START_COMMIT) {
- return false;
+ void start() {
+ // Do NOT call super.start() — all consumer access must happen on the
background thread
+ // to satisfy KafkaConsumer's single-thread requirement. The background
thread calls
+ // initializeConsumer() as its first action.
+ running.set(true);
+
+ try {
+ pollingExecutor.execute(this::backgroundPoll);
+ } catch (Exception ex) {
+ LOG.error("Worker {} failed to execute the task.", taskId, ex);
+ throw new ConnectException(
+ String.format("Worker %s failed to execute the poll task", taskId));
+ }
+ LOG.info(
+ "Worker {} started with async control event processing (poll interval:
{}ms)",
+ taskId,
+ pollInterval.toMillis());
+ }
+
+ /**
+ * Background polling task that subscribes to the control topic and
continuously polls for events,
+ * buffering them for processing by the main thread. All KafkaConsumer
access is confined to this
+ * thread.
+ */
+ private void backgroundPoll() {
+ LOG.info("Background control topic polling thread started on {}", taskId);
+ try {
+ // Initialize consumer on this thread — KafkaConsumer is NOT thread-safe,
+ // so subscribe + all poll calls must happen on the same thread.
+ initializeConsumer();
+
+ while (running.get() && !Thread.currentThread().isInterrupted()) {
+ try {
+ consumeAvailable(pollInterval);
+ } catch (WakeupException e) {
+ // Expected during shutdown — wakeupConsumer() was called
+ LOG.debug("Consumer wakeup received during shutdown for {}", taskId,
e);
+ break;
+ }
+ }
+ } catch (Exception e) {
+ if (running.compareAndSet(true, false)) {
+ LOG.error("Worker {} failed while polling control events", taskId, e);
+ errorRef.compareAndSet(null, e);
+ }
+ } finally {
+ LOG.info("Background control topic polling thread stopped on {}",
taskId);
}
+ }
+
+ /**
+ * Process all available events from the queue (non-blocking). This is
called from the main put()
+ * path. Throws ConnectException if the background thread encountered an
error — the caller is
+ * responsible for stopping the worker.
+ */
+ void process() {
+ Exception ex = errorRef.getAndSet(null);
+ if (ex != null) {
+ throw new ConnectException(
+ String.format("Worker %s failed while processing async polling.",
taskId), ex);
+ }
+ Envelope envelope;
+ while ((envelope = controlEventQueue.poll()) != null) {
+ handleStartCommit(((StartCommit) envelope.event().payload()).commitId());
Review Comment:
This cast is safe today but the queue's contract is implicit. `receive()`
only enqueues `START_COMMIT`, so nothing else can reach here now — but the
guarantee lives one method away, and a future change would turn this into a
`ClassCastException` rather than a clean, attributable failure. A small
`instanceof` check with a warn-and-skip on an unexpected payload type would
make the drain boundary self-defending.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -186,7 +186,17 @@ private void processControlEvents() {
config.connectorName(), config.taskId()));
}
if (worker != null) {
- worker.process();
+ try {
+ worker.process();
+ } catch (Exception e) {
+ LOG.error(
+ "Worker {}-{} failed during control event processing, stopping
worker.",
+ config.connectorName(),
+ config.taskId(),
+ e);
+ stopWorker();
Review Comment:
This is the one I'd most want to nail down before merge — I think the
`stopWorker()`-then-restart path can silently drop data.
When `process()` throws, we `stopWorker()` → `worker.stop()` →
`terminateBackGroundPolling()` clears the queue, and `sinkWriter.close()`
throws away the in-memory writer state. A fresh worker then answers the
still-outstanding `START_COMMIT` with an empty `completeWrite()`, so the
coordinator commits a snapshot that's missing this task's already-written
files. The resulting snapshot is a perfectly valid Iceberg snapshot — it just
quietly has fewer rows, which no downstream reader can detect. This is one of
the genuinely hard KC error-recovery cases.
A couple of ways to handle: let the `ConnectException` propagate untouched
so the framework restarts the task from the last committed Kafka offset
(re-reading and re-writing the uncommitted records), rather than doing a silent
in-process worker swap — or pair that with making `backgroundPoll` non-fatal on
single-record decode errors so we rarely hit this at all. Which recovery model
do you want here?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -58,19 +79,89 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
- }
- void process() {
- consumeAvailable(Duration.ZERO);
+ this.taskId = config.connectorName() + "-" + config.taskId();
+ this.controlEventQueue = new ConcurrentLinkedQueue<>();
+ this.running = new AtomicBoolean(false);
+ this.pollInterval = Duration.ofMillis(config.controlPollIntervalMs());
+ this.pollingExecutor =
+ Executors.newSingleThreadExecutor(
+ r -> {
+ Thread thread = new Thread(r, "worker-control-poller-" + taskId);
+ thread.setDaemon(true);
+ return thread;
+ });
}
@Override
- protected boolean receive(Envelope envelope) {
- Event event = envelope.event();
- if (event.payload().type() != PayloadType.START_COMMIT) {
- return false;
+ void start() {
+ // Do NOT call super.start() — all consumer access must happen on the
background thread
+ // to satisfy KafkaConsumer's single-thread requirement. The background
thread calls
+ // initializeConsumer() as its first action.
+ running.set(true);
+
+ try {
+ pollingExecutor.execute(this::backgroundPoll);
+ } catch (Exception ex) {
+ LOG.error("Worker {} failed to execute the task.", taskId, ex);
+ throw new ConnectException(
+ String.format("Worker %s failed to execute the poll task", taskId));
+ }
+ LOG.info(
+ "Worker {} started with async control event processing (poll interval:
{}ms)",
+ taskId,
+ pollInterval.toMillis());
+ }
+
+ /**
+ * Background polling task that subscribes to the control topic and
continuously polls for events,
+ * buffering them for processing by the main thread. All KafkaConsumer
access is confined to this
+ * thread.
+ */
+ private void backgroundPoll() {
+ LOG.info("Background control topic polling thread started on {}", taskId);
+ try {
+ // Initialize consumer on this thread — KafkaConsumer is NOT thread-safe,
+ // so subscribe + all poll calls must happen on the same thread.
+ initializeConsumer();
+
+ while (running.get() && !Thread.currentThread().isInterrupted()) {
+ try {
+ consumeAvailable(pollInterval);
+ } catch (WakeupException e) {
+ // Expected during shutdown — wakeupConsumer() was called
+ LOG.debug("Consumer wakeup received during shutdown for {}", taskId,
e);
Review Comment:
Passing `e` as the third arg makes SLF4J log the full stack for what's a
fully expected shutdown wakeup — that's noise in verbose logs. Drop the `e`.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/TestIcebergSinkConfig.java:
##########
@@ -110,4 +110,52 @@ public void testCheckClassName() {
result =
IcebergSinkConfig.checkClassName("org.apache.kafka.clients.producer.KafkaProducer");
assertThat(result).isFalse();
}
+
+ @Test
+ public void testControlPollIntervalMsDefault() {
+ Map<String, String> props =
+ ImmutableMap.of(
+ "iceberg.catalog.type", "rest",
+ "topics", "source-topic",
+ "iceberg.tables", "db.landing");
+ IcebergSinkConfig config = new IcebergSinkConfig(props);
+ assertThat(config.controlPollIntervalMs()).isEqualTo(100);
+ }
+
+ @Test
+ public void testControlPollIntervalMsCustom() {
+ Map<String, String> props =
+ ImmutableMap.of(
+ "iceberg.catalog.type", "rest",
+ "topics", "source-topic",
+ "iceberg.tables", "db.landing",
+ "iceberg.control.poll.interval-ms", "500");
+ IcebergSinkConfig config = new IcebergSinkConfig(props);
+ assertThat(config.controlPollIntervalMs()).isEqualTo(500);
+ }
+
+ @Test
+ public void testControlPollIntervalMsZero() {
Review Comment:
Since the constraint is `atLeast(10)`, the interesting boundary is 9
(reject) and 10 (accept) — `0` is well clear of the edge. Worth a case at
exactly 9 and exactly 10.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -111,6 +119,270 @@ public void testSave() {
assertThat(dataComplete.commitId()).isEqualTo(commitId);
assertThat(dataComplete.assignments()).hasSize(1);
assertThat(dataComplete.assignments().get(0).offset()).isEqualTo(1L);
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testBackgroundPollingBuffersEvents() {
+ when(config.catalogName()).thenReturn("catalog");
+ when(config.controlPollIntervalMs()).thenReturn(50);
+
+ try (MockedStatic<KafkaUtils> mockKafkaUtils =
mockStatic(KafkaUtils.class)) {
+ ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ mockKafkaUtils
+ .when(() -> KafkaUtils.consumerGroupMetadata(any()))
+ .thenReturn(consumerGroupMetadata);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ TopicPartition topicPartition = new TopicPartition(SRC_TOPIC_NAME, 0);
+ when(context.assignment()).thenReturn(ImmutableSet.of(topicPartition));
+
+ SinkWriter sinkWriter = mock(SinkWriter.class);
+ when(sinkWriter.completeWrite())
+ .thenReturn(new SinkWriterResult(ImmutableList.of(),
ImmutableMap.of()));
+
+ initConsumer();
+
+ Worker worker = new Worker(config, clientFactory, sinkWriter, context);
+ worker.start();
+
+ // Add multiple events to consumer
+ UUID commitId1 = UUID.randomUUID();
+ Event event1 = new Event(config.connectGroupId(), new
StartCommit(commitId1));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(event1)));
+
+ UUID commitId2 = UUID.randomUUID();
+ Event event2 = new Event(config.connectGroupId(), new
StartCommit(commitId2));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(event2)));
+
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(5))
+ .pollInterval(Duration.ofMillis(10))
+ .until(() -> worker.pendingEventCount() >= 2);
+
+ // Process should handle both buffered events
+ worker.process();
+
+ // Should have 2 DATA_COMPLETE events (one per commit)
+ assertThat(producer.history().size()).isGreaterThanOrEqualTo(2);
Review Comment:
`isGreaterThanOrEqualTo(2)` lets a spurious extra event pass. You inject
exactly 2 `START_COMMIT` with an empty `completeWrite()`, so it's exactly 2
`DATA_COMPLETE` — `assertThat(producer.history()).hasSize(2)` will catch a
regression this won't.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -109,17 +198,55 @@ protected boolean receive(Envelope envelope) {
events.add(readyEvent);
send(events, results.sourceOffsets());
+ }
- return true;
+ @Override
+ protected boolean receive(Envelope envelope) {
+ if (envelope.event().payload().type() == PayloadType.START_COMMIT) {
+ controlEventQueue.offer(envelope);
+ LOG.debug("Worker {} buffered START_COMMIT event", taskId);
+ return true;
+ }
+ return false;
}
@Override
void stop() {
- super.stop();
+ LOG.info("Worker {} stopping.", taskId);
+ terminateBackGroundPolling();
sinkWriter.close();
+ super.stop();
+ LOG.info("Worker {} stopped.", taskId);
+ }
+
+ private void terminateBackGroundPolling() {
Review Comment:
Tiny naming nit: `terminateBackGroundPolling` has a stray capital —
`Background` is one word. Rename here and at the call site.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -58,19 +79,89 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
- }
- void process() {
- consumeAvailable(Duration.ZERO);
+ this.taskId = config.connectorName() + "-" + config.taskId();
+ this.controlEventQueue = new ConcurrentLinkedQueue<>();
+ this.running = new AtomicBoolean(false);
+ this.pollInterval = Duration.ofMillis(config.controlPollIntervalMs());
+ this.pollingExecutor =
+ Executors.newSingleThreadExecutor(
+ r -> {
+ Thread thread = new Thread(r, "worker-control-poller-" + taskId);
+ thread.setDaemon(true);
+ return thread;
+ });
}
@Override
- protected boolean receive(Envelope envelope) {
- Event event = envelope.event();
- if (event.payload().type() != PayloadType.START_COMMIT) {
- return false;
+ void start() {
+ // Do NOT call super.start() — all consumer access must happen on the
background thread
+ // to satisfy KafkaConsumer's single-thread requirement. The background
thread calls
+ // initializeConsumer() as its first action.
+ running.set(true);
+
+ try {
+ pollingExecutor.execute(this::backgroundPoll);
+ } catch (Exception ex) {
+ LOG.error("Worker {} failed to execute the task.", taskId, ex);
+ throw new ConnectException(
Review Comment:
This `ConnectException` drops the original cause — the `LOG.error` captures
`ex`, but the thrown exception has a null cause chain. Pass `ex` as the second
arg.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -111,6 +119,270 @@ public void testSave() {
assertThat(dataComplete.commitId()).isEqualTo(commitId);
assertThat(dataComplete.assignments()).hasSize(1);
assertThat(dataComplete.assignments().get(0).offset()).isEqualTo(1L);
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testBackgroundPollingBuffersEvents() {
+ when(config.catalogName()).thenReturn("catalog");
+ when(config.controlPollIntervalMs()).thenReturn(50);
+
+ try (MockedStatic<KafkaUtils> mockKafkaUtils =
mockStatic(KafkaUtils.class)) {
+ ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ mockKafkaUtils
+ .when(() -> KafkaUtils.consumerGroupMetadata(any()))
+ .thenReturn(consumerGroupMetadata);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ TopicPartition topicPartition = new TopicPartition(SRC_TOPIC_NAME, 0);
+ when(context.assignment()).thenReturn(ImmutableSet.of(topicPartition));
+
+ SinkWriter sinkWriter = mock(SinkWriter.class);
+ when(sinkWriter.completeWrite())
+ .thenReturn(new SinkWriterResult(ImmutableList.of(),
ImmutableMap.of()));
+
+ initConsumer();
+
+ Worker worker = new Worker(config, clientFactory, sinkWriter, context);
+ worker.start();
+
+ // Add multiple events to consumer
+ UUID commitId1 = UUID.randomUUID();
+ Event event1 = new Event(config.connectGroupId(), new
StartCommit(commitId1));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(event1)));
+
+ UUID commitId2 = UUID.randomUUID();
+ Event event2 = new Event(config.connectGroupId(), new
StartCommit(commitId2));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(event2)));
+
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(5))
+ .pollInterval(Duration.ofMillis(10))
+ .until(() -> worker.pendingEventCount() >= 2);
+
+ // Process should handle both buffered events
+ worker.process();
+
+ // Should have 2 DATA_COMPLETE events (one per commit)
+ assertThat(producer.history().size()).isGreaterThanOrEqualTo(2);
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerIgnoresNonRelevantEvents() {
+ when(config.catalogName()).thenReturn("catalog");
+
+ try (MockedStatic<KafkaUtils> mockKafkaUtils =
mockStatic(KafkaUtils.class)) {
+ ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ mockKafkaUtils
+ .when(() -> KafkaUtils.consumerGroupMetadata(any()))
+ .thenReturn(consumerGroupMetadata);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ when(context.assignment()).thenReturn(ImmutableSet.of());
+
+ SinkWriter sinkWriter = mock(SinkWriter.class);
+
+ initConsumer();
+
+ Worker worker = new Worker(config, clientFactory, sinkWriter, context);
+ worker.start();
+
+ // Add events with different group IDs (should be ignored by Channel's
group filter)
+ UUID commitId = UUID.randomUUID();
+ Event event = new Event("different-group-id", new StartCommit(commitId));
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(event)));
+
+ // Also add a non-START_COMMIT event with correct group (should be
ignored by Worker)
+ Event commitComplete =
+ new Event(config.connectGroupId(), new CommitComplete(commitId,
EventTestUtil.now()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(commitComplete)));
+
+ // Wait a bit for the background thread to process the records
+ Awaitility.await()
+ .pollDelay(Duration.ofMillis(200))
+ .atMost(Duration.ofSeconds(5))
+ .until(() -> true);
+
+ worker.process();
+
+ // Should not produce any events since no matching START_COMMIT was
received
+ assertThat(producer.history()).isEmpty();
+ assertThat(worker.pendingEventCount()).isEqualTo(0);
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerGracefulShutdown() {
+ when(config.catalogName()).thenReturn("catalog");
+
+ try (MockedStatic<KafkaUtils> mockKafkaUtils =
mockStatic(KafkaUtils.class)) {
+ ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ mockKafkaUtils
+ .when(() -> KafkaUtils.consumerGroupMetadata(any()))
+ .thenReturn(consumerGroupMetadata);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ when(context.assignment()).thenReturn(ImmutableSet.of());
+
+ SinkWriter sinkWriter = mock(SinkWriter.class);
+
+ initConsumer();
+
+ Worker worker = new Worker(config, clientFactory, sinkWriter, context);
+ worker.start();
+
+ // Stop worker immediately — should complete without exceptions
+ worker.stop();
+
+ assertThat(producer.history()).isEmpty();
+ }
+ }
+
+ @Test
+ public void testWorkerHandlesEmptyQueue() {
+ when(config.catalogName()).thenReturn("catalog");
+
+ try (MockedStatic<KafkaUtils> mockKafkaUtils =
mockStatic(KafkaUtils.class)) {
+ ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ mockKafkaUtils
+ .when(() -> KafkaUtils.consumerGroupMetadata(any()))
+ .thenReturn(consumerGroupMetadata);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ when(context.assignment()).thenReturn(ImmutableSet.of());
+
+ SinkWriter sinkWriter = mock(SinkWriter.class);
+
+ initConsumer();
+
+ Worker worker = new Worker(config, clientFactory, sinkWriter, context);
+ worker.start();
+
+ // Call process multiple times with no events
+ worker.process();
+ worker.process();
+ worker.process();
+
+ assertThat(producer.history()).isEmpty();
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerMultipleStartCommits() {
+ when(config.catalogName()).thenReturn("catalog");
+
+ try (MockedStatic<KafkaUtils> mockKafkaUtils =
mockStatic(KafkaUtils.class)) {
+ ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ mockKafkaUtils
+ .when(() -> KafkaUtils.consumerGroupMetadata(any()))
+ .thenReturn(consumerGroupMetadata);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ TopicPartition topicPartition = new TopicPartition(SRC_TOPIC_NAME, 0);
+ when(context.assignment()).thenReturn(ImmutableSet.of(topicPartition));
+
+ IcebergWriterResult writeResult1 =
+ new IcebergWriterResult(
+ TableIdentifier.parse(TABLE_NAME),
+ ImmutableList.of(EventTestUtil.createDataFile()),
+ ImmutableList.of(),
+ StructType.of());
+
+ IcebergWriterResult writeResult2 =
+ new IcebergWriterResult(
+ TableIdentifier.parse(TABLE_NAME),
+ ImmutableList.of(EventTestUtil.createDataFile()),
+ ImmutableList.of(),
+ StructType.of());
+
+ Map<TopicPartition, Offset> offsets =
+ ImmutableMap.of(topicPartition, new Offset(1L, EventTestUtil.now()));
+
+ SinkWriter sinkWriter = mock(SinkWriter.class);
+ when(sinkWriter.completeWrite())
+ .thenReturn(new SinkWriterResult(ImmutableList.of(writeResult1),
offsets))
+ .thenReturn(new SinkWriterResult(ImmutableList.of(writeResult2),
offsets));
+
+ initConsumer();
+
+ Worker worker = new Worker(config, clientFactory, sinkWriter, context);
+ worker.start();
+
+ // Add multiple START_COMMIT events
+ UUID commitId1 = UUID.randomUUID();
+ Event event1 = new Event(config.connectGroupId(), new
StartCommit(commitId1));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(event1)));
+
+ UUID commitId2 = UUID.randomUUID();
+ Event event2 = new Event(config.connectGroupId(), new
StartCommit(commitId2));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(event2)));
+
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(5))
+ .pollInterval(Duration.ofMillis(10))
+ .until(() -> worker.pendingEventCount() >= 2);
+
+ // Process both commits
+ worker.process();
+
+ // Should have events for both commits (2 data written + 2 data complete)
+ assertThat(producer.history()).hasSize(4);
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testBackgroundPollingErrorPropagation() {
Review Comment:
Two gaps in this test. It never calls `worker.stop()`, so if the Awaitility
condition times out it exits leaving a daemon thread holding the `MockConsumer`
the next test reuses (`ChannelTestBase.after()` doesn't close it). And it
checks that the *first* `process()` throws but never asserts what the *second*
call does — which is exactly the `getAndSet(null)`-clears-the-error behavior I
flagged in `Worker.process()`, the most fragile part of the design and
currently untested. I'd wrap every worker-starting test body in try/finally
with `worker.stop()`, and add an explicit "call `process()` twice after a
background failure" assertion.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -58,19 +79,89 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
- }
- void process() {
- consumeAvailable(Duration.ZERO);
+ this.taskId = config.connectorName() + "-" + config.taskId();
+ this.controlEventQueue = new ConcurrentLinkedQueue<>();
+ this.running = new AtomicBoolean(false);
+ this.pollInterval = Duration.ofMillis(config.controlPollIntervalMs());
+ this.pollingExecutor =
Review Comment:
Design note, not a blocker: a per-worker `newSingleThreadExecutor` means
24–48 pools and threads at the scale you mention. A single shared executor with
one thread per worker would trim the overhead, at the cost of a little coupling
(managing it at `CommitterImpl` level). Fine as a first cut — flagging for
later.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -51,7 +51,7 @@ abstract class Channel {
private final Consumer<String, byte[]> consumer;
private final SinkTaskContext context;
private final Admin admin;
- private final Map<Integer, Long> controlTopicOffsets = Maps.newHashMap();
+ private final Map<Integer, Long> controlTopicOffsets =
Maps.newConcurrentMap();
Review Comment:
What cross-thread access is the `ConcurrentMap` switch meant to guard?
As far as I can trace it, the Worker's main thread never reads
`controlTopicOffsets` (Worker doesn't call `controlTopicOffsets()` anywhere),
and the Coordinator has its own `Channel` instance that runs single-threaded on
`CoordinatorThread`. So on the Worker path this doesn't actually protect
anything, and I worry it reads as safety where the real cross-thread boundary
is elsewhere. If there's a Worker read I'm missing, a one-line comment pointing
at it would help; otherwise I'd lean toward reverting this and keeping the map
where it's genuinely single-threaded. (Separately — not this PR —
`Coordinator.doCommit()` handing the live map to `commitToTable` on the
executor looked worth a second glance for snapshot consistency, but the diff
doesn't touch it so I'll leave that as a question.)
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/ChannelTestBase.java:
##########
@@ -132,6 +133,7 @@ public void after() throws IOException {
}
protected void initConsumer() {
+ consumer.subscribe(ImmutableList.of(CTL_TOPIC_NAME));
Review Comment:
Adding `subscribe()` here plus reordering `initConsumer()` before
`worker.start()` in `testSave` sets up a race on a non-thread-safe
`MockConsumer`. The test's `initConsumer()` can run concurrently with the
background thread's `initializeConsumer()` (which also subscribes) on the same
mock — it passes on the timing assumption that setup finishes first, which
isn't guaranteed under CPU contention. Worth gating the background thread's
start on a latch, or doing all mock setup before the thread can touch the
consumer.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -111,6 +119,270 @@ public void testSave() {
assertThat(dataComplete.commitId()).isEqualTo(commitId);
assertThat(dataComplete.assignments()).hasSize(1);
assertThat(dataComplete.assignments().get(0).offset()).isEqualTo(1L);
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testBackgroundPollingBuffersEvents() {
+ when(config.catalogName()).thenReturn("catalog");
+ when(config.controlPollIntervalMs()).thenReturn(50);
+
+ try (MockedStatic<KafkaUtils> mockKafkaUtils =
mockStatic(KafkaUtils.class)) {
+ ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ mockKafkaUtils
+ .when(() -> KafkaUtils.consumerGroupMetadata(any()))
+ .thenReturn(consumerGroupMetadata);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ TopicPartition topicPartition = new TopicPartition(SRC_TOPIC_NAME, 0);
+ when(context.assignment()).thenReturn(ImmutableSet.of(topicPartition));
+
+ SinkWriter sinkWriter = mock(SinkWriter.class);
+ when(sinkWriter.completeWrite())
+ .thenReturn(new SinkWriterResult(ImmutableList.of(),
ImmutableMap.of()));
+
+ initConsumer();
+
+ Worker worker = new Worker(config, clientFactory, sinkWriter, context);
+ worker.start();
+
+ // Add multiple events to consumer
+ UUID commitId1 = UUID.randomUUID();
+ Event event1 = new Event(config.connectGroupId(), new
StartCommit(commitId1));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(event1)));
+
+ UUID commitId2 = UUID.randomUUID();
+ Event event2 = new Event(config.connectGroupId(), new
StartCommit(commitId2));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(event2)));
+
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(5))
+ .pollInterval(Duration.ofMillis(10))
+ .until(() -> worker.pendingEventCount() >= 2);
+
+ // Process should handle both buffered events
+ worker.process();
+
+ // Should have 2 DATA_COMPLETE events (one per commit)
+ assertThat(producer.history().size()).isGreaterThanOrEqualTo(2);
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerIgnoresNonRelevantEvents() {
+ when(config.catalogName()).thenReturn("catalog");
+
+ try (MockedStatic<KafkaUtils> mockKafkaUtils =
mockStatic(KafkaUtils.class)) {
+ ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ mockKafkaUtils
+ .when(() -> KafkaUtils.consumerGroupMetadata(any()))
+ .thenReturn(consumerGroupMetadata);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ when(context.assignment()).thenReturn(ImmutableSet.of());
+
+ SinkWriter sinkWriter = mock(SinkWriter.class);
+
+ initConsumer();
+
+ Worker worker = new Worker(config, clientFactory, sinkWriter, context);
+ worker.start();
+
+ // Add events with different group IDs (should be ignored by Channel's
group filter)
+ UUID commitId = UUID.randomUUID();
+ Event event = new Event("different-group-id", new StartCommit(commitId));
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
AvroUtil.encode(event)));
+
+ // Also add a non-START_COMMIT event with correct group (should be
ignored by Worker)
+ Event commitComplete =
+ new Event(config.connectGroupId(), new CommitComplete(commitId,
EventTestUtil.now()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
AvroUtil.encode(commitComplete)));
+
+ // Wait a bit for the background thread to process the records
+ Awaitility.await()
+ .pollDelay(Duration.ofMillis(200))
Review Comment:
This is a disguised `Thread.sleep(200)`, and I don't think the test proves
what it wants.
`pollDelay(200ms).until(() -> true)` returns after a fixed 200ms with no
real condition, so it's flaky under CI load. And `pendingEventCount() == 0` is
trivially true the instant the worker starts, because the filtered events never
enter the queue — so the assertion can't distinguish "the background thread
consumed and correctly ignored them" from "the background thread hasn't gotten
to them yet." To actually exercise the ignore path you need an observable
signal: a `@VisibleForTesting` consumed-record counter to await on, or waiting
until the consumer position advances past offset 2, then asserting the queue
stayed empty.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -58,19 +79,89 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
- }
- void process() {
- consumeAvailable(Duration.ZERO);
+ this.taskId = config.connectorName() + "-" + config.taskId();
+ this.controlEventQueue = new ConcurrentLinkedQueue<>();
+ this.running = new AtomicBoolean(false);
+ this.pollInterval = Duration.ofMillis(config.controlPollIntervalMs());
+ this.pollingExecutor =
+ Executors.newSingleThreadExecutor(
+ r -> {
+ Thread thread = new Thread(r, "worker-control-poller-" + taskId);
+ thread.setDaemon(true);
+ return thread;
+ });
}
@Override
- protected boolean receive(Envelope envelope) {
- Event event = envelope.event();
- if (event.payload().type() != PayloadType.START_COMMIT) {
- return false;
+ void start() {
+ // Do NOT call super.start() — all consumer access must happen on the
background thread
+ // to satisfy KafkaConsumer's single-thread requirement. The background
thread calls
+ // initializeConsumer() as its first action.
+ running.set(true);
+
+ try {
+ pollingExecutor.execute(this::backgroundPoll);
+ } catch (Exception ex) {
+ LOG.error("Worker {} failed to execute the task.", taskId, ex);
+ throw new ConnectException(
+ String.format("Worker %s failed to execute the poll task", taskId));
+ }
+ LOG.info(
+ "Worker {} started with async control event processing (poll interval:
{}ms)",
+ taskId,
+ pollInterval.toMillis());
+ }
+
+ /**
+ * Background polling task that subscribes to the control topic and
continuously polls for events,
+ * buffering them for processing by the main thread. All KafkaConsumer
access is confined to this
+ * thread.
+ */
+ private void backgroundPoll() {
+ LOG.info("Background control topic polling thread started on {}", taskId);
+ try {
+ // Initialize consumer on this thread — KafkaConsumer is NOT thread-safe,
+ // so subscribe + all poll calls must happen on the same thread.
+ initializeConsumer();
+
+ while (running.get() && !Thread.currentThread().isInterrupted()) {
+ try {
+ consumeAvailable(pollInterval);
+ } catch (WakeupException e) {
+ // Expected during shutdown — wakeupConsumer() was called
+ LOG.debug("Consumer wakeup received during shutdown for {}", taskId,
e);
+ break;
+ }
+ }
+ } catch (Exception e) {
+ if (running.compareAndSet(true, false)) {
+ LOG.error("Worker {} failed while polling control events", taskId, e);
+ errorRef.compareAndSet(null, e);
+ }
+ } finally {
+ LOG.info("Background control topic polling thread stopped on {}",
taskId);
}
+ }
+
+ /**
+ * Process all available events from the queue (non-blocking). This is
called from the main put()
+ * path. Throws ConnectException if the background thread encountered an
error — the caller is
+ * responsible for stopping the worker.
+ */
+ void process() {
+ Exception ex = errorRef.getAndSet(null);
Review Comment:
`errorRef.getAndSet(null)` clears the error destructively on read, which I
think is a subtle trap.
Today the flow works because `CommitterImpl.processControlEvents()` rethrows
and stops the worker. But the moment a caller reads the error and *doesn't*
stop the worker — a future catch block, or a second `process()` before the
framework tears down — the next call sees `null` and cheerfully keeps draining
the queue against a background thread that's already dead and will never refill
it. The sink looks alive while writing data that never commits.
I'd use a non-destructive `errorRef.get()` here and only clear it in
`stop()`/`terminateBackGroundPolling()`, or keep a separate `volatile boolean
backgroundFailed` distinct from the exception. If one-shot delivery really is
the intent, worth documenting that and marking the class single-caller. wdyt?
--
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]
