You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "BewareMyPower (via GitHub)" <gi...@apache.org> on 2025/09/10 13:49:08 UTC
[PR] [fix][client] Avoid recycling the same ConcurrentBitSetRecyclable among different threads [pulsar]
BewareMyPower opened a new pull request, #24725:
URL: https://github.com/apache/pulsar/pull/24725
Fixes #24724
### Modifications
Change `pendingIndividualBatchIndexAcks` to a `ConcurrentSkipListMap` and call `pollFirst()` in the loop to ensure concurrent `flush` call won't copy the reference to the `ConcurrentBitSetRecyclable` twice to `newMultiMessageAckCommon` in different threads, which recycles the object.
### Verifying this change
It's hard to write a unit test because this race condition is hard to simulate.
### 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: <!-- ENTER URL HERE -->
### Additional information
This PR is a simple fix on the issue that avoids refactorings. However, it might not be worth allowing `flush` to be called concurrently. It's hard to test the race condition and could send
The advantage is just to avoid coarse grained lock on all methods that access the following fields:
- `pendingIndividualAcks`
- `pendingIndividualBatchIndexAcks`
- `lastCumulativeAck`
However, from my perspective, it's a pre-mature optimization that makes code error-prone. Because acquiring a lock for all `acknowledgeAsync` calls should not be a bottle neck of consumer side, whose time consuming tasks are mainly the business logic that handles messages and `receive` calls that have many lock acquirements as well.
The use of `ConcurrentBitSetRecyclable` is also a pre-mature optimization that results to the bug described in #24724.
As I've shared in https://lists.apache.org/thread/b5r13oz24y935p6o8tfwf578xk35wwpf, sharing a recyclable object among different threads is not a good practice, which introduces more overhead and bypasses the performance benefits of the thread-local stacks. There might be still other potential issues with the use of `ConcurrentBitSetRecyclable`, I might open another PR for the code refactoring.
--
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][client] Avoid recycling the same ConcurrentBitSetRecyclable among different threads [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24725:
URL: https://github.com/apache/pulsar/pull/24725#issuecomment-3280919174
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24725?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 74.14%. Comparing base ([`415c6fa`](https://app.codecov.io/gh/apache/pulsar/commit/415c6fa74e908bad2da1e7f986185c6ef17cb9a1?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`40fb81f`](https://app.codecov.io/gh/apache/pulsar/commit/40fb81f994090782e4c38848ec994feeb47601e6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24725?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24725 +/- ##
============================================
- Coverage 74.27% 74.14% -0.14%
+ Complexity 33569 33117 -452
============================================
Files 1896 1896
Lines 148111 148109 -2
Branches 17164 17163 -1
============================================
- Hits 110009 109814 -195
- Misses 29370 29519 +149
- Partials 8732 8776 +44
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24725/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/24725/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.48% <57.14%> (-0.01%)` | :arrow_down: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24725/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `22.68% <100.00%> (+0.01%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24725/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.65% <100.00%> (-0.15%)` | :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/24725?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...impl/PersistentAcknowledgmentsGroupingTracker.java](https://app.codecov.io/gh/apache/pulsar/pull/24725?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FPersistentAcknowledgmentsGroupingTracker.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1BlcnNpc3RlbnRBY2tub3dsZWRnbWVudHNHcm91cGluZ1RyYWNrZXIuamF2YQ==) | `87.65% <100.00%> (-0.08%)` | :arrow_down: |
... and [83 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24725/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][client] Avoid recycling the same ConcurrentBitSetRecyclable among different threads [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #24725:
URL: https://github.com/apache/pulsar/pull/24725
--
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][client] Avoid recycling the same ConcurrentBitSetRecyclable among different threads [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #24725:
URL: https://github.com/apache/pulsar/pull/24725#discussion_r2336840102
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -484,16 +483,15 @@ private void flushAsync(ClientCnx cnx) {
}
}
- if (!pendingIndividualBatchIndexAcks.isEmpty()) {
- Iterator<Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable>> iterator =
- pendingIndividualBatchIndexAcks.entrySet().iterator();
-
- while (iterator.hasNext()) {
- Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable> entry = iterator.next();
- entriesToAck.add(Triple.of(
- entry.getKey().getLedgerId(), entry.getKey().getEntryId(), entry.getValue()));
- iterator.remove();
+ while (!pendingIndividualBatchIndexAcks.isEmpty()) {
+ Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable> entry =
+ pendingIndividualBatchIndexAcks.pollFirstEntry();
+ if (entry == null) {
+ // The entry has been removed in a different thread
Review Comment:
The check `!pendingIndividualBatchIndexAcks.isEmpty()` in the while condition creates a race condition. Another thread could remove the last entry between the isEmpty() check and pollFirstEntry() call, causing unnecessary loop iterations. Consider using `while (true)` and relying solely on the null check from `pollFirstEntry()`.
```suggestion
while (true) {
Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable> entry =
pendingIndividualBatchIndexAcks.pollFirstEntry();
if (entry == null) {
```
--
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][client] Avoid recycling the same ConcurrentBitSetRecyclable among different threads [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24725:
URL: https://github.com/apache/pulsar/pull/24725#discussion_r2336863661
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java:
##########
@@ -484,16 +483,15 @@ private void flushAsync(ClientCnx cnx) {
}
}
- if (!pendingIndividualBatchIndexAcks.isEmpty()) {
- Iterator<Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable>> iterator =
- pendingIndividualBatchIndexAcks.entrySet().iterator();
-
- while (iterator.hasNext()) {
- Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable> entry = iterator.next();
- entriesToAck.add(Triple.of(
- entry.getKey().getLedgerId(), entry.getKey().getEntryId(), entry.getValue()));
- iterator.remove();
+ while (!pendingIndividualBatchIndexAcks.isEmpty()) {
+ Map.Entry<MessageIdAdv, ConcurrentBitSetRecyclable> entry =
+ pendingIndividualBatchIndexAcks.pollFirstEntry();
+ if (entry == null) {
+ // The entry has been removed in a different thread
Review Comment:
Addressed
--
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