You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "3pacccccc (via GitHub)" <gi...@apache.org> on 2025/11/06 15:29:03 UTC
[PR] [fix][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
3pacccccc opened a new pull request, #24955:
URL: https://github.com/apache/pulsar/pull/24955
Fixes https://github.com/apache/pulsar/issues/24906
### Motivation
the test log is: https://gist.github.com/3pacccccc/7050810fa187f17046d826e7971c174a
From the error logs, we can see the exact failure:
```
java.util.ConcurrentModificationException: null
at java.base/java.util.ArrayList$Itr.checkForComodification(ArrayList.java:1013) ~[?:?]
at java.base/java.util.ArrayList$Itr.next(ArrayList.java:967) ~[?:?]
at org.apache.bookkeeper.client.PulsarMockLedgerHandle.getLength(PulsarMockLedgerHandle.java:235) ~[testmocks-4.2.0-SNAPSHOT.jar:4.17.2]
at org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.getEstimatedEntrySize(RangeEntryCacheImpl.java:503) ~[managed-ledger-4.2.0-SNAPSHOT.jar:4.2.0-SNAPSHOT]
at org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.asyncReadEntriesByPosition(RangeEntryCacheImpl.java:310) ~[managed-ledger-4.2.0-SNAPSHOT.jar:4.2.0-SNAPSHOT]
at org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.asyncReadEntry0(RangeEntryCacheImpl.java:286) ~[managed-ledger-4.2.0-SNAPSHOT.jar:4.2.0-SNAPSHOT]
at org.apache.bookkeeper.mledger.impl.cache.RangeEntryCacheImpl.asyncReadEntry(RangeEntryCacheImpl.java:268) ~[managed-ledger-4.2.0-SNAPSHOT.jar:4.2.0-SNAPSHOT]
at org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.asyncReadEntry(ManagedLedgerImpl.java:2438) ~[managed-ledger-4.2.0-SNAPSHOT.jar:?]
at org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.internalReadFromLedger(ManagedLedgerImpl.java:2408) ~[managed-ledger-4.2.0-SNAPSHOT.jar:?]
at org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.asyncReadEntries(ManagedLedgerImpl.java:2089) ~[managed-ledger-4.2.0-SNAPSHOT.jar:?]
at org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.asyncReadEntriesWithSkip(ManagedCursorImpl.java:918) ~[managed-ledger-4.2.0-SNAPSHOT.jar:?]
at org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.asyncReadEntriesWithSkipOrWait(ManagedCursorImpl.java:1071) ~[managed-ledger-4.2.0-SNAPSHOT.jar:?]
at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.readMoreEntries(PersistentDispatcherMultipleConsumers.java:430) ~[classes/:?]
at org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers.lambda$readMoreEntriesAsync$4(PersistentDispatcherMultipleConsumers.java:324) ~[classes/:?]
```
The root cause analysis shows:
1. **Thread safety issue**: The `entries` list in `PulsarMockLedgerHandle` was implemented as an `ArrayList`, which is not thread-safe
2. **Concurrent access**: Multiple threads were concurrently reading from and potentially modifying the entries list during test execution
3. **Cache invalidation**: When the `ConcurrentModificationException` occurred, it caused all caches to be removed, making all subsequent read operations fail[1]
4. **Complete read failure**: With the cache cleared and the test's interceptor configured to fail all BookKeeper reads (to enforce cache-only behavior), all subsequent read operations failed with `NonRecoverableLedgerException: Should not read from BK since cache should be used`[2]
[1]: https://github.com/apache/pulsar/blob/9737d038485e0c15f251dc334d6963fd0207953e/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java#L274
[2]: https://github.com/apache/pulsar/blob/b6cfecce5f3a1eecbf6f5df81cb835fbbfe35980/pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionBrokerCacheTest.java#L198
### Modifications
Replace `ArrayList` with `CopyOnWriteArrayList` for the `entries` field in `PulsarMockLedgerHandle`
### Verifying this change
- [x] Make sure that the change passes the CI checks.
- [ ] 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/3pacccccc/pulsar/pull/30
--
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][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24955:
URL: https://github.com/apache/pulsar/pull/24955#issuecomment-3499062246
This change causes a test to fail:
```
Error: Tests run: 21, Failures: 1, Errors: 0, Skipped: 6, Time elapsed: 1189 s <<< FAILURE! -- in org.apache.bookkeeper.mledger.offload.jcloud.impl.BlobStoreManagedLedgerOffloaderTest
Error: org.apache.bookkeeper.mledger.offload.jcloud.impl.BlobStoreManagedLedgerOffloaderTest.testReadHandlerState -- Time elapsed: 60.01 s <<< FAILURE!
org.testng.internal.thread.ThreadTimeoutException: Method org.apache.bookkeeper.mledger.offload.jcloud.impl.BlobStoreManagedLedgerOffloaderTest.testReadHandlerState() didn't finish within the time-out 60000
at java.base/java.util.Arrays.copyOf(Arrays.java:3482)
at java.base/java.util.concurrent.CopyOnWriteArrayList.add(CopyOnWriteArrayList.java:465)
at org.apache.bookkeeper.client.PulsarMockLedgerHandle.addEntry(PulsarMockLedgerHandle.java:162)
at org.apache.bookkeeper.mledger.offload.jcloud.impl.BlobStoreManagedLedgerOffloaderBase.buildReadHandle(BlobStoreManagedLedgerOffloaderBase.java:160)
at org.apache.bookkeeper.mledger.offload.jcloud.impl.BlobStoreManagedLedgerOffloaderTest.testReadHandlerState(BlobStoreManagedLedgerOffloaderTest.java:179)
at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
at org.testng.internal.invokers.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:139)
at org.testng.internal.invokers.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:47)
at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:76)
at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:11)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)
```
perhaps it adds too much overhead for that test.
--
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][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #24955:
URL: https://github.com/apache/pulsar/pull/24955#discussion_r2508189672
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java:
##########
@@ -53,7 +54,7 @@
*/
public class PulsarMockLedgerHandle extends LedgerHandle {
- final ArrayList<LedgerEntryImpl> entries = Lists.newArrayList();
+ final List<LedgerEntryImpl> entries = Collections.synchronizedList(new ArrayList<>());
Review Comment:
There are two spaces after the equals sign. This should be a single space for consistency with standard Java formatting conventions.
```suggestion
final List<LedgerEntryImpl> entries = Collections.synchronizedList(new ArrayList<>());
```
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java:
##########
@@ -193,6 +198,7 @@ public void asyncAddEntry(final ByteBuf data, final AddCallback cb, final Object
data.readBytes(storedData);
entries.add(LedgerEntryImpl.create(ledgerId, lastEntry,
storedData.length, Unpooled.wrappedBuffer(storedData)));
+ totalLengthCounter.addAndGet(storedData.length);
Review Comment:
The addition of an entry to the `entries` list and the increment of `totalLengthCounter` are not atomic. This creates a race condition where `getLength()` could return an inconsistent value if called between these two operations. Consider incrementing the counter before adding to the list, or use a synchronized block to ensure atomicity of both operations together.
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java:
##########
@@ -160,6 +164,7 @@ public long addEntry(byte[] data) throws InterruptedException, BKException {
lastEntry = entries.size();
entries.add(LedgerEntryImpl.create(ledgerId, lastEntry, data.length, Unpooled.wrappedBuffer(data)));
+ totalLengthCounter.addAndGet(data.length);
Review Comment:
The addition of an entry to the `entries` list and the increment of `totalLengthCounter` are not atomic. This creates a race condition where `getLength()` could return an inconsistent value if called between these two operations. Consider incrementing the counter before adding to the list, or use a synchronized block to ensure atomicity of both operations together.
```suggestion
synchronized (this) {
lastEntry = entries.size();
entries.add(LedgerEntryImpl.create(ledgerId, lastEntry, data.length, Unpooled.wrappedBuffer(data)));
totalLengthCounter.addAndGet(data.length);
}
```
--
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][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24955:
URL: https://github.com/apache/pulsar/pull/24955#discussion_r2508565003
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java:
##########
@@ -53,7 +54,7 @@
*/
public class PulsarMockLedgerHandle extends LedgerHandle {
- final ArrayList<LedgerEntryImpl> entries = Lists.newArrayList();
+ final List<LedgerEntryImpl> entries = Collections.synchronizedList(new ArrayList<>());
Review Comment:
removed
--
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][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- merged PR #24955:
URL: https://github.com/apache/pulsar/pull/24955
--
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][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24955:
URL: https://github.com/apache/pulsar/pull/24955#issuecomment-3499080828
I think we should simply use `Collections.synchronizedList(new ArrayList<>())` instead, however it's possible that it introduces deadlocks.
Another optimization would be to add a counter to replace the iteration:
https://github.com/apache/pulsar/blob/186b503579f67bc0eaffc99a7e45026e5c3116ab/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java#L232-L240
--
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][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24955:
URL: https://github.com/apache/pulsar/pull/24955#issuecomment-3501356345
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24955?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.21%. Comparing base ([`7c343d0`](https://app.codecov.io/gh/apache/pulsar/commit/7c343d0209a623bd75fbf0265a261fd2e70d20c1?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`8e824a6`](https://app.codecov.io/gh/apache/pulsar/commit/8e824a6fd9cb323feea5d286c1540f7995b6fc08?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/24955?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24955 +/- ##
============================================
- Coverage 74.24% 74.21% -0.03%
+ Complexity 33911 33863 -48
============================================
Files 1913 1913
Lines 149510 149510
Branches 17373 17373
============================================
- Hits 110997 110954 -43
+ Misses 29685 29681 -4
- Partials 8828 8875 +47
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24955/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/24955/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.29% <ø> (+<0.01%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24955/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `22.69% <ø> (-0.03%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24955/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.73% <ø> (-0.06%)` | :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.
[see 87 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24955/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][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24955:
URL: https://github.com/apache/pulsar/pull/24955#discussion_r2508565736
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java:
##########
@@ -160,6 +164,7 @@ public long addEntry(byte[] data) throws InterruptedException, BKException {
lastEntry = entries.size();
entries.add(LedgerEntryImpl.create(ledgerId, lastEntry, data.length, Unpooled.wrappedBuffer(data)));
+ totalLengthCounter.addAndGet(data.length);
Review Comment:
switched order by incrementing the counter before adding to the list
##########
testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java:
##########
@@ -193,6 +198,7 @@ public void asyncAddEntry(final ByteBuf data, final AddCallback cb, final Object
data.readBytes(storedData);
entries.add(LedgerEntryImpl.create(ledgerId, lastEntry,
storedData.length, Unpooled.wrappedBuffer(storedData)));
+ totalLengthCounter.addAndGet(storedData.length);
Review Comment:
switched order by incrementing the counter before adding to the list
--
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][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on PR #24955:
URL: https://github.com/apache/pulsar/pull/24955#issuecomment-3500919013
> I think we should simply use `Collections.synchronizedList(new ArrayList<>())` instead, however it's possible that it introduces deadlocks.
>
> Another optimization would be to add a counter to replace the iteration:
>
> https://github.com/apache/pulsar/blob/186b503579f67bc0eaffc99a7e45026e5c3116ab/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java#L232-L240
@lhotari changed to `Collections.synchronizedList(new ArrayList<>())`, please 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24955:
URL: https://github.com/apache/pulsar/pull/24955#issuecomment-3501143245
> @lhotari changed to `Collections.synchronizedList(new ArrayList<>())`, please take a look
It's necessary to implement https://github.com/apache/pulsar/blob/186b503579f67bc0eaffc99a7e45026e5c3116ab/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java#L232-L240 using a counter. The ConcurrentModificationException won't get fixed with synchronized list. It's not a large change to do so.
--
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][test]fix flaky KeySharedSubscriptionBrokerCacheTest.testReplayQueueReadsGettingCached [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on PR #24955:
URL: https://github.com/apache/pulsar/pull/24955#issuecomment-3501178895
> > @lhotari changed to `Collections.synchronizedList(new ArrayList<>())`, please take a look
>
> It's necessary to implement
>
> https://github.com/apache/pulsar/blob/186b503579f67bc0eaffc99a7e45026e5c3116ab/testmocks/src/main/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java#L232-L240
>
> using a counter. The ConcurrentModificationException won't get fixed with synchronized list. It's not a large change to do so.
@lhotari sure, I'll add a counter for 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org