[feat][broker][PIP-195]Implement Filter out all delayed messages and skip them when reading messages from bookies - part7 #19035

coderzc · 2022-12-23T01:28:49Z

PIP: #16763

Motivation

#16763

Modifications

Implement Filter out all delayed messages and skip them when reading messages from bookies.

The logical changes are mainly in ManagedLedgerImpl and OpReadEntry.

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (10MB)
Extended integration test for recovery after broker failure

Documentation

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

Matching PR in forked repository

PR in forked repository:

codelipenghui · 2022-12-26T01:58:13Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java

     */
    void asyncReadEntries(int numberOfEntriesToRead, ReadEntriesCallback callback, Object ctx,
-                          PositionImpl maxPosition);
+                          PositionImpl maxPosition, Predicate<PositionImpl> skipCondition);


It's better to add a new method.
I mean, we'd better not change the existing API directly.
This class is annotated with

@InterfaceAudience.LimitedPrivate @InterfaceStability.Stable

codelipenghui

Great change!
I like the brilliant solution for skipping the entries.

I just left some minor comments.

codelipenghui · 2022-12-29T01:18:25Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java

+     * @param ctx                   opaque context
+     * @param maxPosition           max position can read
+     */
+    void asyncReadEntriesWithSkip(int numberOfEntriesToRead, long maxSizeBytes, ReadEntriesCallback callback,


It's better to add a default implementation. Otherwise, we will break the user's existing implementations.

default implementation?

Yes, I mean just add default

default void asyncReadEntriesWithSkip()

codelipenghui · 2022-12-29T01:59:00Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

+        if (lastPosition == null || entriesCount != 0) {
+            lastPosition = (PositionImpl) returnedEntries.get(entriesCount - 1).getPosition();
+        }


If the returnedEntries is empty and the lastPosition is null, we will get an exception.

It seems to we can only judge entriesCount != 0

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

codelipenghui · 2022-12-29T02:13:00Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

+            for (; entryId <= lastEntry; entryId++) {
+                if (opReadEntry.skipCondition.test(PositionImpl.get(ledger.getId(), entryId))) {
+                    break;
+                }
+                lastValidEntry = entryId;
+            }


Maybe you can try to optimize with only one loop to find the first invalid entry range 😁

mattisonchao · 2022-12-29T03:01:51Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

@@ -57,13 +61,13 @@ public static OpReadEntry create(ManagedCursorImpl cursor, PositionImpl readPosi
            maxPosition = PositionImpl.LATEST;
        }
        op.maxPosition = maxPosition;
+        op.skipCondition = skipCondition;


it looks like we forget to reset it when recycle this obj

You are right.

mattisonchao · 2022-12-29T03:04:04Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

        op.ctx = ctx;
        op.nextReadPosition = PositionImpl.get(op.readPosition);
        return op;
    }

