You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "harangozop (via GitHub)" <gi...@apache.org> on 2025/08/21 06:43:20 UTC
[PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
harangozop opened a new pull request, #24654:
URL: https://github.com/apache/pulsar/pull/24654
<!--
### Contribution Checklist
- PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://pulsar.apache.org/contribute/develop-semantic-title/)*.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
<!-- Either this PR fixes an issue, -->
Fixes #24646
### Motivation
<!-- Explain here the context, and why you're making that change. What is the problem you're trying to solve. -->
When using Kafka Connect source connectors with transform filters that drop records (e.g., Debezium SMT Filter or Kafka Connect Filter + Predicate), the Pulsar Kafka Connect source can hang if a polled batch contains only records that are filtered out.
In this scenario, no records are emitted and the source waits indefinitely for an offset flush that never occurs.
This change ensures that batches are finalized even when all records are filtered: when the last record in a batch is processed and none were emitted/acked, we still trigger and complete the offset flush (treating “nothing to flush” as success). This prevents read() from blocking and allows the source to continue polling subsequent data.
### Modifications
<!-- Describe the modifications you've done. -->
### Verifying this change
- [x] Make sure that the change passes the CI checks.
**This change added tests and can be verified as follows:**
*(example:)*
- *Added integration test that expects a flush when a filter is dropping every message in a given batch.*
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: https://github.com/harangozop/pulsar/pull/2
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "harangozop (via GitHub)" <gi...@apache.org>.
harangozop commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2340784452
##########
pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java:
##########
@@ -96,6 +103,182 @@ public void testOpenAndReadConnectorConfig() throws Exception {
testOpenAndReadTask(config);
}
+
+ @Test(timeOut = 30000)
+ public void testFlushWhenAllMessagesFilteredWithoutBlocking() throws Exception {
+
+ Map<String, Object> config = getConfig();
+ config.put(TaskConfig.TASK_CLASS_CONFIG, "org.apache.kafka.connect.file.FileStreamSourceTask");
+
+ config.put("transforms", "Filter");
+ config.put("transforms.Filter.type", "org.apache.kafka.connect.transforms.Filter");
+ config.put("transforms.Filter.predicate", "DropMeTopic");
+
+ config.put("predicates", "DropMeTopic");
+ config.put("predicates.DropMeTopic.type", "org.apache.kafka.connect.transforms.predicates.TopicNameMatches");
+ config.put("predicates.DropMeTopic.pattern", ".*my-property/my-ns/kafka-connect-source.*");
+
+ kafkaConnectSource = new KafkaConnectSource();
+ kafkaConnectSource.open(config, context);
+
+ OffsetStorageWriter original = kafkaConnectSource.getOffsetWriter();
+ OffsetStorageWriter spyWriter = org.mockito.Mockito.spy(original);
+ java.lang.reflect.Field f = AbstractKafkaConnectSource.class.getDeclaredField("offsetWriter");
+ f.setAccessible(true);
+ f.set(kafkaConnectSource, spyWriter);
Review Comment:
Good catch, I've introduced a new `wrapOffsetWriterWithSpy` helper method to setup the spy.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "harangozop (via GitHub)" <gi...@apache.org>.
harangozop commented on PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#issuecomment-3280587968
Thanks @lhotari I've added the helper method to the tests!
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2325145601
##########
pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java:
##########
@@ -96,6 +103,182 @@ public void testOpenAndReadConnectorConfig() throws Exception {
testOpenAndReadTask(config);
}
+
+ @Test(timeOut = 30000)
+ public void testFlushWhenAllMessagesFilteredWithoutBlocking() throws Exception {
+
+ Map<String, Object> config = getConfig();
+ config.put(TaskConfig.TASK_CLASS_CONFIG, "org.apache.kafka.connect.file.FileStreamSourceTask");
+
+ config.put("transforms", "Filter");
+ config.put("transforms.Filter.type", "org.apache.kafka.connect.transforms.Filter");
+ config.put("transforms.Filter.predicate", "DropMeTopic");
+
+ config.put("predicates", "DropMeTopic");
+ config.put("predicates.DropMeTopic.type", "org.apache.kafka.connect.transforms.predicates.TopicNameMatches");
+ config.put("predicates.DropMeTopic.pattern", ".*my-property/my-ns/kafka-connect-source.*");
+
+ kafkaConnectSource = new KafkaConnectSource();
+ kafkaConnectSource.open(config, context);
+
+ OffsetStorageWriter original = kafkaConnectSource.getOffsetWriter();
+ OffsetStorageWriter spyWriter = org.mockito.Mockito.spy(original);
+ java.lang.reflect.Field f = AbstractKafkaConnectSource.class.getDeclaredField("offsetWriter");
+ f.setAccessible(true);
+ f.set(kafkaConnectSource, spyWriter);
Review Comment:
There's also a valid point about the repetition. It could be addressed by extracting a method `wrapOffsetWriterWithSpy` etc.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "harangozop (via GitHub)" <gi...@apache.org>.
harangozop commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2292908515
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java:
##########
@@ -76,6 +76,7 @@ public abstract class AbstractKafkaConnectSource<T> implements Source<T> {
public OffsetStorageWriter offsetWriter;
// number of outstandingRecords that have been polled but not been acked
private final AtomicInteger outstandingRecords = new AtomicInteger(0);
+ private final AtomicBoolean flushing = new AtomicBoolean(false);
Review Comment:
I can refactor the flushFuture into an AtomicReference if that seems suitable.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2324263017
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java:
##########
@@ -162,38 +165,103 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
sourceTask.initialize(sourceTaskContext);
sourceTask.start(taskConfig);
}
+ private void onOffsetsFlushed(Throwable error, CompletableFuture<Void> snapshotFlushFuture) {
+ if (error != null) {
+ log.error("Failed to flush offsets to storage: ", error);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("No Offsets Added Error", error));
+ return;
+ }
+ try {
+ sourceTask.commit();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished flushing offsets to storage");
+ }
+ snapshotFlushFuture.complete(null);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ log.warn("Flush interrupted, cancelling", ie);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", ie));
+ } catch (Throwable t) {
+ log.warn("Flush failed, cancelling", t);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", t));
+ }
+ }
+
+ private void triggerOffsetsFlushIfNeeded() {
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
+ // Only flush when we have a batch in flight, nothing outstanding, and a pending future
+ if (snapshotFlushFuture == null || snapshotFlushFuture.isDone() || outstandingRecords.get() != 0) {
+ return;
+ }
+ if (!flushing.compareAndSet(false, true)) {
+ return; // someone else is flushing
+ }
+ try {
+ if (offsetWriter.beginFlush()) {
+ offsetWriter.doFlush((error, ignored) -> {
+ try {
+ onOffsetsFlushed(error, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ });
+ } else {
+ try {
+ onOffsetsFlushed(null, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ } catch (ConnectException alreadyFlushing) {
+ // Another thread initiated the flush; let their callback complete the future.
+ // Keep 'flushing' = true until read() finalizes the batch.
+ } catch (Exception t) {
+ try {
+ onOffsetsFlushed(t, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ }
@Override
public synchronized Record<T> read() throws Exception {
while (true) {
if (currentBatch == null) {
- flushFuture = new CompletableFuture<>();
List<SourceRecord> recordList = sourceTask.poll();
if (recordList == null || recordList.isEmpty()) {
continue;
}
outstandingRecords.addAndGet(recordList.size());
currentBatch = recordList.iterator();
+
+ final CompletableFuture<Void> newFuture = new CompletableFuture<>();
+ flushFutureRef.set(newFuture);
}
if (currentBatch.hasNext()) {
AbstractKafkaSourceRecord<T> processRecord = processSourceRecord(currentBatch.next());
if (processRecord == null || processRecord.isEmpty()) {
outstandingRecords.decrementAndGet();
- continue;
+ triggerOffsetsFlushIfNeeded();
} else {
return processRecord;
}
} else {
- // there is no records any more, then waiting for the batch to complete writing
- // to sink and the offsets are committed as well, then do next round read.
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
try {
- flushFuture.get();
+ if (snapshotFlushFuture != null) {
+ snapshotFlushFuture.get();
+ }
} catch (ExecutionException ex) {
- // log the error, continue execution
log.error("execution exception while get flushFuture", ex);
throw new Exception("Flush failed", ex.getCause());
} finally {
- flushFuture = null;
+ flushing.set(false);
+ // Clear only if this is still the current batch future
+ flushFutureRef.compareAndSet(snapshotFlushFuture, null);
Review Comment:
does this require `if (flushFutureRef.compareAndSet(snapshotFlushFuture, null)) { flushing.set(false); }` ?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2325135035
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java:
##########
@@ -162,38 +165,103 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
sourceTask.initialize(sourceTaskContext);
sourceTask.start(taskConfig);
}
+ private void onOffsetsFlushed(Throwable error, CompletableFuture<Void> snapshotFlushFuture) {
+ if (error != null) {
+ log.error("Failed to flush offsets to storage: ", error);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("No Offsets Added Error", error));
+ return;
+ }
+ try {
+ sourceTask.commit();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished flushing offsets to storage");
+ }
+ snapshotFlushFuture.complete(null);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ log.warn("Flush interrupted, cancelling", ie);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", ie));
+ } catch (Throwable t) {
+ log.warn("Flush failed, cancelling", t);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", t));
+ }
+ }
+
+ private void triggerOffsetsFlushIfNeeded() {
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
+ // Only flush when we have a batch in flight, nothing outstanding, and a pending future
+ if (snapshotFlushFuture == null || snapshotFlushFuture.isDone() || outstandingRecords.get() != 0) {
+ return;
+ }
+ if (!flushing.compareAndSet(false, true)) {
+ return; // someone else is flushing
+ }
+ try {
+ if (offsetWriter.beginFlush()) {
+ offsetWriter.doFlush((error, ignored) -> {
+ try {
+ onOffsetsFlushed(error, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ });
+ } else {
+ try {
+ onOffsetsFlushed(null, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ } catch (ConnectException alreadyFlushing) {
+ // Another thread initiated the flush; let their callback complete the future.
+ // Keep 'flushing' = true until read() finalizes the batch.
+ } catch (Exception t) {
+ try {
+ onOffsetsFlushed(t, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ }
@Override
public synchronized Record<T> read() throws Exception {
while (true) {
if (currentBatch == null) {
- flushFuture = new CompletableFuture<>();
List<SourceRecord> recordList = sourceTask.poll();
if (recordList == null || recordList.isEmpty()) {
continue;
}
outstandingRecords.addAndGet(recordList.size());
currentBatch = recordList.iterator();
+
+ final CompletableFuture<Void> newFuture = new CompletableFuture<>();
+ flushFutureRef.set(newFuture);
}
if (currentBatch.hasNext()) {
AbstractKafkaSourceRecord<T> processRecord = processSourceRecord(currentBatch.next());
if (processRecord == null || processRecord.isEmpty()) {
outstandingRecords.decrementAndGet();
- continue;
+ triggerOffsetsFlushIfNeeded();
Review Comment:
@harangozop Please check this Copilot review comment.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #24654:
URL: https://github.com/apache/pulsar/pull/24654
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#issuecomment-3209749309
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24654?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `73.33333%` with `12 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 74.26%. Comparing base ([`34f8657`](https://app.codecov.io/gh/apache/pulsar/commit/34f86576758715f881d755747a74929c189df5ce?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`1b9b11a`](https://app.codecov.io/gh/apache/pulsar/commit/1b9b11a8d50a08e6e4c5d0969c602621ab0f3eb7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 6 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24654?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...r/io/kafka/connect/AbstractKafkaConnectSource.java](https://app.codecov.io/gh/apache/pulsar/pull/24654?src=pr&el=tree&filepath=pulsar-io%2Fkafka-connect-adaptor%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fio%2Fkafka%2Fconnect%2FAbstractKafkaConnectSource.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWlvL2thZmthLWNvbm5lY3QtYWRhcHRvci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2lvL2thZmthL2Nvbm5lY3QvQWJzdHJhY3RLYWZrYUNvbm5lY3RTb3VyY2UuamF2YQ==) | 73.33% | [9 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24654?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24654?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24654 +/- ##
============================================
- Coverage 74.32% 74.26% -0.07%
- Complexity 33180 33186 +6
============================================
Files 1882 1884 +2
Lines 146855 146927 +72
Branches 16867 16873 +6
============================================
- Hits 109152 109116 -36
- Misses 29038 29146 +108
Partials 8665 8665
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24654/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24654/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.64% <ø> (-0.14%)` | :arrow_down: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24654/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.25% <ø> (-0.19%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24654/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.76% <73.33%> (-0.05%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24654?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...r/io/kafka/connect/AbstractKafkaConnectSource.java](https://app.codecov.io/gh/apache/pulsar/pull/24654?src=pr&el=tree&filepath=pulsar-io%2Fkafka-connect-adaptor%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fio%2Fkafka%2Fconnect%2FAbstractKafkaConnectSource.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWlvL2thZmthLWNvbm5lY3QtYWRhcHRvci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2lvL2thZmthL2Nvbm5lY3QvQWJzdHJhY3RLYWZrYUNvbm5lY3RTb3VyY2UuamF2YQ==) | `76.51% <73.33%> (+12.80%)` | :arrow_up: |
... and [98 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24654/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2290572664
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java:
##########
@@ -179,20 +245,21 @@ public synchronized Record<T> read() throws Exception {
AbstractKafkaSourceRecord<T> processRecord = processSourceRecord(currentBatch.next());
if (processRecord == null || processRecord.isEmpty()) {
outstandingRecords.decrementAndGet();
- continue;
+ // If the entire batch is filtered, flush offsets now so it won't block later
+ triggerOffsetsFlushIfNeeded();
} else {
return processRecord;
}
} else {
- // there is no records any more, then waiting for the batch to complete writing
- // to sink and the offsets are committed as well, then do next round read.
+ // No more records in this batch: wait for offsets to be committed before next batch
+ triggerOffsetsFlushIfNeeded();
Review Comment:
It seems that this changes the previous logic since it would only trigger the flush when "ack" has been called. I guess the existing tests don't cover this. I think it's essential so that records aren't lost if the processing isn't acknowledged by calling `ack`
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "harangozop (via GitHub)" <gi...@apache.org>.
harangozop commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2292976598
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java:
##########
@@ -179,20 +245,21 @@ public synchronized Record<T> read() throws Exception {
AbstractKafkaSourceRecord<T> processRecord = processSourceRecord(currentBatch.next());
if (processRecord == null || processRecord.isEmpty()) {
outstandingRecords.decrementAndGet();
- continue;
+ // If the entire batch is filtered, flush offsets now so it won't block later
+ triggerOffsetsFlushIfNeeded();
} else {
return processRecord;
}
} else {
- // there is no records any more, then waiting for the batch to complete writing
- // to sink and the offsets are committed as well, then do next round read.
+ // No more records in this batch: wait for offsets to be committed before next batch
+ triggerOffsetsFlushIfNeeded();
Review Comment:
For most of the cases calling `triggerOffsetsFlushIfNeeded` here is useless, won't do anything, since the flush is only executed when outstanding records reaches zero, no matter how many times you call this function. Alone itself won't decrease the outstanding records.
That can happen on two paths:
- when a record is filtered (no emit)
- and when an emitted record is acked
I'll remove the invocation here, and add more tests to cover:
- partially acked -> shouldn't flush
- all acked -> should flush
- no ack -> shouldn't flush
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "harangozop (via GitHub)" <gi...@apache.org>.
harangozop commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2336112120
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java:
##########
@@ -162,38 +165,103 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
sourceTask.initialize(sourceTaskContext);
sourceTask.start(taskConfig);
}
+ private void onOffsetsFlushed(Throwable error, CompletableFuture<Void> snapshotFlushFuture) {
+ if (error != null) {
+ log.error("Failed to flush offsets to storage: ", error);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("No Offsets Added Error", error));
+ return;
+ }
+ try {
+ sourceTask.commit();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished flushing offsets to storage");
+ }
+ snapshotFlushFuture.complete(null);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ log.warn("Flush interrupted, cancelling", ie);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", ie));
+ } catch (Throwable t) {
+ log.warn("Flush failed, cancelling", t);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", t));
+ }
+ }
+
+ private void triggerOffsetsFlushIfNeeded() {
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
+ // Only flush when we have a batch in flight, nothing outstanding, and a pending future
+ if (snapshotFlushFuture == null || snapshotFlushFuture.isDone() || outstandingRecords.get() != 0) {
+ return;
+ }
+ if (!flushing.compareAndSet(false, true)) {
+ return; // someone else is flushing
+ }
+ try {
+ if (offsetWriter.beginFlush()) {
+ offsetWriter.doFlush((error, ignored) -> {
+ try {
+ onOffsetsFlushed(error, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ });
+ } else {
+ try {
+ onOffsetsFlushed(null, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ } catch (ConnectException alreadyFlushing) {
+ // Another thread initiated the flush; let their callback complete the future.
+ // Keep 'flushing' = true until read() finalizes the batch.
+ } catch (Exception t) {
+ try {
+ onOffsetsFlushed(t, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ }
@Override
public synchronized Record<T> read() throws Exception {
while (true) {
if (currentBatch == null) {
- flushFuture = new CompletableFuture<>();
List<SourceRecord> recordList = sourceTask.poll();
if (recordList == null || recordList.isEmpty()) {
continue;
}
outstandingRecords.addAndGet(recordList.size());
currentBatch = recordList.iterator();
+
+ final CompletableFuture<Void> newFuture = new CompletableFuture<>();
+ flushFutureRef.set(newFuture);
}
if (currentBatch.hasNext()) {
AbstractKafkaSourceRecord<T> processRecord = processSourceRecord(currentBatch.next());
if (processRecord == null || processRecord.isEmpty()) {
outstandingRecords.decrementAndGet();
- continue;
+ triggerOffsetsFlushIfNeeded();
Review Comment:
The `continue` here is redundant.
This branch is the last statement in the while body; after the trigger fn call the execution reaches the loop end.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "harangozop (via GitHub)" <gi...@apache.org>.
harangozop commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2336177014
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java:
##########
@@ -162,38 +165,103 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
sourceTask.initialize(sourceTaskContext);
sourceTask.start(taskConfig);
}
+ private void onOffsetsFlushed(Throwable error, CompletableFuture<Void> snapshotFlushFuture) {
+ if (error != null) {
+ log.error("Failed to flush offsets to storage: ", error);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("No Offsets Added Error", error));
+ return;
+ }
+ try {
+ sourceTask.commit();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished flushing offsets to storage");
+ }
+ snapshotFlushFuture.complete(null);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ log.warn("Flush interrupted, cancelling", ie);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", ie));
+ } catch (Throwable t) {
+ log.warn("Flush failed, cancelling", t);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", t));
+ }
+ }
+
+ private void triggerOffsetsFlushIfNeeded() {
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
+ // Only flush when we have a batch in flight, nothing outstanding, and a pending future
+ if (snapshotFlushFuture == null || snapshotFlushFuture.isDone() || outstandingRecords.get() != 0) {
+ return;
+ }
+ if (!flushing.compareAndSet(false, true)) {
+ return; // someone else is flushing
+ }
+ try {
+ if (offsetWriter.beginFlush()) {
+ offsetWriter.doFlush((error, ignored) -> {
+ try {
+ onOffsetsFlushed(error, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ });
+ } else {
+ try {
+ onOffsetsFlushed(null, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ } catch (ConnectException alreadyFlushing) {
+ // Another thread initiated the flush; let their callback complete the future.
+ // Keep 'flushing' = true until read() finalizes the batch.
+ } catch (Exception t) {
+ try {
+ onOffsetsFlushed(t, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ }
@Override
public synchronized Record<T> read() throws Exception {
while (true) {
if (currentBatch == null) {
- flushFuture = new CompletableFuture<>();
List<SourceRecord> recordList = sourceTask.poll();
if (recordList == null || recordList.isEmpty()) {
continue;
}
outstandingRecords.addAndGet(recordList.size());
currentBatch = recordList.iterator();
+
+ final CompletableFuture<Void> newFuture = new CompletableFuture<>();
+ flushFutureRef.set(newFuture);
}
if (currentBatch.hasNext()) {
AbstractKafkaSourceRecord<T> processRecord = processSourceRecord(currentBatch.next());
if (processRecord == null || processRecord.isEmpty()) {
outstandingRecords.decrementAndGet();
- continue;
+ triggerOffsetsFlushIfNeeded();
} else {
return processRecord;
}
} else {
- // there is no records any more, then waiting for the batch to complete writing
- // to sink and the offsets are committed as well, then do next round read.
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
try {
- flushFuture.get();
+ if (snapshotFlushFuture != null) {
+ snapshotFlushFuture.get();
+ }
} catch (ExecutionException ex) {
- // log the error, continue execution
log.error("execution exception while get flushFuture", ex);
throw new Exception("Flush failed", ex.getCause());
} finally {
- flushFuture = null;
+ flushing.set(false);
+ // Clear only if this is still the current batch future
+ flushFutureRef.compareAndSet(snapshotFlushFuture, null);
Review Comment:
No, when calling `flushFutureRef.get()` it already blocks until the flush is ready, so it's safe to set flushing to false unconditionally.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2324268637
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java:
##########
@@ -162,38 +165,103 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
sourceTask.initialize(sourceTaskContext);
sourceTask.start(taskConfig);
}
+ private void onOffsetsFlushed(Throwable error, CompletableFuture<Void> snapshotFlushFuture) {
+ if (error != null) {
+ log.error("Failed to flush offsets to storage: ", error);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("No Offsets Added Error", error));
+ return;
+ }
+ try {
+ sourceTask.commit();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished flushing offsets to storage");
+ }
+ snapshotFlushFuture.complete(null);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ log.warn("Flush interrupted, cancelling", ie);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", ie));
+ } catch (Throwable t) {
+ log.warn("Flush failed, cancelling", t);
+ offsetWriter.cancelFlush();
+ snapshotFlushFuture.completeExceptionally(new Exception("Failed to commit offsets", t));
+ }
+ }
+
+ private void triggerOffsetsFlushIfNeeded() {
+ final CompletableFuture<Void> snapshotFlushFuture = flushFutureRef.get();
+ // Only flush when we have a batch in flight, nothing outstanding, and a pending future
+ if (snapshotFlushFuture == null || snapshotFlushFuture.isDone() || outstandingRecords.get() != 0) {
+ return;
+ }
+ if (!flushing.compareAndSet(false, true)) {
+ return; // someone else is flushing
+ }
+ try {
+ if (offsetWriter.beginFlush()) {
+ offsetWriter.doFlush((error, ignored) -> {
+ try {
+ onOffsetsFlushed(error, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ });
+ } else {
+ try {
+ onOffsetsFlushed(null, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ } catch (ConnectException alreadyFlushing) {
+ // Another thread initiated the flush; let their callback complete the future.
+ // Keep 'flushing' = true until read() finalizes the batch.
+ } catch (Exception t) {
+ try {
+ onOffsetsFlushed(t, snapshotFlushFuture);
+ } finally {
+ flushing.set(false);
+ }
+ }
+ }
@Override
public synchronized Record<T> read() throws Exception {
while (true) {
if (currentBatch == null) {
- flushFuture = new CompletableFuture<>();
List<SourceRecord> recordList = sourceTask.poll();
if (recordList == null || recordList.isEmpty()) {
continue;
}
outstandingRecords.addAndGet(recordList.size());
currentBatch = recordList.iterator();
+
+ final CompletableFuture<Void> newFuture = new CompletableFuture<>();
+ flushFutureRef.set(newFuture);
}
if (currentBatch.hasNext()) {
AbstractKafkaSourceRecord<T> processRecord = processSourceRecord(currentBatch.next());
if (processRecord == null || processRecord.isEmpty()) {
outstandingRecords.decrementAndGet();
- continue;
+ triggerOffsetsFlushIfNeeded();
Review Comment:
The `continue` statement was removed but the logic still needs to skip to the next iteration when a record is filtered out. Add `continue;` after the flush trigger to maintain the original loop behavior.
```suggestion
triggerOffsetsFlushIfNeeded();
continue;
```
##########
pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java:
##########
@@ -96,6 +103,182 @@ public void testOpenAndReadConnectorConfig() throws Exception {
testOpenAndReadTask(config);
}
+
+ @Test(timeOut = 30000)
+ public void testFlushWhenAllMessagesFilteredWithoutBlocking() throws Exception {
+
+ Map<String, Object> config = getConfig();
+ config.put(TaskConfig.TASK_CLASS_CONFIG, "org.apache.kafka.connect.file.FileStreamSourceTask");
+
+ config.put("transforms", "Filter");
+ config.put("transforms.Filter.type", "org.apache.kafka.connect.transforms.Filter");
+ config.put("transforms.Filter.predicate", "DropMeTopic");
+
+ config.put("predicates", "DropMeTopic");
+ config.put("predicates.DropMeTopic.type", "org.apache.kafka.connect.transforms.predicates.TopicNameMatches");
+ config.put("predicates.DropMeTopic.pattern", ".*my-property/my-ns/kafka-connect-source.*");
+
+ kafkaConnectSource = new KafkaConnectSource();
+ kafkaConnectSource.open(config, context);
+
+ OffsetStorageWriter original = kafkaConnectSource.getOffsetWriter();
+ OffsetStorageWriter spyWriter = org.mockito.Mockito.spy(original);
+ java.lang.reflect.Field f = AbstractKafkaConnectSource.class.getDeclaredField("offsetWriter");
+ f.setAccessible(true);
+ f.set(kafkaConnectSource, spyWriter);
+
+ try (OutputStream os = Files.newOutputStream(tempFile.toPath())) {
+ os.write("first\n".getBytes());
+ os.flush();
+ os.write("second\n".getBytes());
+ os.flush();
+ }
+
+ Thread t = new Thread(() -> {
+ try {
+ kafkaConnectSource.read();
+ } catch (Exception ignored) {
+ // Ignore, it just needs the loop running through a batch where every message is filtered out
Review Comment:
Catching and ignoring all exceptions can hide important test failures. Consider catching more specific exceptions or at least logging the exception to aid in debugging test failures.
```suggestion
} catch (Exception e) {
log.error("Exception in read thread", e);
```
##########
pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java:
##########
@@ -96,6 +103,182 @@ public void testOpenAndReadConnectorConfig() throws Exception {
testOpenAndReadTask(config);
}
+
+ @Test(timeOut = 30000)
+ public void testFlushWhenAllMessagesFilteredWithoutBlocking() throws Exception {
+
+ Map<String, Object> config = getConfig();
+ config.put(TaskConfig.TASK_CLASS_CONFIG, "org.apache.kafka.connect.file.FileStreamSourceTask");
+
+ config.put("transforms", "Filter");
+ config.put("transforms.Filter.type", "org.apache.kafka.connect.transforms.Filter");
+ config.put("transforms.Filter.predicate", "DropMeTopic");
+
+ config.put("predicates", "DropMeTopic");
+ config.put("predicates.DropMeTopic.type", "org.apache.kafka.connect.transforms.predicates.TopicNameMatches");
+ config.put("predicates.DropMeTopic.pattern", ".*my-property/my-ns/kafka-connect-source.*");
+
+ kafkaConnectSource = new KafkaConnectSource();
+ kafkaConnectSource.open(config, context);
+
+ OffsetStorageWriter original = kafkaConnectSource.getOffsetWriter();
+ OffsetStorageWriter spyWriter = org.mockito.Mockito.spy(original);
+ java.lang.reflect.Field f = AbstractKafkaConnectSource.class.getDeclaredField("offsetWriter");
+ f.setAccessible(true);
+ f.set(kafkaConnectSource, spyWriter);
Review Comment:
This reflection-based field access pattern is repeated in multiple tests. Extract this spy setup logic into a private helper method to reduce code duplication and improve maintainability.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2325140280
##########
pulsar-io/kafka-connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java:
##########
@@ -96,6 +103,182 @@ public void testOpenAndReadConnectorConfig() throws Exception {
testOpenAndReadTask(config);
}
+
+ @Test(timeOut = 30000)
+ public void testFlushWhenAllMessagesFilteredWithoutBlocking() throws Exception {
+
+ Map<String, Object> config = getConfig();
+ config.put(TaskConfig.TASK_CLASS_CONFIG, "org.apache.kafka.connect.file.FileStreamSourceTask");
+
+ config.put("transforms", "Filter");
+ config.put("transforms.Filter.type", "org.apache.kafka.connect.transforms.Filter");
+ config.put("transforms.Filter.predicate", "DropMeTopic");
+
+ config.put("predicates", "DropMeTopic");
+ config.put("predicates.DropMeTopic.type", "org.apache.kafka.connect.transforms.predicates.TopicNameMatches");
+ config.put("predicates.DropMeTopic.pattern", ".*my-property/my-ns/kafka-connect-source.*");
+
+ kafkaConnectSource = new KafkaConnectSource();
+ kafkaConnectSource.open(config, context);
+
+ OffsetStorageWriter original = kafkaConnectSource.getOffsetWriter();
+ OffsetStorageWriter spyWriter = org.mockito.Mockito.spy(original);
+ java.lang.reflect.Field f = AbstractKafkaConnectSource.class.getDeclaredField("offsetWriter");
+ f.setAccessible(true);
+ f.set(kafkaConnectSource, spyWriter);
Review Comment:
org.apache.commons.lang3.reflect.FieldUtils#writeDeclaredField(java.lang.Object, java.lang.String, java.lang.Object, boolean) could be used.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "peterh-wob (via GitHub)" <gi...@apache.org>.
peterh-wob commented on PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#issuecomment-3257228736
hey @lhotari I've pushed the changes you suggested, mind giving it another look when you get a chance? Thanks!
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Improve Kafka Connect source offset flushing logic [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24654:
URL: https://github.com/apache/pulsar/pull/24654#discussion_r2290559311
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java:
##########
@@ -76,6 +76,7 @@ public abstract class AbstractKafkaConnectSource<T> implements Source<T> {
public OffsetStorageWriter offsetWriter;
// number of outstandingRecords that have been polled but not been acked
private final AtomicInteger outstandingRecords = new AtomicInteger(0);
+ private final AtomicBoolean flushing = new AtomicBoolean(false);
Review Comment:
Please also make the `private CompletableFuture<Void> flushFuture;` field a `volatile` field. It's a thread safety problem that exists before this PR.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org