You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "kumarpritam863 (via GitHub)" <gi...@apache.org> on 2025/12/11 04:31:29 UTC
[PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
kumarpritam863 opened a new pull request, #14816:
URL: https://github.com/apache/iceberg/pull/14816
## Problem Statement
Currently, workers poll the **control topic** inside the main record-processing loop using `poll(Duration.ZERO)`. This tightly-coupled design leads to several critical issues:
1. **Missed START_COMMIT at startup**
Initial partition assignment delays cause workers to frequently miss the first `START_COMMIT` from the coordinator → delayed snapshot creation at destination.
2. **Throughput blocked by control-topic latency**
Any intermittent Kafka latency or outage on the control topic completely blocks file writing → severe throughput degradation.
3. **Unreliable zero-duration polling**
`Duration.ZERO` only works reliably with constant traffic. When no records are buffered on the broker, workers miss `START_COMMIT` cycles (both at start and mid-stream).
4. **Wasteful processing of irrelevant events**
All workers consume the same control topic → every worker unnecessarily processes coordinator events intended for other workers → significant CPU waste and reduced throughput.
## Solution
This PR **completely decouples** control-topic processing from the main record-processing flow by:
- Introducing **one dedicated consumer thread per worker** that exclusively subscribes to the control topic.
- Pre-processing and caching coordinator events (`START_COMMIT`, `COMMITTED`, etc.) in a thread-safe queue.
- Making these events instantly available to the main worker thread without any polling in the hot path.
The main poll loop now only consumes data topics and is no longer blocked or distracted by the control topic.
## Benefits
- Workers continue processing & writing records even during transient control-topic / Kafka issues
- No processing of other workers' coordinator events
- **Zero missed START_COMMIT cycles** (startup or intermediate) → faster & consistent snapshots
- **Significant throughput improvement** (~1.8–2.2× in load tests)
- **End-to-end latency reduced by ~commit-interval × 2**
- Memory impact is negligible → one extra thread per worker is fully justified
## Validation
- Tested with 24–48 parallel workers under production-like load
- Verified no missed commits even with artificial delays on control topic
- Memory footprint remains within previous bounds
- All unit & integration tests pass
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "vitaly-sinev (via GitHub)" <gi...@apache.org>.
vitaly-sinev commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-3736974394
Hi! Just a note that this PR is still relevant and actively awaited.
It’s important for my use case, so I’d really appreciate it if it could stay open and be considered for merging.
Happy to help with testing or feedback if needed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-3832799421
@bryanck I have resolved all the autogenerated reviews. Can you take a look once.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-3902025581
@bryanck fyi.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-4094577501
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-3640057745
@bryanck for review.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#discussion_r2751211719
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -110,14 +202,60 @@ protected boolean receive(Envelope envelope) {
events.add(readyEvent);
send(events, results.sourceOffsets());
+ }
+
+ @Override
+ protected boolean receive(Envelope envelope) {
+ Event event = envelope.event();
+ PayloadType type = event.payload().type();
- return true;
+ // Only buffer control events that need processing
+ if (type == PayloadType.START_COMMIT
+ || type == PayloadType.COMMIT_COMPLETE
+ || type == PayloadType.COMMIT_TO_TABLE) {
+ controlEventQueue.offer(envelope);
+ LOG.info("Worker {} buffered control event: {}", workerIdentifier, type);
+ return true;
+ }
+
+ return false;
}
@Override
void stop() {
- super.stop();
+ LOG.info("Worker {} stopping.", workerIdentifier);
+ terminateBackGroundPolling();
sinkWriter.close();
+ super.stop();
+ LOG.info("Worker {} stopped.", workerIdentifier);
+ }
+
+ private void terminateBackGroundPolling() {
+ running.set(false);
+ pollingExecutor.shutdownNow();
+ try {
+ if (!pollingExecutor.awaitTermination(1, TimeUnit.MINUTES)) {
+ LOG.warn(
+ "Polling thread did not terminate in time on worker {}, forcing shutdown",
+ workerIdentifier);
+ throw new ConnectException(
+ String.format(
+ "Background polling thread of worker %s did not terminated gracefully.",
Review Comment:
Corrected grammar from 'did not terminated' to 'did not terminate'.
```suggestion
"Background polling thread of worker %s did not terminate gracefully.",
```
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -111,6 +116,314 @@ 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() throws InterruptedException {
+ 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)));
+
+ // Wait for background polling to buffer events
+ Thread.sleep(300);
+
+ // 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() throws InterruptedException {
+ 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)
+ 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)));
+
+ Thread.sleep(200);
+ worker.process();
+
+ // Should not produce any events since the group ID doesn't match
+ assertThat(producer.history()).isEmpty();
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerGracefulShutdown() throws InterruptedException {
+ 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
+ worker.stop();
+
+ // Should complete without exceptions
+ assertThat(producer.history()).isEmpty();
+ }
+ }
+
+ @Test
+ public void testWorkerProcessesMultipleEventTypes() throws InterruptedException {
+ 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));
+
+ 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();
+
+ UUID commitId = UUID.randomUUID();
+
+ // Add START_COMMIT event
+ Event startCommit = new Event(config.connectGroupId(), new StartCommit(commitId));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", AvroUtil.encode(startCommit)));
+
+ // Add COMMIT_COMPLETE event
+ Event commitComplete =
+ new Event(config.connectGroupId(), new CommitComplete(commitId, EventTestUtil.now()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key", AvroUtil.encode(commitComplete)));
+
+ // Add COMMIT_TO_TABLE event
+ Event commitToTable =
+ new Event(
+ config.connectGroupId(),
+ new CommitToTable(
+ commitId,
+ TableReference.of("catalog", TableIdentifier.parse(TABLE_NAME)),
+ 1L,
+ EventTestUtil.now()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 3, "key", AvroUtil.encode(commitToTable)));
+
+ Thread.sleep(300);
Review Comment:
Using `Thread.sleep()` in tests creates timing-dependent behavior that can lead to flaky tests. Consider using a polling mechanism with a timeout to verify events have been buffered.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -59,19 +76,96 @@ class Worker extends Channel {
this.config = config;
this.context = context;
this.sinkWriter = sinkWriter;
+
+ this.workerIdentifier = config.connectorName() + "-" + config.taskId();
+ // Initialize async processing components
+ 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-" + config.connectorName() + "-" + config.taskId());
+ thread.setDaemon(true); // Not daemon to ensure graceful shutdown
Review Comment:
The comment states 'Not daemon to ensure graceful shutdown' but the code sets `setDaemon(true)`. The comment should say 'Daemon thread for automatic cleanup' or the code should use `setDaemon(false)` if graceful shutdown is required.
```suggestion
thread.setDaemon(true); // Daemon thread for automatic cleanup
```
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -111,6 +116,314 @@ 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() throws InterruptedException {
+ 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)));
+
+ // Wait for background polling to buffer events
+ Thread.sleep(300);
Review Comment:
Using `Thread.sleep()` in tests creates timing-dependent behavior that can lead to flaky tests. Consider using a polling mechanism with a timeout to verify events have been buffered.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -111,6 +116,314 @@ 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() throws InterruptedException {
+ 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)));
+
+ // Wait for background polling to buffer events
+ Thread.sleep(300);
+
+ // 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() throws InterruptedException {
+ 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)
+ 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)));
+
+ Thread.sleep(200);
+ worker.process();
+
+ // Should not produce any events since the group ID doesn't match
+ assertThat(producer.history()).isEmpty();
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerGracefulShutdown() throws InterruptedException {
+ 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
+ worker.stop();
+
+ // Should complete without exceptions
+ assertThat(producer.history()).isEmpty();
+ }
+ }
+
+ @Test
+ public void testWorkerProcessesMultipleEventTypes() throws InterruptedException {
+ 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));
+
+ 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();
+
+ UUID commitId = UUID.randomUUID();
+
+ // Add START_COMMIT event
+ Event startCommit = new Event(config.connectGroupId(), new StartCommit(commitId));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", AvroUtil.encode(startCommit)));
+
+ // Add COMMIT_COMPLETE event
+ Event commitComplete =
+ new Event(config.connectGroupId(), new CommitComplete(commitId, EventTestUtil.now()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key", AvroUtil.encode(commitComplete)));
+
+ // Add COMMIT_TO_TABLE event
+ Event commitToTable =
+ new Event(
+ config.connectGroupId(),
+ new CommitToTable(
+ commitId,
+ TableReference.of("catalog", TableIdentifier.parse(TABLE_NAME)),
+ 1L,
+ EventTestUtil.now()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 3, "key", AvroUtil.encode(commitToTable)));
+
+ Thread.sleep(300);
+
+ // All events should be buffered
+ worker.process();
+
+ // Should have processed the START_COMMIT event
+ assertThat(producer.history()).isNotEmpty();
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerHandlesEmptyQueue() throws InterruptedException {
+ 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 testWorkerWithCustomPollInterval() throws InterruptedException {
+ when(config.catalogName()).thenReturn("catalog");
+ when(config.controlPollIntervalMs()).thenReturn(1000); // 1 second
+
+ 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();
+
+ UUID commitId = UUID.randomUUID();
+ Event event = new Event(config.connectGroupId(), new StartCommit(commitId));
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", AvroUtil.encode(event)));
+
+ // Wait for longer than poll interval to ensure event is buffered
+ Thread.sleep(1500);
+
+ worker.process();
+
+ assertThat(producer.history()).isNotEmpty();
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerMultipleStartCommits() throws InterruptedException {
+ 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)));
+
+ Thread.sleep(300);
Review Comment:
Using `Thread.sleep()` in tests creates timing-dependent behavior that can lead to flaky tests. Consider using a polling mechanism with a timeout to verify events have been buffered.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -96,6 +99,8 @@ public void testSave() {
byte[] bytes = AvroUtil.encode(commitRequest);
consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", bytes));
+ // Give background thread time to poll and buffer the event
+ Thread.sleep(500);
Review Comment:
Using `Thread.sleep()` in tests creates timing-dependent behavior that can lead to flaky tests. Consider using a polling mechanism with a timeout to verify the event has been buffered, or use test utilities like Awaitility to wait for the expected condition.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -111,6 +116,314 @@ 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() throws InterruptedException {
+ 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)));
+
+ // Wait for background polling to buffer events
+ Thread.sleep(300);
+
+ // 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() throws InterruptedException {
+ 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)
+ 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)));
+
+ Thread.sleep(200);
Review Comment:
Using `Thread.sleep()` in tests creates timing-dependent behavior that can lead to flaky tests. Consider using a polling mechanism with a timeout to verify the queue remains empty.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestWorker.java:
##########
@@ -111,6 +116,314 @@ 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() throws InterruptedException {
+ 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)));
+
+ // Wait for background polling to buffer events
+ Thread.sleep(300);
+
+ // 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() throws InterruptedException {
+ 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)
+ 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)));
+
+ Thread.sleep(200);
+ worker.process();
+
+ // Should not produce any events since the group ID doesn't match
+ assertThat(producer.history()).isEmpty();
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerGracefulShutdown() throws InterruptedException {
+ 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
+ worker.stop();
+
+ // Should complete without exceptions
+ assertThat(producer.history()).isEmpty();
+ }
+ }
+
+ @Test
+ public void testWorkerProcessesMultipleEventTypes() throws InterruptedException {
+ 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));
+
+ 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();
+
+ UUID commitId = UUID.randomUUID();
+
+ // Add START_COMMIT event
+ Event startCommit = new Event(config.connectGroupId(), new StartCommit(commitId));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", AvroUtil.encode(startCommit)));
+
+ // Add COMMIT_COMPLETE event
+ Event commitComplete =
+ new Event(config.connectGroupId(), new CommitComplete(commitId, EventTestUtil.now()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key", AvroUtil.encode(commitComplete)));
+
+ // Add COMMIT_TO_TABLE event
+ Event commitToTable =
+ new Event(
+ config.connectGroupId(),
+ new CommitToTable(
+ commitId,
+ TableReference.of("catalog", TableIdentifier.parse(TABLE_NAME)),
+ 1L,
+ EventTestUtil.now()));
+ consumer.addRecord(
+ new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 3, "key", AvroUtil.encode(commitToTable)));
+
+ Thread.sleep(300);
+
+ // All events should be buffered
+ worker.process();
+
+ // Should have processed the START_COMMIT event
+ assertThat(producer.history()).isNotEmpty();
+
+ worker.stop();
+ }
+ }
+
+ @Test
+ public void testWorkerHandlesEmptyQueue() throws InterruptedException {
+ 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 testWorkerWithCustomPollInterval() throws InterruptedException {
+ when(config.catalogName()).thenReturn("catalog");
+ when(config.controlPollIntervalMs()).thenReturn(1000); // 1 second
+
+ 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();
+
+ UUID commitId = UUID.randomUUID();
+ Event event = new Event(config.connectGroupId(), new StartCommit(commitId));
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key", AvroUtil.encode(event)));
+
+ // Wait for longer than poll interval to ensure event is buffered
+ Thread.sleep(1500);
Review Comment:
Using `Thread.sleep()` in tests creates timing-dependent behavior that can lead to flaky tests. Consider using a polling mechanism with a timeout to verify the event has been buffered.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-3916202975
@bryanck can you review this once. this is kind of important
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-4139199928
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #14816: Decouple control topic processing from record poll loop – improve throughput & reliability
URL: https://github.com/apache/iceberg/pull/14816
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-3736515653
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-3853766087
@danielcweeks @ajantha-bhat can we please review this.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Decouple control topic processing from record poll loop – improve throughput & reliability [iceberg]
Posted by "kumarpritam863 (via GitHub)" <gi...@apache.org>.
kumarpritam863 commented on PR #14816:
URL: https://github.com/apache/iceberg/pull/14816#issuecomment-3738315772
@bryanck take a look.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org