-    @Override
-    public void readEntriesComplete(List<Entry> returnedEntries, Object ctx) {
+    void internalReadEntriesComplete(List<Entry> returnedEntries, Object ctx, PositionImpl lastPosition) {


I'm not sure if it is good to extend this method because we defined a parameter named context.

context already be opReadEntry.ctx passing

mattisonchao

Overall LGTM, please apply @codelipenghui comments.

mattisonchao · 2022-12-29T03:09:12Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

+        if (lastPosition == null || entriesCount != 0) {
+            lastPosition = (PositionImpl) returnedEntries.get(entriesCount - 1).getPosition();
+        }


codecov-commenter · 2022-12-29T09:03:59Z

Codecov Report

Attention: Patch coverage is 36.36364% with 35 lines in your changes missing coverage. Please review.

Project coverage is 45.79%. Comparing base (08591d9) to head (fadff0b).
Report is 2133 commits behind head on master.

Files with missing lines	Patch %	Lines
...che/bookkeeper/mledger/impl/ManagedLedgerImpl.java	0.00%	18 Missing and 1 partial ⚠️
...a/org/apache/bookkeeper/mledger/ManagedCursor.java	0.00%	6 Missing ⚠️
...che/bookkeeper/mledger/impl/ManagedCursorImpl.java	44.44%	5 Missing ⚠️
...sistent/PersistentDispatcherMultipleConsumers.java	72.72%	2 Missing and 1 partial ⚠️
...rg/apache/bookkeeper/mledger/impl/OpReadEntry.java	80.00%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19035      +/-   ##
============================================
- Coverage     49.85%   45.79%   -4.07%     
- Complexity     8658    11061    +2403     
============================================
  Files           500      772     +272     
  Lines         54930    74360   +19430     
  Branches       5867     8009    +2142     
============================================
+ Hits          27386    34053    +6667     
- Misses        24464    36518   +12054     
- Partials       3080     3789     +709

Flag	Coverage Δ
unittests	`45.79% <36.36%> (-4.07%)`	⬇️

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

Files with missing lines	Coverage Δ
...ava/org/apache/bookkeeper/mledger/impl/OpScan.java	`0.00% <ø> (ø)`
...rg/apache/bookkeeper/mledger/impl/OpReadEntry.java	`65.97% <80.00%> (+0.42%)`	⬆️
...sistent/PersistentDispatcherMultipleConsumers.java	`59.31% <72.72%> (+0.98%)`	⬆️
...che/bookkeeper/mledger/impl/ManagedCursorImpl.java	`36.70% <44.44%> (+0.02%)`	⬆️
...a/org/apache/bookkeeper/mledger/ManagedCursor.java	`46.15% <0.00%> (-39.57%)`	⬇️
...che/bookkeeper/mledger/impl/ManagedLedgerImpl.java	`53.78% <0.00%> (+0.01%)`	⬆️

... and 327 files with indirect coverage changes

codelipenghui · 2022-12-30T02:01:24Z

/pulsarbot run-failure-checks

codelipenghui · 2022-12-30T02:53:30Z

@coderzc

Error:  Tests run: 556, Failures: 1, Errors: 0, Skipped: 520, Time elapsed: 46.303 s <<< FAILURE! - in org.apache.bookkeeper.mledger.impl.ManagedCursorTest
[1957](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1958)
  Error:  testAsyncReadWithMaxSizeByte(org.apache.bookkeeper.mledger.impl.ManagedCursorTest)  Time elapsed: 20.008 s  <<< FAILURE!
[1958](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1959)
  org.testng.internal.thread.ThreadTimeoutException: Method org.apache.bookkeeper.mledger.impl.ManagedCursorTest.testAsyncReadWithMaxSizeByte() didn't finish within the time-out 20000
[1959](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1960)
  	at [email protected]/jdk.internal.misc.Unsafe.park(Native Method)
[1960](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1961)
  	at [email protected]/java.util.concurrent.locks.LockSupport.park(LockSupport.java:211)
[1961](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1962)
  	at [email protected]/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:715)
[1962](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1963)
  	at [email protected]/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1047)
[1963](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1964)
  	at [email protected]/java.util.concurrent.CountDownLatch.await(CountDownLatch.java:230)
[1964](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1965)
  	at app//org.apache.bookkeeper.mledger.impl.ManagedCursorTest.readAndCheck(ManagedCursorTest.java:572)
[1965](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1966)
  	at app//org.apache.bookkeeper.mledger.impl.ManagedCursorTest.testAsyncReadWithMaxSizeByte(ManagedCursorTest.java:549)
[1966](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1967)
  	at [email protected]/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
[1967](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1968)
  	at [email protected]/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
[1968](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1969)
  	at [email protected]/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
[1969](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1970)
  	at [email protected]/java.lang.reflect.Method.invoke(Method.java:568)
[1970](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1971)
  	at app//org.testng.internal.invokers.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:139)
[1971](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1972)
  	at app//org.testng.internal.invokers.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:47)
[1972](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1973)
  	at app//org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:76)
[1973](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1974)
  	at app//org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:11)
[1974](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1975)
  	at [email protected]/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[1975](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1976)
  	at [email protected]/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
[1976](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1977)
  	at [email protected]/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
