You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "coderzc (via GitHub)" <gi...@apache.org> on 2025/06/23 11:11:02 UTC
[PR] [fix][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
coderzc opened a new pull request, #24441:
URL: https://github.com/apache/pulsar/pull/24441
…
Fixes #xyz
<!-- or this PR is one task of an issue -->
Main Issue: #xyz
<!-- If the PR belongs to a PIP, please add the PIP link here -->
PIP: #xyz
<!-- Details of when a PIP is required and how the PIP process work, please see: https://github.com/apache/pulsar/blob/master/pip/README.md -->
### Motivation
When some ledgers are not offloaded successfully, they will be removed from the metadata store so that seek by timestamp might fail.
```
Error opening ledger for reading at position 4098289:438 - org.apache.bookkeeper.mledger.ManagedLedgerException$LedgerNotExistException: No such ledger exists on Metadata Server
```
skipNonRecoverableData is true, but it does not handle the seek case. The workaround is to see by message id.
### Modifications
When `skipNonRecoverableData` is true, OpFindNewest can auto skip non-recoverable data.
### 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*
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] 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: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#issuecomment-3002754174
@codelipenghui @BewareMyPower @poorbarcode PTAL
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2168198484
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,31 +119,93 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
break;
case searching:
if (condition.test(entry)) {
// mid - last
lastMatchedPosition = position;
- min = mid();
+ min = mid;
} else {
// start - mid
- max = mid() - 1;
+ max = mid - 1;
}
+ this.mid = mid();
if (max <= min) {
callback.findEntryComplete(lastMatchedPosition, OpFindNewest.this.ctx);
return;
}
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
}
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException
+ && ledger.getConfig().isAutoSkipNonRecoverableData()) {
+ log.info("[{}] Ledger {} is not recoverable, skip non-recoverable data, state:{}", ledger.getName(),
+ searchPosition, state);
+ Position nextPosition = null;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ Long nextLedgerId = ledger.getNextValidLedger(searchPosition.getLedgerId());
Review Comment:
@poorbarcode Do you have a chance to take a look again?
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2162024181
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,31 +119,93 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
break;
case searching:
if (condition.test(entry)) {
// mid - last
lastMatchedPosition = position;
- min = mid();
+ min = mid;
} else {
// start - mid
- max = mid() - 1;
+ max = mid - 1;
}
+ this.mid = mid();
if (max <= min) {
callback.findEntryComplete(lastMatchedPosition, OpFindNewest.this.ctx);
return;
}
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
}
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException
+ && ledger.getConfig().isAutoSkipNonRecoverableData()) {
+ log.info("[{}] Ledger {} is not recoverable, skip non-recoverable data, state:{}", ledger.getName(),
+ searchPosition, state);
+ Position nextPosition = null;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ Long nextLedgerId = ledger.getNextValidLedger(searchPosition.getLedgerId());
+ if (nextLedgerId != null) {
+ nextPosition = PositionFactory.create(nextLedgerId, 0);
Review Comment:
Can the next ledger be empty?
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,31 +119,93 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
break;
case searching:
if (condition.test(entry)) {
// mid - last
lastMatchedPosition = position;
- min = mid();
+ min = mid;
} else {
// start - mid
- max = mid() - 1;
+ max = mid - 1;
}
+ this.mid = mid();
if (max <= min) {
callback.findEntryComplete(lastMatchedPosition, OpFindNewest.this.ctx);
return;
}
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
}
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException
+ && ledger.getConfig().isAutoSkipNonRecoverableData()) {
+ log.info("[{}] Ledger {} is not recoverable, skip non-recoverable data, state:{}", ledger.getName(),
+ searchPosition, state);
+ Position nextPosition = null;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ Long nextLedgerId = ledger.getNextValidLedger(searchPosition.getLedgerId());
Review Comment:
Two concerns
### 1. Will the next ledger be out of the range that users want to search?
---
### 2. Will this progress go to an infinite loop as follows?
0. Entry ranges: [min...unrecoverable...max]
0. ...(find first, find latest...)
0. Calculate `mid()`, the result is an `unrecoverable` entry, we mark it as `Position-U`, then try the next ledger, we mark this position as `Position-Next`
0. Read `Position-Next` successfully, and calculate next `mid()` by `[min, Position-Next]`, get `Position-U` again.
0. Loop to step 3
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2168291149
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,34 +118,131 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
break;
case searching:
if (condition.test(entry)) {
// mid - last
lastMatchedPosition = position;
- min = mid();
+ min = mid;
} else {
// start - mid
- max = mid() - 1;
+ max = mid - 1;
}
+ this.mid = mid();
if (max <= min) {
callback.findEntryComplete(lastMatchedPosition, OpFindNewest.this.ctx);
return;
}
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
}
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException
+ && ledger.getConfig().isAutoSkipNonRecoverableData()) {
+ log.info("[{}] Ledger {} is not recoverable, skip non-recoverable data, state:{}", ledger.getName(),
Review Comment:
Do we need to add `try/catch` block to this new added logic in case of exceptions causing the callback failing to execute?
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#issuecomment-2999894321
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24441?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `75.00000%` with `17 lines` in your changes missing coverage. Please review.
> Project coverage is 74.27%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`e53848c`)](https://app.codecov.io/gh/apache/pulsar/commit/e53848c1d6d143bbb23fc1cdd5830baec3a09c23?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1163 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24441?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...g/apache/bookkeeper/mledger/impl/OpFindNewest.java](https://app.codecov.io/gh/apache/pulsar/pull/24441?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FOpFindNewest.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL09wRmluZE5ld2VzdC5qYXZh) | 75.00% | [2 Missing and 15 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24441?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24441?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24441 +/- ##
============================================
+ Coverage 73.57% 74.27% +0.70%
- Complexity 32624 32753 +129
============================================
Files 1877 1868 -9
Lines 139502 145566 +6064
Branches 15299 16659 +1360
============================================
+ Hits 102638 108126 +5488
+ Misses 28908 28870 -38
- Partials 7956 8570 +614
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24441/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/24441/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.69% <0.00%> (+2.11%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24441/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.34% <0.00%> (-0.98%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24441/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.76% <75.00%> (+0.91%)` | :arrow_up: |
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/24441?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...g/apache/bookkeeper/mledger/impl/OpFindNewest.java](https://app.codecov.io/gh/apache/pulsar/pull/24441?src=pr&el=tree&filepath=managed-ledger%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbookkeeper%2Fmledger%2Fimpl%2FOpFindNewest.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9pbXBsL09wRmluZE5ld2VzdC5qYXZh) | `81.20% <75.00%> (-8.35%)` | :arrow_down: |
... and [1087 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24441/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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2161388423
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,31 +119,93 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
break;
case searching:
if (condition.test(entry)) {
// mid - last
lastMatchedPosition = position;
- min = mid();
+ min = mid;
} else {
// start - mid
- max = mid() - 1;
+ max = mid - 1;
}
+ this.mid = mid();
if (max <= min) {
callback.findEntryComplete(lastMatchedPosition, OpFindNewest.this.ctx);
return;
}
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
}
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException
+ && ledger.getConfig().isAutoSkipNonRecoverableData()) {
+ log.info("[{}] Ledger {} is not recoverable, skip non-recoverable data, state:{}", ledger.getName(),
+ searchPosition, state);
+ Position nextPosition = null;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ Long nextLedgerId = ledger.getNextValidLedger(searchPosition.getLedgerId());
+ if (nextLedgerId != null) {
+ nextPosition = PositionFactory.create(nextLedgerId, 0);
+ }
+ } else {
+ nextPosition = ledger.getNextValidPosition(searchPosition);
+ }
+ switch (state) {
+ case checkFirst:
+ // If we failed to read the first entry, try next valid ledger
+ if (nextPosition != null && nextPosition.getEntryId() != -1) {
+ long numberOfEntries =
+ ledger.getNumberOfEntries(Range.closedOpen(searchPosition, nextPosition));
+ searchPosition = nextPosition;
+ min += numberOfEntries;
+ find();
+ return;
+ }
+ break;
+ case checkLast:
+ // If we failed to read the last entry, try previous ledger's last entry
+ Position prevPosition;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ prevPosition =
+ ledger.getPreviousPosition(PositionFactory.create(searchPosition.getLedgerId(), -1L));
+ } else {
+ prevPosition = ledger.getPreviousPosition(searchPosition);
+ }
+ if (prevPosition.getEntryId() != -1) {
+ long numberOfEntries =
+ ledger.getNumberOfEntries(Range.openClosed(prevPosition, searchPosition));
+ searchPosition = prevPosition;
+ max -= numberOfEntries;
+ find();
+ return;
+ }
+ break;
+ case searching:
+ // In searching state, if we failed to read the mid entry, try next valid ledger
+ if (nextPosition != null && nextPosition.getEntryId() != -1) {
+ searchPosition = nextPosition;
Review Comment:
When skipping a non-recoverable mid entry, the binary search bounds (`min`/`max`) and `startPosition` remain tied to the original ledger, causing subsequent `find()` calls to compute positions incorrectly. Consider recalculating the entry index offsets or resetting bounds relative to the new `searchPosition`.
```suggestion
if (nextPosition != null && nextPosition.getEntryId() != -1) {
long numberOfEntries = ledger.getNumberOfEntries(Range.closedOpen(searchPosition, nextPosition));
searchPosition = nextPosition;
min += numberOfEntries;
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java:
##########
@@ -563,6 +566,78 @@ public void testSeekByTimestamp() throws Exception {
}
}
+ @Test(timeOut = 30_000)
+ public void testSeekByTimestampWithSkipNonRecoverableData() throws Exception {
+ String topicName = "persistent://prop/use/ns-abc/testSeekByTimestampWithSkipNonRecoverableData";
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, "my-sub", MessageId.earliest);
+
+ @Cleanup
+ Producer<String> producer =
+ pulsarClient.newProducer(Schema.STRING).topic(topicName).enableBatching(false).create();
+ for (int i = 0; i < 55; i++) {
+ producer.send(("message-" + i));
+ Thread.sleep(10);
+ }
+
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get();
+
+ Map<Long, MessageIdImpl> timestampToMessageId = new HashMap<>();
+ List<Long> ledgerIds = new ArrayList<>();
+ @Cleanup
+ Reader<String> reader =
+ pulsarClient.newReader(Schema.STRING).topic(topicName).startMessageId(MessageId.earliest).create();
+ while (reader.hasMessageAvailable()) {
+ Message<String> message = reader.readNext();
+ log.info("message: {} ----- {}", message.getMessageId(), message.getPublishTime());
+ timestampToMessageId.put(message.getPublishTime(), (MessageIdImpl) message.getMessageId());
+ long ledgerId = ((MessageIdImpl) message.getMessageId()).getLedgerId();
+ if (!ledgerIds.contains(ledgerId)) {
+ ledgerIds.add(ledgerId);
+ }
+ }
+
+ Assert.assertEquals(timestampToMessageId.size(), 55);
+
+ LinkedHashSet<Long> deletedLedgerIds = new LinkedHashSet<>();
+ deletedLedgerIds.add(ledgerIds.get(0));
+ deletedLedgerIds.add(ledgerIds.get(ledgerIds.size() - 1));
+ int mid = ledgerIds.size() / 2;
+ deletedLedgerIds.add(ledgerIds.get(mid));
+
+ for (Long deletedLedgerId : deletedLedgerIds) {
+ pulsar.getBookKeeperClient().deleteLedger(deletedLedgerId);
+ log.info("delete ledger: {}", deletedLedgerId);
+ }
+
+ restartBroker();
+
+ @Cleanup
+ org.apache.pulsar.client.api.Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .receiverQueueSize(0)
+ .topic(topicName).subscriptionName("my-sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ long[] timestamps = timestampToMessageId.keySet().stream().mapToLong(Long::longValue).toArray();
+ ArraySorter.sort(timestamps);
+
+ timestampToMessageId.values().removeIf(messageId -> deletedLedgerIds.contains(messageId.getLedgerId()));
+
+ for (int i = 0; i < timestamps.length - 5; i++) {
Review Comment:
[nitpick] The hard-coded `- 5` offset in the loop condition is unclear; extract it into a named constant or add a comment explaining why five entries are skipped.
```suggestion
final int SKIPPED_ENTRIES_COUNT = 5; // Number of entries to skip, based on test requirements.
for (int i = 0; i < timestamps.length - SKIPPED_ENTRIES_COUNT; i++) {
```
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2167951919
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,34 +120,131 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
Review Comment:
Yes, they are duplicated, I remove line:104
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2162991713
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java:
##########
@@ -563,6 +566,91 @@ public void testSeekByTimestamp() throws Exception {
}
}
+ @Test(timeOut = 30_000)
+ public void testSeekByTimestampWithSkipNonRecoverableData() throws Exception {
+ String topicName = "persistent://prop/use/ns-abc/testSeekByTimestampWithSkipNonRecoverableData";
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, "my-sub", MessageId.earliest);
+
+ @Cleanup
+ Producer<String> producer =
+ pulsarClient.newProducer(Schema.STRING).topic(topicName).enableBatching(false).create();
+ for (int i = 0; i < 55; i++) {
+ producer.send(("message-" + i));
+ Thread.sleep(10);
+ }
+
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get();
+
+ Map<Long, MessageIdImpl> timestampToMessageId = new HashMap<>();
+ List<Long> ledgerIds = new ArrayList<>();
+ @Cleanup
+ Reader<String> reader =
+ pulsarClient.newReader(Schema.STRING).topic(topicName).startMessageId(MessageId.earliest).create();
+ while (reader.hasMessageAvailable()) {
+ Message<String> message = reader.readNext();
+ log.info("message: {} ----- {}", message.getMessageId(), message.getPublishTime());
+ timestampToMessageId.put(message.getPublishTime(), (MessageIdImpl) message.getMessageId());
+ long ledgerId = ((MessageIdImpl) message.getMessageId()).getLedgerId();
+ if (!ledgerIds.contains(ledgerId)) {
+ ledgerIds.add(ledgerId);
+ }
+ }
+
+ Assert.assertEquals(timestampToMessageId.size(), 55);
+
+ LinkedHashSet<Long> deletedLedgerIds = new LinkedHashSet<>();
+ deletedLedgerIds.add(ledgerIds.get(0));
+ deletedLedgerIds.add(ledgerIds.get(ledgerIds.size() - 1));
+ int mid = ledgerIds.size() / 2;
+ deletedLedgerIds.add(ledgerIds.get(mid));
+
+ for (Long deletedLedgerId : deletedLedgerIds) {
+ pulsar.getBookKeeperClient().deleteLedger(deletedLedgerId);
+ log.info("delete ledger: {}", deletedLedgerId);
+ }
+
+ restartBroker();
Review Comment:
I think we don't need to restart the broker here. If the goal is only to reload the managed ledger, you can just unload the namespace.
```suggestion
admin.namespaces().unload(TopicName.get(topicName).getNamespace());
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java:
##########
@@ -563,6 +566,91 @@ public void testSeekByTimestamp() throws Exception {
}
}
+ @Test(timeOut = 30_000)
+ public void testSeekByTimestampWithSkipNonRecoverableData() throws Exception {
+ String topicName = "persistent://prop/use/ns-abc/testSeekByTimestampWithSkipNonRecoverableData";
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, "my-sub", MessageId.earliest);
+
+ @Cleanup
+ Producer<String> producer =
+ pulsarClient.newProducer(Schema.STRING).topic(topicName).enableBatching(false).create();
+ for (int i = 0; i < 55; i++) {
+ producer.send(("message-" + i));
+ Thread.sleep(10);
+ }
+
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get();
+
+ Map<Long, MessageIdImpl> timestampToMessageId = new HashMap<>();
+ List<Long> ledgerIds = new ArrayList<>();
+ @Cleanup
+ Reader<String> reader =
+ pulsarClient.newReader(Schema.STRING).topic(topicName).startMessageId(MessageId.earliest).create();
+ while (reader.hasMessageAvailable()) {
+ Message<String> message = reader.readNext();
+ log.info("message: {} ----- {}", message.getMessageId(), message.getPublishTime());
+ timestampToMessageId.put(message.getPublishTime(), (MessageIdImpl) message.getMessageId());
+ long ledgerId = ((MessageIdImpl) message.getMessageId()).getLedgerId();
+ if (!ledgerIds.contains(ledgerId)) {
+ ledgerIds.add(ledgerId);
+ }
+ }
+
+ Assert.assertEquals(timestampToMessageId.size(), 55);
+
+ LinkedHashSet<Long> deletedLedgerIds = new LinkedHashSet<>();
+ deletedLedgerIds.add(ledgerIds.get(0));
+ deletedLedgerIds.add(ledgerIds.get(ledgerIds.size() - 1));
+ int mid = ledgerIds.size() / 2;
+ deletedLedgerIds.add(ledgerIds.get(mid));
+
+ for (Long deletedLedgerId : deletedLedgerIds) {
+ pulsar.getBookKeeperClient().deleteLedger(deletedLedgerId);
+ log.info("delete ledger: {}", deletedLedgerId);
+ }
+
+ restartBroker();
+
+ @Cleanup
+ org.apache.pulsar.client.api.Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .receiverQueueSize(0)
+ .topic(topicName).subscriptionName("my-sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest).subscribe();
+ long[] timestamps = timestampToMessageId.keySet().stream().mapToLong(Long::longValue).toArray();
+ ArraySorter.sort(timestamps);
+
+ timestampToMessageId.values().removeIf(messageId -> deletedLedgerIds.contains(messageId.getLedgerId()));
+
+ final int lastNonRecoverableEntryNums = 5;
+
+ for (int i = 0; i < timestamps.length - lastNonRecoverableEntryNums; i++) {
+ MessageIdImpl nextValidMessageId = (MessageIdImpl) timestampToMessageId.get(timestamps[i]);
Review Comment:
```suggestion
MessageIdImpl nextValidMessageId = timestampToMessageId.get(timestamps[i]);
```
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java:
##########
@@ -439,6 +449,153 @@ void testMessageExpiryWithTimestampNonRecoverableException() throws Exception {
}
+ public void testFindMessageWithTimestampAutoSkipNonRecoverable() throws Exception {
+
+ final String ledgerAndCursorName = "testFindMessageWithTimestampAutoSkipNonRecoverable";
+ final int entriesPerLedger = 5;
+ final int totalEntries = 50;
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setRetentionSizeInMB(10);
+ config.setMaxEntriesPerLedger(entriesPerLedger);
+ config.setRetentionTime(1, TimeUnit.HOURS);
+ config.setAutoSkipNonRecoverableData(true);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(ledgerAndCursorName, config);
+
+ long initTimeMillis = System.currentTimeMillis();
+ for (int i = 0; i < totalEntries; i++) {
+ ledger.addEntry(createMessageWrittenToLedger("msg" + i, initTimeMillis + i));
+ }
+ // {0、1、2、3、4} (0) 3 x
+ // {5、6、7、8、9} (1) 4
+ // {10、11、12、13、14} (2) 5
+ // {15、16、17、18、19} (3) 6
+ // {20、21、22、23、24} (4) 7 x
+ // {25、26、27、28、29} (5) 8 x
+ // {30、31、32、33、34} (6) 9
+ // {35、36、37、38、39} (7) 10
+ // {40、41、42、43、44} (8) 11
+ // {45、46、47、48、49} (9) 12 x
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(ledger.getState(), ManagedLedgerImpl.State.LedgerOpened));
+
+ List<LedgerInfo> ledgers = ledger.getLedgersInfoAsList();
+ LedgerInfo lastLedgerInfo = ledgers.get(ledgers.size() - 1);
+ // The `lastLedgerInfo` should be newly opened, and it does not contain any entries.
+ // Please refer to: https://github.com/apache/pulsar/pull/22034
+ assertEquals(lastLedgerInfo.getEntries(), 0);
+ assertEquals(ledgers.size(), totalEntries / entriesPerLedger + 1);
+
+ bkc.deleteLedger(ledgers.get(0).getLedgerId());
+ bkc.deleteLedger(ledgers.get(4).getLedgerId());
+ bkc.deleteLedger(ledgers.get(5).getLedgerId());
+ bkc.deleteLedger(ledgers.get(9).getLedgerId());
+
+ MessageId messageId = findMessageIdByPublishTime(initTimeMillis + 17, ledger).join();
+ log.info("messageId: {}", messageId);
+ assertEquals(messageId, new MessageIdImpl(ledgers.get(3).getLedgerId(), 2, -1));
+
+ messageId = findMessageIdByPublishTime(initTimeMillis + 27, ledger).join();
+ log.info("messageId: {}", messageId);
+ assertEquals(messageId, new MessageIdImpl(ledgers.get(4).getLedgerId(), 0, -1));
+
+ messageId = findMessageIdByPublishTime(initTimeMillis + 43, ledger).join();
+ log.info("messageId: {}", messageId);
+ assertEquals(messageId, new MessageIdImpl(ledgers.get(8).getLedgerId(), 3, -1));
+
+ messageId = findMessageIdByPublishTime(initTimeMillis + 48, ledger).join();
+ log.info("messageId: {}", messageId);
+ assertEquals(messageId, new MessageIdImpl(ledgers.get(9).getLedgerId(), 0, -1));
+
+ ledger.close();
+ factory.shutdown();
+ }
+
+ public void testFindMessageByCursorWithTimestampAutoSkipNonRecoverable() throws Exception {
+
+ final String ledgerAndCursorName = "testFindMessageByCursorWithTimestampAutoSkipNonRecoverable";
+ final int entriesPerLedger = 5;
+ final int totalEntries = 50;
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setRetentionSizeInMB(10);
+ config.setMaxEntriesPerLedger(entriesPerLedger);
+ config.setRetentionTime(1, TimeUnit.HOURS);
+ config.setAutoSkipNonRecoverableData(true);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(ledgerAndCursorName, config);
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor(ledgerAndCursorName);
+
+ long initTimeMillis = System.currentTimeMillis();
+ for (int i = 0; i < totalEntries; i++) {
+ ledger.addEntry(createMessageWrittenToLedger("msg" + i, initTimeMillis + i));
+ }
+ // {0、1、2、3、4} (0) 3 x
+ // {5、6、7、8、9} (1) 4
+ // {10、11、12、13、14} (2) 5
+ // {15、16、17、18、19} (3) 6
+ // {20、21、22、23、24} (4) 7 x
+ // {25、26、27、28、29} (5) 8 x
+ // {30、31、32、33、34} (6) 9
+ // {35、36、37、38、39} (7) 10
+ // {40、41、42、43、44} (8) 11
+ // {45、46、47、48、49} (9) 12 x
Review Comment:
`、` is not a ASCII character, could you replace it with `, `?
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SubscriptionSeekTest.java:
##########
@@ -563,6 +566,91 @@ public void testSeekByTimestamp() throws Exception {
}
}
+ @Test(timeOut = 30_000)
+ public void testSeekByTimestampWithSkipNonRecoverableData() throws Exception {
+ String topicName = "persistent://prop/use/ns-abc/testSeekByTimestampWithSkipNonRecoverableData";
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscription(topicName, "my-sub", MessageId.earliest);
+
+ @Cleanup
+ Producer<String> producer =
+ pulsarClient.newProducer(Schema.STRING).topic(topicName).enableBatching(false).create();
+ for (int i = 0; i < 55; i++) {
+ producer.send(("message-" + i));
+ Thread.sleep(10);
+ }
+
+ PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get();
+
Review Comment:
```suggestion
```
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#issuecomment-2998545848
> As I understand, it should be an expected behavior for the ledger removed from the metadata storage. It might cause by BUGs, or users deleted the ledger manually without deleting the topic, right?
Yes, this is an expected behavior, this fix only to unblock unblock seeking by timestamp for topic with non-recoverable ledger. for why this expected behavior occurs, we need to continue investigating.
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#issuecomment-2997007044
> When some ledgers are not offloaded successfully, they will be removed from the metadata store so that seek by timestamp might fail.
@coderzc Could you please elaborate above description? It's not clear about why the ledger will be removed from the metadata store when they are not offloaded successfully.
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2162869038
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,31 +119,93 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
break;
case searching:
if (condition.test(entry)) {
// mid - last
lastMatchedPosition = position;
- min = mid();
+ min = mid;
} else {
// start - mid
- max = mid() - 1;
+ max = mid - 1;
}
+ this.mid = mid();
if (max <= min) {
callback.findEntryComplete(lastMatchedPosition, OpFindNewest.this.ctx);
return;
}
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
}
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException
+ && ledger.getConfig().isAutoSkipNonRecoverableData()) {
+ log.info("[{}] Ledger {} is not recoverable, skip non-recoverable data, state:{}", ledger.getName(),
+ searchPosition, state);
+ Position nextPosition = null;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ Long nextLedgerId = ledger.getNextValidLedger(searchPosition.getLedgerId());
Review Comment:
> 2. Will this progress go to an infinite loop as follows?
3. If read Position-Next successfully, and calculate next mid() by [min, Position-U), since last mid don't been changed.
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2162906619
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,31 +119,93 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
break;
case searching:
if (condition.test(entry)) {
// mid - last
lastMatchedPosition = position;
- min = mid();
+ min = mid;
} else {
// start - mid
- max = mid() - 1;
+ max = mid - 1;
}
+ this.mid = mid();
if (max <= min) {
callback.findEntryComplete(lastMatchedPosition, OpFindNewest.this.ctx);
return;
}
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
}
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException
+ && ledger.getConfig().isAutoSkipNonRecoverableData()) {
+ log.info("[{}] Ledger {} is not recoverable, skip non-recoverable data, state:{}", ledger.getName(),
+ searchPosition, state);
+ Position nextPosition = null;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ Long nextLedgerId = ledger.getNextValidLedger(searchPosition.getLedgerId());
Review Comment:
> 1. Will the next ledger be out of the range that users want to search?
@poorbarcode I've added boundary checks and refactored the code, 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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2162917043
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,31 +119,93 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
break;
case searching:
if (condition.test(entry)) {
// mid - last
lastMatchedPosition = position;
- min = mid();
+ min = mid;
} else {
// start - mid
- max = mid() - 1;
+ max = mid - 1;
}
+ this.mid = mid();
if (max <= min) {
callback.findEntryComplete(lastMatchedPosition, OpFindNewest.this.ctx);
return;
}
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
}
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException
+ && ledger.getConfig().isAutoSkipNonRecoverableData()) {
+ log.info("[{}] Ledger {} is not recoverable, skip non-recoverable data, state:{}", ledger.getName(),
+ searchPosition, state);
+ Position nextPosition = null;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ Long nextLedgerId = ledger.getNextValidLedger(searchPosition.getLedgerId());
+ if (nextLedgerId != null) {
+ nextPosition = PositionFactory.create(nextLedgerId, 0);
Review Comment:
Add check logic
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc closed pull request #24441: [fix][ml] Enhance OpFindNewest to support skip non-recoverable data
URL: https://github.com/apache/pulsar/pull/24441
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2161405816
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,31 +119,93 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
break;
case searching:
if (condition.test(entry)) {
// mid - last
lastMatchedPosition = position;
- min = mid();
+ min = mid;
} else {
// start - mid
- max = mid() - 1;
+ max = mid - 1;
}
+ this.mid = mid();
if (max <= min) {
callback.findEntryComplete(lastMatchedPosition, OpFindNewest.this.ctx);
return;
}
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
find();
}
}
@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
+ if (exception instanceof ManagedLedgerException.NonRecoverableLedgerException
+ && ledger.getConfig().isAutoSkipNonRecoverableData()) {
+ log.info("[{}] Ledger {} is not recoverable, skip non-recoverable data, state:{}", ledger.getName(),
+ searchPosition, state);
+ Position nextPosition = null;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ Long nextLedgerId = ledger.getNextValidLedger(searchPosition.getLedgerId());
+ if (nextLedgerId != null) {
+ nextPosition = PositionFactory.create(nextLedgerId, 0);
+ }
+ } else {
+ nextPosition = ledger.getNextValidPosition(searchPosition);
+ }
+ switch (state) {
+ case checkFirst:
+ // If we failed to read the first entry, try next valid ledger
+ if (nextPosition != null && nextPosition.getEntryId() != -1) {
+ long numberOfEntries =
+ ledger.getNumberOfEntries(Range.closedOpen(searchPosition, nextPosition));
+ searchPosition = nextPosition;
+ min += numberOfEntries;
+ find();
+ return;
+ }
+ break;
+ case checkLast:
+ // If we failed to read the last entry, try previous ledger's last entry
+ Position prevPosition;
+ if (exception instanceof ManagedLedgerException.LedgerNotExistException) {
+ prevPosition =
+ ledger.getPreviousPosition(PositionFactory.create(searchPosition.getLedgerId(), -1L));
+ } else {
+ prevPosition = ledger.getPreviousPosition(searchPosition);
+ }
+ if (prevPosition.getEntryId() != -1) {
+ long numberOfEntries =
+ ledger.getNumberOfEntries(Range.openClosed(prevPosition, searchPosition));
+ searchPosition = prevPosition;
+ max -= numberOfEntries;
+ find();
+ return;
+ }
+ break;
+ case searching:
+ // In searching state, if we failed to read the mid entry, try next valid ledger
+ if (nextPosition != null && nextPosition.getEntryId() != -1) {
+ searchPosition = nextPosition;
Review Comment:
I don't think we should adjust min, because we still don't know if the message of searchPosition meets the conditions.
--
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][ml] Enhance OpFindNewest to support skip non-recoverable data [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #24441:
URL: https://github.com/apache/pulsar/pull/24441#discussion_r2167843132
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpFindNewest.java:
##########
@@ -115,34 +120,131 @@ public void readEntryComplete(Entry entry, Object ctx) {
} else {
// start binary search
state = State.searching;
- searchPosition = ledger.getPositionAfterN(startPosition, mid(), PositionBound.startExcluded);
+ this.mid = mid();
+ searchPosition = ledger.getPositionAfterN(startPosition, this.mid, PositionBound.startExcluded);
Review Comment:
@coderzc could you please also help check line:104 which was added by https://github.com/apache/pulsar/pull/22891/files#diff-7d6cd459f9134d4aeda2418630110e5abefdf6d367f6e83747c65f8637a6f851R99-R100? It should be duplicated with line:102?
--
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