Skip to content

Avoid problem that topic becomes unavailable due to failure of cursor recovery #2673

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.

Sign up for GitHub

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

Merged
merlimat merged 2 commits into from
Oct 2, 2018
Merged

Avoid problem that topic becomes unavailable due to failure of cursor recovery #2673

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
30a0788
Avoid problem that topic becomes unavailable due to failure of cursor…
Sep 28, 2018
20592e2
Add unit test
Oct 1, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,11 +348,16 @@ private void recoveredCursor(PositionImpl position, Map<String, Long> properties
if (!ledger.ledgerExists(position.getLedgerId())) {
Long nextExistingLedger = ledger.getNextValidLedger(position.getLedgerId());
if (nextExistingLedger == null) {
log.info("[{}-{}] Couldn't find next next valid ledger for recovery {}", ledger.getName(), name,
log.info("[{}] [{}] Couldn't find next next valid ledger for recovery {}", ledger.getName(), name,
position);
}
position = nextExistingLedger != null ? PositionImpl.get(nextExistingLedger, -1) : position;
}
if (position.compareTo(ledger.getLastPosition()) > 0) {
log.warn("[{}] [{}] Current position {} is ahead of last position {}", ledger.getName(), name, position,
ledger.getLastPosition());
position = PositionImpl.get(ledger.getLastPosition());
}
log.info("[{}] Cursor {} recovered to position {}", ledger.getName(), name, position);

messagesConsumedCounter = -getNumberOfEntries(Range.openClosed(position, ledger.getLastPosition()));
Expand Down
60 changes: 60 additions & 0 deletions managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,11 @@
*/
package org.apache.bookkeeper.mledger.impl;

import static org.mockito.Mockito.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.*;

import com.google.common.base.Charsets;
Expand All @@ -42,6 +47,7 @@
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.BookKeeper.DigestType;
import org.apache.bookkeeper.client.LedgerEntry;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
Expand All @@ -66,6 +72,9 @@
import org.apache.bookkeeper.mledger.proto.MLDataFormats.PositionInfo;
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.zookeeper.KeeperException.Code;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.Test;
Expand Down Expand Up @@ -2708,5 +2717,56 @@ public void testEstimatedUnackedSize() throws Exception {
assertEquals(cursor.getEstimatedSizeSinceMarkDeletePosition(), 10 * entryData.length);
}

@Test(timeOut = 20000)
public void testRecoverCursorAheadOfLastPosition() throws Exception {
final String mlName = "my_test_ledger";
final PositionImpl lastPosition = new PositionImpl(1L, 10L);
final PositionImpl nextPosition = new PositionImpl(3L, -1L);

final String cursorName = "my_test_cursor";
final long cursorsLedgerId = -1L;
final long markDeleteLedgerId = 2L;
final long markDeleteEntryId = -1L;

MetaStoreImplZookeeper mockMetaStore = mock(MetaStoreImplZookeeper.class);
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) {
ManagedCursorInfo info = ManagedCursorInfo.newBuilder().setCursorsLedgerId(cursorsLedgerId)
.setMarkDeleteLedgerId(markDeleteLedgerId).setMarkDeleteEntryId(markDeleteEntryId)
.setLastActive(0L).build();
Stat stat = mock(Stat.class);
MetaStoreCallback<ManagedCursorInfo> callback = (MetaStoreCallback<ManagedCursorInfo>) invocation
.getArguments()[2];
callback.operationComplete(info, stat);
return null;
}
}).when(mockMetaStore).asyncGetCursorInfo(eq(mlName), eq(cursorName), any(MetaStoreCallback.class));

ManagedLedgerImpl ml = mock(ManagedLedgerImpl.class);
when(ml.getName()).thenReturn(mlName);
when(ml.getStore()).thenReturn(mockMetaStore);
when(ml.getLastPosition()).thenReturn(lastPosition);
when(ml.getNextValidLedger(markDeleteLedgerId)).thenReturn(3L);
when(ml.getNextValidPosition(lastPosition)).thenReturn(nextPosition);
when(ml.ledgerExists(markDeleteLedgerId)).thenReturn(false);

BookKeeper mockBookKeeper = mock(BookKeeper.class);
final ManagedCursorImpl cursor = new ManagedCursorImpl(mockBookKeeper, new ManagedLedgerConfig(), ml,
cursorName);

cursor.recover(new VoidCallback() {
@Override
public void operationComplete() {
assertEquals(cursor.getMarkDeletedPosition(), lastPosition);
assertEquals(cursor.getReadPosition(), nextPosition);
}

@Override
public void operationFailed(ManagedLedgerException exception) {
fail("Cursor recovery should not fail");
}
});
}

private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class);
}