[1977](https://github.com/apache/pulsar/actions/runs/3799672325/jobs/6471334338#step:10:1978)
  	at [email protected]/java.lang.Thread.run(Thread.java:833)

I noticed this test failed multiple times. Maybe related to the changes of this PR.

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Dec 23, 2022

coderzc marked this pull request as draft December 23, 2022 02:15

coderzc self-assigned this Dec 25, 2022

coderzc added this to the 2.12.0 milestone Dec 25, 2022

coderzc marked this pull request as ready for review December 25, 2022 12:59

codelipenghui reviewed Dec 26, 2022

View reviewed changes

coderzc force-pushed the cursor_read_filter branch 5 times, most recently from 1b042a0 to 3aefb59 Compare December 28, 2022 07:42

coderzc requested a review from codelipenghui December 28, 2022 07:43

coderzc added type/feature The PR added a new feature or issue requested a new feature area/broker labels Dec 28, 2022

codelipenghui reviewed Dec 29, 2022

View reviewed changes

mattisonchao reviewed Dec 29, 2022

View reviewed changes

mattisonchao approved these changes Dec 29, 2022

View reviewed changes

coderzc force-pushed the cursor_read_filter branch 2 times, most recently from 4197fb6 to dc17954 Compare December 29, 2022 09:40

codelipenghui approved these changes Dec 29, 2022

View reviewed changes

coderzc added 3 commits December 30, 2022 11:24

Filter out and skip unnecessary entry

c6132f4

add a new method

e2efb0d

Apply comment

95012e7

coderzc force-pushed the cursor_read_filter branch 2 times, most recently from b49f41f to c57f05a Compare December 30, 2022 10:38

rebase master & fix test

fadff0b

coderzc force-pushed the cursor_read_filter branch from c57f05a to fadff0b Compare December 30, 2022 11:34

codelipenghui merged commit 9ec1d07 into apache:master Jan 3, 2023

coderzc mentioned this pull request Jan 3, 2023

[fix][broker][PIP-195] fix cursor skip read #19124

Merged

5 tasks

coderzc mentioned this pull request Jan 11, 2023

[feat][broker][PIP-195] Implement delayed message index bucket snapshot(merge/delete) - part8 #19138

Merged

5 tasks

dao-jun mentioned this pull request Dec 16, 2023

[improve][ml] Filter out deleted entries before read entries from ledger. #21739

Merged

15 tasks

Technoboy- mentioned this pull request Mar 27, 2024

[fix][broker] Fix OpReadEntry.skipCondition NPE issue #22367

Merged

4 tasks

dao-jun mentioned this pull request Apr 23, 2024

[improve][ml] Optimize read entries with skipCondition #22560

Open

15 tasks

dao-jun mentioned this pull request May 30, 2024

[broker] Revert PR 6404:"Consumer received duplicated deplayed messages upon restart" #22803

Open

15 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[feat][broker][PIP-195]Implement Filter out all delayed messages and skip them when reading messages from bookies - part7 #19035

[feat][broker][PIP-195]Implement Filter out all delayed messages and skip them when reading messages from bookies - part7 #19035

coderzc commented Dec 23, 2022 •

edited

Loading

codelipenghui Dec 26, 2022

coderzc Dec 27, 2022

codelipenghui left a comment

codelipenghui Dec 29, 2022

coderzc Dec 29, 2022

codelipenghui Dec 29, 2022

codelipenghui Dec 29, 2022

mattisonchao Dec 29, 2022

coderzc Dec 29, 2022 •

edited

Loading

codelipenghui Dec 29, 2022

mattisonchao Dec 29, 2022

coderzc Dec 29, 2022

mattisonchao Dec 29, 2022

coderzc Dec 29, 2022

mattisonchao Dec 29, 2022

coderzc Dec 29, 2022

mattisonchao left a comment •

edited

Loading

mattisonchao Dec 29, 2022

codecov-commenter commented Dec 29, 2022 •

edited

Loading

codelipenghui commented Dec 30, 2022

codelipenghui commented Dec 30, 2022

[feat][broker][PIP-195]Implement Filter out all delayed messages and skip them when reading messages from bookies - part7 #19035

[feat][broker][PIP-195]Implement Filter out all delayed messages and skip them when reading messages from bookies - part7 #19035

Conversation

coderzc commented Dec 23, 2022 • edited Loading

Motivation

Modifications

Verifying this change

Documentation

Matching PR in forked repository

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coderzc Dec 29, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mattisonchao left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Dec 29, 2022 • edited Loading

Codecov Report

codelipenghui commented Dec 30, 2022

codelipenghui commented Dec 30, 2022

coderzc commented Dec 23, 2022 •

edited

Loading

coderzc Dec 29, 2022 •

edited

Loading

mattisonchao left a comment •

edited

Loading

codecov-commenter commented Dec 29, 2022 •

edited

Loading