-
Notifications
You must be signed in to change notification settings - Fork 3.6k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[improve][broker][PIP-286] Make the TopicCompactionService to support find entry based on publishTime or index #21208
Conversation
2898e90
to
83edfc2
Compare
83edfc2
to
8de7f06
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM
|
||
@Override | ||
public CompletableFuture<Entry> findEntryByEntryIndex(long entryIndex) { | ||
final Predicate<Entry> predicate = entry -> { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How do you handle if the entry index is minus value?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It will return the first entry of compacted data, this is also consistent with the method definition, what do you think?
@@ -106,6 +111,78 @@ public CompletableFuture<Position> getLastCompactedPosition() { | |||
return CompletableFuture.completedFuture(compactedTopic.getCompactionHorizon().orElse(null)); | |||
} | |||
|
|||
@Override | |||
public CompletableFuture<Entry> findEntryByPublishTime(long publishTime) { | |||
final Predicate<Entry> predicate = entry -> { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How do you handle if the publishTime is minus value?
pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java
Show resolved
Hide resolved
pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java
Outdated
Show resolved
Hide resolved
pulsar-broker/src/main/java/org/apache/pulsar/compaction/PulsarTopicCompactionService.java
Outdated
Show resolved
Hide resolved
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM :)
CompletableFuture<Entry> findEntryByPublishTime(long publishTime); | ||
|
||
/** | ||
* Find the first entry that greater or equal to target entryIndex. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It requires the broker entry metadata. So we'd better improve the Javadoc to explain the behavior of broker entry metadata missed.
And please also add a test for this case.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ok. I changed the test to cover this case.
/pulsarbot run-failure-checks |
PIP: #20867
Motivation
#20867
Modifications
Add
findEntryByPublishTime
andfindEntryByEntryIndex
in the TopicCompactionService API.Implement them in the PulsarTopicCompactionService using binary search.
Verifying this change
(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:)
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: