[fix][io] Improve Kafka Connect source offset flushing logic #24654

harangozop · 2025-08-21T06:43:17Z

Motivation

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

Verifying this change

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:

If the box was checked, please highlight the changes

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: harangozop#2

lhotari

LGTM, great work @harangozop! thanks for the contribution

codecov-commenter · 2025-08-21T09:33:06Z

Codecov Report

❌ Patch coverage is 50.94340% with 26 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.16%. Comparing base (34f8657) to head (d9517c7).
⚠️ Report is 61 commits behind head on master.

Files with missing lines	Patch %	Lines
...r/io/kafka/connect/AbstractKafkaConnectSource.java	50.94%	20 Missing and 6 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #24654      +/-   ##
============================================
- Coverage     74.32%   74.16%   -0.17%     
- Complexity    33180    33592     +412     
============================================
  Files          1882     1900      +18     
  Lines        146855   148357    +1502     
  Branches      16867    17202     +335     
============================================
+ Hits         109152   110028     +876     
- Misses        29038    29548     +510     
- Partials       8665     8781     +116

Flag	Coverage Δ
inttests	`26.24% <ø> (-0.55%)`	⬇️
systests	`22.77% <ø> (-0.67%)`	⬇️
unittests	`73.68% <50.94%> (-0.13%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...r/io/kafka/connect/AbstractKafkaConnectSource.java	`69.07% <50.94%> (+5.37%)`	⬆️

... and 221 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java

lhotari

Please check the review comments

…thread safety Added more test cases to cover: - partially acked -> shouldn't flush - all acked -> should flush - no ack -> shouldn't flush

peterh-wob · 2025-09-05T06:32:52Z

hey @lhotari I've pushed the changes you suggested, mind giving it another look when you get a chance? Thanks!

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java

lhotari

Great work @harangozop! I added one review comment.

Copilot

Pull Request Overview

This PR fixes a hanging issue in Kafka Connect source connectors when all records in a polled batch are filtered out by transforms (e.g., Debezium SMT Filter). The fix ensures that offset flushing continues even when no records are emitted, preventing the source from blocking indefinitely.

Refactored offset flushing logic to trigger flushes for filtered-out records
Added comprehensive test coverage for various flush scenarios
Centralized flush handling with atomic operations to prevent race conditions

Reviewed Changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.

File	Description
AbstractKafkaConnectSource.java	Refactored offset flushing with centralized logic and atomic operations
KafkaConnectSourceTest.java	Added comprehensive tests for flush scenarios including filtered records
pom.xml	Added awaitility test dependency for async verification

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java

...connect-adaptor/src/test/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSourceTest.java

lhotari

LGTM, just an opportunity left to reduce duplication in test code

harangozop · 2025-09-11T13:10:54Z

Thanks @lhotari I've added the helper method to the tests!

…r into a helper method for better readability

lhotari

LGTM, 🎯 perfect work @harangozop

(cherry picked from commit a824f04)

…24654) (cherry picked from commit a824f04) (cherry picked from commit 9fc8dfb)

…24654) (cherry picked from commit a824f04) (cherry picked from commit bd2ec48)

…24654) (cherry picked from commit a824f04)

…24654)

harangozop added 2 commits August 21, 2025 08:18

[fix][io] Introduce a test case to simulate hanging source connector

fc6e1f4

[fix][io] Improve Kafka Connect source offset flushing logic

1b9b11a

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Aug 21, 2025

lhotari added this to the 4.1.0 milestone Aug 21, 2025

lhotari approved these changes Aug 21, 2025

View reviewed changes

lhotari added release/4.0.7 release/3.0.14 release/3.3.9 labels Aug 21, 2025

lhotari assigned harangozop Aug 21, 2025

lhotari reviewed Aug 21, 2025

View reviewed changes

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java Show resolved Hide resolved

lhotari reviewed Aug 21, 2025

View reviewed changes

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java Outdated Show resolved Hide resolved

lhotari requested changes Aug 21, 2025

View reviewed changes

[fix][io] Introduce AtomicReference to store flushFuture to increase …

fd18f57

…thread safety Added more test cases to cover: - partially acked -> shouldn't flush - all acked -> should flush - no ack -> shouldn't flush

coderzc modified the milestones: 4.1.0, 4.2.0 Sep 1, 2025

lhotari reviewed Sep 5, 2025

View reviewed changes

...ect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java Show resolved Hide resolved

lhotari requested a review from Copilot September 5, 2025 06:56

lhotari reviewed Sep 5, 2025

View reviewed changes

Copilot AI reviewed Sep 5, 2025

View reviewed changes

lhotari mentioned this pull request Sep 10, 2025

[improve][io] support kafka connect transforms and predicates #24221

Merged

5 tasks

lhotari approved these changes Sep 10, 2025

View reviewed changes

[fix][io] Refactor KafkaConnectSourceTests to wrap OffsetStorageWrite…

d9517c7

…r into a helper method for better readability

harangozop force-pushed the bugfix/24646-kafka-connect-source-flush branch from 9d8facc to d9517c7 Compare September 11, 2025 17:40

lhotari approved these changes Sep 11, 2025

View reviewed changes

lhotari merged commit a824f04 into apache:master Sep 11, 2025
51 checks passed

lhotari added the release/4.1.1 label Sep 11, 2025

lhotari pushed a commit that referenced this pull request Sep 11, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (#24654)

8bf4e78

(cherry picked from commit a824f04)

lhotari added the cherry-picked/branch-4.1 label Sep 11, 2025

lhotari pushed a commit that referenced this pull request Sep 11, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (#24654)

6bd9458

(cherry picked from commit a824f04)

lhotari added the cherry-picked/branch-3.3 label Sep 11, 2025

lhotari pushed a commit that referenced this pull request Sep 11, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (#24654)

9fc8dfb

(cherry picked from commit a824f04)

lhotari added the cherry-picked/branch-4.0 label Sep 11, 2025

lhotari pushed a commit that referenced this pull request Sep 11, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (#24654)

bd2ec48

(cherry picked from commit a824f04)

lhotari added the cherry-picked/branch-3.0 label Sep 11, 2025

manas-ctds pushed a commit to datastax/pulsar that referenced this pull request Sep 12, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (apache#…

befbc10

…24654) (cherry picked from commit a824f04) (cherry picked from commit 9fc8dfb)

srinath-ctds pushed a commit to datastax/pulsar that referenced this pull request Sep 12, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (apache#…

cad8e4d

…24654) (cherry picked from commit a824f04) (cherry picked from commit 9fc8dfb)

ganesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Sep 15, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (apache#…

ba870b6

…24654) (cherry picked from commit a824f04) (cherry picked from commit bd2ec48)

ganesh-ctds pushed a commit to datastax/pulsar that referenced this pull request Sep 15, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (apache#…

257a878

…24654) (cherry picked from commit a824f04) (cherry picked from commit bd2ec48)

srinath-ctds pushed a commit to datastax/pulsar that referenced this pull request Sep 15, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (apache#…

10be3b4

…24654) (cherry picked from commit a824f04) (cherry picked from commit bd2ec48)

nodece pushed a commit to nodece/pulsar that referenced this pull request Sep 16, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (apache#…

931c74c

…24654) (cherry picked from commit a824f04)

KannarFr pushed a commit to CleverCloud/pulsar that referenced this pull request Sep 22, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (apache#…

68abfe2

…24654)

walkinggo pushed a commit to walkinggo/pulsar that referenced this pull request Oct 8, 2025

[fix][io] Improve Kafka Connect source offset flushing logic (apache#…

52c51cb

…24654)

[fix][io] Improve Kafka Connect source offset flushing logic #24654

[fix][io] Improve Kafka Connect source offset flushing logic #24654

Uh oh!

Conversation

harangozop commented Aug 21, 2025

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Matching PR in forked repository

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

codecov-commenter commented Aug 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

peterh-wob commented Sep 5, 2025

Uh oh!

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

harangozop commented Sep 11, 2025

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

codecov-commenter commented Aug 21, 2025 •

edited

Loading