Kafka Connect: Make CommitState.isCommitReady() O(1) - #16453
Conversation
e35cdff to
3bfe7b0
Compare
|
@manuzhang @bryanck It seems you are actively reviewing PRs in Kafka-Connect path, do you mind taking a look at this simple fix PR? |
|
@pvary It seems you are actively reviewing PRs in Kafka-Connect path, do you mind taking a look at this simple fix PR? |
|
@HenryCaiHaiying: That might be some collateral. I don't have deep enough knowledge about Kafka Connect to review specific changes 😢 |
|
Thanks @pvary for taking a look. Who would be a good person to review this PR? |
|
I'll take a deeper look |
| .config("iceberg.control.commit.timeout-ms", Integer.MAX_VALUE) | ||
| .config("iceberg.kafka.auto.offset.reset", "earliest"); | ||
| .config("iceberg.kafka.auto.offset.reset", "earliest") | ||
| .config("offset.flush.interval.ms", 500); |
There was a problem hiding this comment.
Are you sure it helps with the tests stablilization? in this case it will be passed as part of the connector config but I sink it is worker-only config so will it be just ignored? Also the same value is specified in the worker config here https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect-runtime/docker/docker-compose.yml#L108
|
Thanks @HenryCaiHaiying for the PR! |
|
@AnatolyPopov I removed offset.flush.interval.ms setting introduced earlier. I added offset.flush.interval.ms setting in the earlier commit to make the flaky integration tests stable. But I think the root cause of why those integration tests failing is fixed by this PR: #16438, so remove my custom setting. |
AnatolyPopov
left a comment
AnatolyPopov
left a comment
There was a problem hiding this comment.
Thanks @HenryCaiHaiying, looks good now! Thanks for the PR!
Note: GH actions failure for azure bundle seems unrelated.
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
LGTM!
Test looks solid
|
Thanks. Will it get merged soon? |
|
@HenryCaiHaiying i'm fightiing CI now :D |
This PR addresses the issue mentioned in apache#16361 Previously, each DATA_COMPLETE envelope triggered a re-scan of the entire readyBuffer to count received partitions, making per-commit work O(N^2) in the number of buffered messages. Under control-topic backlog this compounded the backlog and made recovery harder. The fix: maintain a running receivedPartitionCount that is incremented in addReady() and reset in endCurrentCommit(). isCommitReady() becomes a constant-time comparison against expectedPartitionCount. The change looks simple, but we also verified the edge cases when the current commit fails and a new start_commit starts, and also the situation where there are 2 Coordinators running (the previous Coordinator didn't terminate and comes back as a zombie coordinator). In both situations, we maintain the same logic before and after the code change. Added 2 new unit tests to cover the behavior before and after the commit, and a situation where a DATA message is generated from a zombie coordinator. Closes apache#16361 Co-Authored-By: Claude <svc-devxp-claude@slack-corp.com>
f88ae8d to
275bed9
Compare
|
@laskoviymishka I did the rebase/merge from latest, but CI is still failing. I think the CVE-Scan workflow failure is from upstream Netty vulnerabilities which is from the recent netty library version bump. |
|
I'll take a look once the build is fixed. |
|
@HenryCaiHaiying can you rebase on master one more time? CI issue seems to be fixed there |
|
@laskoviymishka The CI run is clean now. |
|
i'll wait for @bryanck review before merge |
|
@bryanck Do you have time to take a look at the PR? |
|
@bryanck @laskoviymishka Ping ... |
|
@HenryCaiHaiying i think it's good to merge! Thanks for that! |
|
Thanks. |
5 participants
This PR addresses the issue mentioned in #16361
Previously, each DATA_COMPLETE envelope triggered a re-scan of the entire readyBuffer to count received partitions, making per-commit work O(N^2) in the number of buffered messages. Under control-topic backlog this compounded the backlog and made recovery harder.
The fix: maintain a running receivedPartitionCount that is incremented in addReady() and reset in endCurrentCommit(). isCommitReady() becomes a constant-time comparison against expectedPartitionCount.
The change looks simple but we also verified the edge cases on when the current commit failed and a new start_commit starts, also in the situation when there is 2 Coordinator running (the previous Coordinator didn't terminate and comes back as a zombie coordinator). In both situations, we maintain the logic before and post the code change. Added 2 new unit tests to cover the situation before and after the commit and a situation when there is a DATA message generated from a zombie coordinator.
Closes #16361