shibd commented on code in PR #15568:
URL: https://github.com/apache/pulsar/pull/15568#discussion_r886682545
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java:
##########
@@ -48,39 +46,46 @@ public class NonDurableCursorImpl extends ManagedCursorImpl
{
// Start from last entry
switch (initialPosition) {
case Latest:
-
initializeCursorPosition(ledger.getLastPositionAndCounter());
+
initializeCursorPosition(ledger.getLastPositionAndCounter(), inclusive);
break;
case Earliest:
-
initializeCursorPosition(ledger.getFirstPositionAndCounter());
+
initializeCursorPosition(ledger.getFirstPositionAndCounter(), inclusive);
break;
}
} else if (startCursorPosition.getLedgerId() ==
PositionImpl.EARLIEST.getLedgerId()) {
// Start from invalid ledger to read from first available entry
-
recoverCursor(ledger.getPreviousPosition(ledger.getFirstPosition()));
+ recoverCursor(ledger.getFirstPosition(), inclusive);
} else {
// Since the cursor is positioning on the mark-delete position, we
need to take 1 step back from the desired
// read-position
- recoverCursor(startCursorPosition);
+ recoverCursor(startCursorPosition, inclusive);
Review Comment:
Is it possible to add a new method in ledger:
`getSpecifyPositionAndCounter(Position startPosition, boolean inclusive)`
and this can call:
`initializeCursorPosition(ledger.getSpecifyPositionAndCounter(startCursorPosition,
inclusive))`
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -939,7 +953,7 @@ public void operationComplete() {
cursor.setActive();
// Update the ack position (ignoring entries that were written
while the cursor was being created)
cursor.initializeCursorPosition(initialPosition ==
InitialPosition.Latest ? getLastPositionAndCounter()
Review Comment:
According to the PIP. Just when startMessageId == latest, need judge
inclusive. We can keep the initializeCursorPosition method unchanged and add
inclusive param to the getLastPositionAndCounter method.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:
##########
@@ -878,61 +884,64 @@ private CompletableFuture<Subscription>
getDurableSubscription(String subscripti
Map<String, Long> properties =
PersistentSubscription.getBaseCursorProperties(replicated);
- ledger.asyncOpenCursor(Codec.encode(subscriptionName),
initialPosition, properties, new OpenCursorCallback() {
- @Override
- public void openCursorComplete(ManagedCursor cursor, Object ctx) {
- if (log.isDebugEnabled()) {
- log.debug("[{}][{}] Opened cursor", topic,
subscriptionName);
- }
+ ledger.asyncOpenCursor(Codec.encode(subscriptionName),
initialPosition, properties, startMessageIdInclusive,
Review Comment:
Seems like only the code format has changed?
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/NonDurableCursorImpl.java:
##########
@@ -48,39 +46,46 @@ public class NonDurableCursorImpl extends ManagedCursorImpl
{
// Start from last entry
switch (initialPosition) {
case Latest:
-
initializeCursorPosition(ledger.getLastPositionAndCounter());
+
initializeCursorPosition(ledger.getLastPositionAndCounter(), inclusive);
Review Comment:
According to the PIP. Just when startMessageId == latest, need judge
inclusive. We can keep the initializeCursorPosition method unchanged and add
inclusive param to the getLastPositionAndCounter method.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
