This is an automated email from the ASF dual-hosted git repository.
jai1 pushed a commit to branch branch-1.22
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/branch-1.22 by this push:
new 1875841 Fix: handle invalid markDelete position at managed-cursor
(#1718)
1875841 is described below
commit 187584155bcacec0900e8f6203a5f81a1e60ec7d
Author: Jai Asher <j...@ccs.neu.edu>
AuthorDate: Wed May 2 13:14:48 2018 -0700
Fix: handle invalid markDelete position at managed-cursor (#1718)
---
.../bookkeeper/mledger/impl/ManagedCursorImpl.java | 22 ++++++++-
.../bookkeeper/mledger/impl/ManagedCursorTest.java | 56 ++++++++++++++++++++++
2 files changed, 77 insertions(+), 1 deletion(-)
diff --git
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
index 22cdf3d..8172784 100644
---
a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
+++
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
@@ -1291,7 +1291,7 @@ public class ManagedCursorImpl implements ManagedCursor {
final MarkDeleteCallback callback, final Object ctx) {
checkNotNull(position);
checkArgument(position instanceof PositionImpl);
-
+
if (STATE_UPDATER.get(this) == State.Closed) {
callback.markDeleteFailed(new ManagedLedgerException("Cursor was
already closed"), ctx);
return;
@@ -1312,6 +1312,16 @@ public class ManagedCursorImpl implements ManagedCursor {
log.debug("[{}] Mark delete cursor {} up to position: {}",
ledger.getName(), name, position);
}
PositionImpl newPosition = (PositionImpl) position;
+
+ if (((PositionImpl)
ledger.getLastConfirmedEntry()).compareTo(newPosition) < 0) {
+ if (log.isDebugEnabled()) {
+ log.debug(
+ "[{}] Failed mark delete due to invalid markDelete {}
is ahead of last-confirmed-entry {} for cursor [{}]",
+ ledger.getName(), position,
ledger.getLastConfirmedEntry(), name);
+ }
+ callback.markDeleteFailed(new ManagedLedgerException("Invalid mark
deleted position"), ctx);
+ return;
+ }
lock.writeLock().lock();
try {
@@ -1509,6 +1519,16 @@ public class ManagedCursorImpl implements ManagedCursor {
ledger.getName(), name, pos,
individualDeletedMessages, markDeletePosition, previousPosition);
}
+ if (((PositionImpl)
ledger.getLastConfirmedEntry()).compareTo(position) < 0) {
+ if (log.isDebugEnabled()) {
+ log.debug(
+ "[{}] Failed mark delete due to invalid markDelete
{} is ahead of last-confirmed-entry {} for cursor [{}]",
+ ledger.getName(), position,
ledger.getLastConfirmedEntry(), name);
+ }
+ callback.deleteFailed(new ManagedLedgerException("Invalid mark
deleted position"), ctx);
+ return;
+ }
+
if (individualDeletedMessages.contains(position) ||
position.compareTo(markDeletePosition) <= 0) {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Position was already deleted {}",
ledger.getName(), name, position);
diff --git
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
index 61e9b3d..59e1594 100644
---
a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
+++
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
@@ -49,6 +49,7 @@ import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
@@ -2614,5 +2615,60 @@ public class ManagedCursorTest extends
MockedBookKeeperTestCase {
assertEquals(entries.size(), totalAddEntries / 2);
}
+ @Test
+ public void testInvalidMarkDelete() throws Exception {
+ ManagedLedger ledger = factory.open("my_test_ledger", new
ManagedLedgerConfig());
+
+ ManagedCursor cursor = ledger.openCursor("c1");
+ Position readPosition = cursor.getReadPosition();
+ Position markDeletePosition = cursor.getMarkDeletedPosition();
+
+ List<Position> addedPositions = new ArrayList<>();
+ for (int i = 0; i < 20; i++) {
+ Position p = ledger.addEntry(("dummy-entry-" +
i).getBytes(Encoding));
+ addedPositions.add(p);
+ }
+
+ // validate: cursor.asyncMarkDelete(..)
+ CountDownLatch markDeleteCallbackLatch = new CountDownLatch(1);
+ Position position = PositionImpl.get(100, 100);
+ AtomicBoolean markDeleteCallFailed = new AtomicBoolean(false);
+ cursor.asyncMarkDelete(position, new MarkDeleteCallback() {
+ @Override
+ public void markDeleteComplete(Object ctx) {
+ markDeleteCallbackLatch.countDown();
+ }
+
+ @Override
+ public void markDeleteFailed(ManagedLedgerException exception,
Object ctx) {
+ markDeleteCallFailed.set(true);
+ markDeleteCallbackLatch.countDown();
+ }
+ }, null);
+ markDeleteCallbackLatch.await();
+ assertEquals(readPosition, cursor.getReadPosition());
+ assertEquals(markDeletePosition, cursor.getMarkDeletedPosition());
+
+ // validate : cursor.asyncDelete(..)
+ CountDownLatch deleteCallbackLatch = new CountDownLatch(1);
+ markDeleteCallFailed.set(false);
+ cursor.asyncDelete(position, new DeleteCallback() {
+ @Override
+ public void deleteComplete(Object ctx) {
+ deleteCallbackLatch.countDown();
+ }
+
+ @Override
+ public void deleteFailed(ManagedLedgerException exception, Object
ctx) {
+ markDeleteCallFailed.set(true);
+ deleteCallbackLatch.countDown();
+ }
+ }, null);
+
+ deleteCallbackLatch.await();
+ assertEquals(readPosition, cursor.getReadPosition());
+ assertEquals(markDeletePosition, cursor.getMarkDeletedPosition());
+ }
+
private static final Logger log =
LoggerFactory.getLogger(ManagedCursorTest.class);
}
--
To stop receiving notification emails like this one, please contact
j...@apache.org.
