Repository: bookkeeper Updated Branches: refs/heads/master 90e4e46bc -> 256e75873
BOOKKEEPER-1092: Ledger Recovery (Part 5) - Add Test Case for Parallel Ledger Recovery This change is based on #180 (ebf7020 is the change for review) - Add test case for parallel ledger recovery - More stats for ledger recovery Author: Sijie Guo <[email protected]> Author: Sijie Guo <[email protected]> Reviewers: Enrico Olivelli <[email protected]>, Jia Zhai <None>, Matteo Merli <[email protected]> This closes #182 from sijie/recovery_improvements_part5 Project: http://git-wip-us.apache.org/repos/asf/bookkeeper/repo Commit: http://git-wip-us.apache.org/repos/asf/bookkeeper/commit/256e7587 Tree: http://git-wip-us.apache.org/repos/asf/bookkeeper/tree/256e7587 Diff: http://git-wip-us.apache.org/repos/asf/bookkeeper/diff/256e7587 Branch: refs/heads/master Commit: 256e7587331eddd51b7c05f27717c9d4dc848500 Parents: 90e4e46 Author: Sijie Guo <[email protected]> Authored: Wed Jun 21 12:00:15 2017 -0700 Committer: Sijie Guo <[email protected]> Committed: Wed Jun 21 12:00:15 2017 -0700 ---------------------------------------------------------------------- .../apache/bookkeeper/client/BookKeeper.java | 9 + .../client/BookKeeperClientStats.java | 1 + .../apache/bookkeeper/client/LedgerHandle.java | 63 +- .../bookkeeper/client/LedgerRecoveryOp.java | 8 +- .../bookkeeper/meta/CleanupLedgerManager.java | 6 + .../proto/BookkeeperInternalCallbacks.java | 4 +- .../client/ParallelLedgerRecoveryTest.java | 618 +++++++++++++++++++ 7 files changed, 691 insertions(+), 18 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/256e7587/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java index acc9f0a..9a09084 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java @@ -20,6 +20,7 @@ */ package org.apache.bookkeeper.client; +import com.google.common.annotations.VisibleForTesting; import io.netty.channel.EventLoopGroup; import io.netty.channel.epoll.EpollEventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; @@ -98,6 +99,7 @@ public class BookKeeper implements AutoCloseable { private OpStatsLogger createOpLogger; private OpStatsLogger openOpLogger; private OpStatsLogger deleteOpLogger; + private OpStatsLogger recoverOpLogger; private OpStatsLogger readOpLogger; private OpStatsLogger addOpLogger; private OpStatsLogger writeLacOpLogger; @@ -447,6 +449,11 @@ public class BookKeeper implements AutoCloseable { return ledgerManager; } + @VisibleForTesting + LedgerManager getUnderlyingLedgerManager() { + return ((CleanupLedgerManager) ledgerManager).getUnderlying(); + } + LedgerIdGenerator getLedgerIdGenerator() { return ledgerIdGenerator; } @@ -1243,6 +1250,7 @@ public class BookKeeper implements AutoCloseable { createOpLogger = stats.getOpStatsLogger(BookKeeperClientStats.CREATE_OP); deleteOpLogger = stats.getOpStatsLogger(BookKeeperClientStats.DELETE_OP); openOpLogger = stats.getOpStatsLogger(BookKeeperClientStats.OPEN_OP); + recoverOpLogger = stats.getOpStatsLogger(BookKeeperClientStats.RECOVER_OP); readOpLogger = stats.getOpStatsLogger(BookKeeperClientStats.READ_OP); addOpLogger = stats.getOpStatsLogger(BookKeeperClientStats.ADD_OP); writeLacOpLogger = stats.getOpStatsLogger(BookKeeperClientStats.WRITE_LAC_OP); @@ -1254,6 +1262,7 @@ public class BookKeeper implements AutoCloseable { OpStatsLogger getCreateOpLogger() { return createOpLogger; } OpStatsLogger getOpenOpLogger() { return openOpLogger; } OpStatsLogger getDeleteOpLogger() { return deleteOpLogger; } + OpStatsLogger getRecoverOpLogger() { return recoverOpLogger; } OpStatsLogger getReadOpLogger() { return readOpLogger; } OpStatsLogger getAddOpLogger() { return addOpLogger; } OpStatsLogger getWriteLacOpLogger() { return writeLacOpLogger; } http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/256e7587/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperClientStats.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperClientStats.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperClientStats.java index 6166c95..e98b2d6 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperClientStats.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeperClientStats.java @@ -29,6 +29,7 @@ public interface BookKeeperClientStats { public final static String CREATE_OP = "LEDGER_CREATE"; public final static String DELETE_OP = "LEDGER_DELETE"; public final static String OPEN_OP = "LEDGER_OPEN"; + public final static String RECOVER_OP = "LEDGER_RECOVER"; public final static String LEDGER_RECOVER_READ_ENTRIES = "LEDGER_RECOVER_READ_ENTRIES"; public final static String LEDGER_RECOVER_ADD_ENTRIES = "LEDGER_RECOVER_ADD_ENTRIES"; http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/256e7587/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java index 7a9d81b..91700b6 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java @@ -22,6 +22,7 @@ package org.apache.bookkeeper.client; import static com.google.common.base.Charsets.UTF_8; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Sets; import com.google.common.util.concurrent.RateLimiter; import io.netty.buffer.ByteBuf; @@ -49,7 +50,9 @@ import org.apache.bookkeeper.client.AsyncCallback.ReadLastConfirmedCallback; import org.apache.bookkeeper.client.BookKeeper.DigestType; import org.apache.bookkeeper.net.BookieSocketAddress; import org.apache.bookkeeper.proto.BookieProtocol; +import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks; import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback; +import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.TimedGenericCallback; import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat.State; import org.apache.bookkeeper.stats.Counter; import org.apache.bookkeeper.stats.Gauge; @@ -75,8 +78,9 @@ public class LedgerHandle implements AutoCloseable { long length; final DigestManager macManager; final DistributionSchedule distributionSchedule; - final RateLimiter throttler; + final boolean enableParallelRecoveryRead; + final int recoveryReadBatchSize; /** * Invalid entry id. This value is returned from methods which @@ -108,7 +112,8 @@ public class LedgerHandle implements AutoCloseable { this.bk = bk; this.metadata = metadata; this.pendingAddOps = new ConcurrentLinkedQueue<PendingAddOp>(); - + this.enableParallelRecoveryRead = bk.getConf().getEnableParallelRecoveryRead(); + this.recoveryReadBatchSize = bk.getConf().getRecoveryReadBatchSize(); if (metadata.isClosed()) { lastAddConfirmed = lastAddPushed = metadata.getLastEntryId(); @@ -1442,15 +1447,42 @@ public class LedgerHandle implements AutoCloseable { bk.getLedgerManager().readLedgerMetadata(ledgerId, cb); } - void recover(final GenericCallback<Void> cb) { + void recover(GenericCallback<Void> finalCb) { + recover(finalCb, null, false); + } + + /** + * Recover the ledger. + * + * @param finalCb + * callback after recovery is done. + * @param listener + * read entry listener on recovery reads. + * @param forceRecovery + * force the recovery procedure even the ledger metadata shows the ledger is closed. + */ + void recover(GenericCallback<Void> finalCb, + final @VisibleForTesting BookkeeperInternalCallbacks.ReadEntryListener listener, + final boolean forceRecovery) { + final GenericCallback<Void> cb = new TimedGenericCallback<Void>( + finalCb, + BKException.Code.OK, + bk.getRecoverOpLogger()); boolean wasClosed = false; boolean wasInRecovery = false; synchronized (this) { if (metadata.isClosed()) { - lastAddConfirmed = lastAddPushed = metadata.getLastEntryId(); - length = metadata.getLength(); - wasClosed = true; + if (forceRecovery) { + wasClosed = false; + // mark the ledger back to in recovery state, so it would proceed ledger recovery again. + wasInRecovery = false; + metadata.markLedgerInRecovery(); + } else { + lastAddConfirmed = lastAddPushed = metadata.getLastEntryId(); + length = metadata.getLength(); + wasClosed = true; + } } else { wasClosed = false; if (metadata.isInRecovery()) { @@ -1472,8 +1504,9 @@ public class LedgerHandle implements AutoCloseable { // if metadata is already in recover, dont try to write again, // just do the recovery from the starting point new LedgerRecoveryOp(LedgerHandle.this, cb) - .parallelRead(bk.getConf().getEnableParallelRecoveryRead()) - .readBatchSize(bk.getConf().getRecoveryReadBatchSize()) + .parallelRead(enableParallelRecoveryRead) + .readBatchSize(recoveryReadBatchSize) + .setEntryListener(listener) .initiate(); return; } @@ -1490,7 +1523,7 @@ public class LedgerHandle implements AutoCloseable { cb.operationComplete(rc, null); } else { metadata = newMeta; - recover(cb); + recover(cb, listener, forceRecovery); } } @@ -1500,12 +1533,16 @@ public class LedgerHandle implements AutoCloseable { } }); } else if (rc == BKException.Code.OK) { + // we only could issue recovery operation after we successfully update the ledger state to in recovery + // otherwise, it couldn't prevent us advancing last confirmed while the other writer is closing the ledger, + // which will cause inconsistent last add confirmed on bookies & zookeeper metadata. new LedgerRecoveryOp(LedgerHandle.this, cb) - .parallelRead(bk.getConf().getEnableParallelRecoveryRead()) - .readBatchSize(bk.getConf().getRecoveryReadBatchSize()) - .initiate(); + .parallelRead(enableParallelRecoveryRead) + .readBatchSize(recoveryReadBatchSize) + .setEntryListener(listener) + .initiate(); } else { - LOG.error("Error writing ledger config " + rc + " of ledger " + ledgerId); + LOG.error("Error writing ledger config {} of ledger {}", rc, ledgerId); cb.operationComplete(rc, null); } } http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/256e7587/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java index 97d1885..5ea8935 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerRecoveryOp.java @@ -110,9 +110,11 @@ class LedgerRecoveryOp implements ReadEntryListener, AddCallback { new ReadLastConfirmedOp.LastConfirmedDataCallback() { public void readLastConfirmedDataComplete(int rc, RecoveryData data) { if (rc == BKException.Code.OK) { - lh.lastAddPushed = lh.lastAddConfirmed = data.lastAddConfirmed; - lh.length = data.length; - startEntryToRead = endEntryToRead = lh.lastAddConfirmed; + synchronized (lh) { + lh.lastAddPushed = lh.lastAddConfirmed = data.lastAddConfirmed; + lh.length = data.length; + startEntryToRead = endEntryToRead = lh.lastAddConfirmed; + } // keep a copy of ledger metadata before proceeding // ledger recovery metadataForRecovery = new LedgerMetadata(lh.getLedgerMetadata()); http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/256e7587/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java index 961e0d1..50117fe 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/CleanupLedgerManager.java @@ -17,6 +17,7 @@ */ package org.apache.bookkeeper.meta; +import com.google.common.annotations.VisibleForTesting; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.LedgerMetadata; import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback; @@ -79,6 +80,11 @@ public class CleanupLedgerManager implements LedgerManager { this.underlying = lm; } + @VisibleForTesting + public LedgerManager getUnderlying() { + return underlying; + } + private void addCallback(GenericCallback callback) { callbacks.put(callback, callback); } http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/256e7587/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java index 8de9783..79790c4 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookkeeperInternalCallbacks.java @@ -83,7 +83,7 @@ public class BookkeeperInternalCallbacks { void operationComplete(int rc, T result); } - public static class TimedGenericCallback<T> implements GenericCallback<T> { + public static class TimedGenericCallback<T> implements GenericCallback<T> { final GenericCallback<T> cb; final int successRc; @@ -107,7 +107,7 @@ public class BookkeeperInternalCallbacks { cb.operationComplete(rc, result); } } - + public interface ReadEntryCallbackCtx { void setLastAddConfirmed(long lac); long getLastAddConfirmed(); http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/256e7587/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java ---------------------------------------------------------------------- diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java new file mode 100644 index 0000000..850a790 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ParallelLedgerRecoveryTest.java @@ -0,0 +1,618 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.bookkeeper.client; + +import static com.google.common.base.Charsets.UTF_8; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.bookkeeper.bookie.Bookie; +import org.apache.bookkeeper.bookie.BookieException; +import org.apache.bookkeeper.bookie.InterleavedLedgerStorage; +import org.apache.bookkeeper.client.BookKeeper.DigestType; +import org.apache.bookkeeper.conf.ClientConfiguration; +import org.apache.bookkeeper.conf.ServerConfiguration; +import org.apache.bookkeeper.net.BookieSocketAddress; +import org.apache.bookkeeper.proto.BookieProtocol; +import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback; +import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener; +import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.Processor; +import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback; +import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory; +import org.apache.bookkeeper.meta.LedgerManager; +import org.apache.bookkeeper.test.BookKeeperClusterTestCase; +import org.apache.bookkeeper.versioning.Version; +import org.apache.zookeeper.AsyncCallback.VoidCallback; +import org.apache.zookeeper.KeeperException; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Enumeration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +public class ParallelLedgerRecoveryTest extends BookKeeperClusterTestCase { + + static final Logger LOG = LoggerFactory.getLogger(ParallelLedgerRecoveryTest.class); + + static class TestLedgerManager implements LedgerManager { + + final LedgerManager lm; + volatile CountDownLatch waitLatch = null; + final ExecutorService executorService; + + TestLedgerManager(LedgerManager lm) { + this.lm = lm; + this.executorService = Executors.newSingleThreadExecutor(); + } + + void setLatch(CountDownLatch waitLatch) { + this.waitLatch = waitLatch; + } + + @Override + public void createLedgerMetadata(long ledgerId, LedgerMetadata metadata, GenericCallback<Void> cb) { + lm.createLedgerMetadata(ledgerId, metadata, cb); + } + + @Override + public void removeLedgerMetadata(long ledgerId, Version version, GenericCallback<Void> cb) { + lm.removeLedgerMetadata(ledgerId, version, cb); + } + + @Override + public void readLedgerMetadata(long ledgerId, GenericCallback<LedgerMetadata> readCb) { + lm.readLedgerMetadata(ledgerId, readCb); + } + + @Override + public LedgerRangeIterator getLedgerRanges() { + return lm.getLedgerRanges(); + } + + @Override + public void writeLedgerMetadata(final long ledgerId, final LedgerMetadata metadata, + final GenericCallback<Void> cb) { + final CountDownLatch cdl = waitLatch; + if (null != cdl) { + executorService.submit(new Runnable() { + @Override + public void run() { + try { + cdl.await(); + } catch (InterruptedException e) { + LOG.error("Interrupted on waiting latch : ", e); + } + lm.writeLedgerMetadata(ledgerId, metadata, cb); + } + }); + } else { + lm.writeLedgerMetadata(ledgerId, metadata, cb); + } + } + + @Override + public void registerLedgerMetadataListener(long ledgerId, LedgerMetadataListener listener) { + lm.registerLedgerMetadataListener(ledgerId, listener); + } + + @Override + public void unregisterLedgerMetadataListener(long ledgerId, LedgerMetadataListener listener) { + lm.unregisterLedgerMetadataListener(ledgerId, listener); + } + + @Override + public void asyncProcessLedgers(Processor<Long> processor, VoidCallback finalCb, Object context, + int successRc, int failureRc) { + lm.asyncProcessLedgers(processor, finalCb, context, successRc, failureRc); + } + + @Override + public void close() throws IOException { + lm.close(); + executorService.shutdown(); + } + } + + static class TestLedgerManagerFactory extends HierarchicalLedgerManagerFactory { + @Override + public LedgerManager newLedgerManager() { + return new TestLedgerManager(super.newLedgerManager()); + } + } + + final DigestType digestType; + + public ParallelLedgerRecoveryTest() { + super(3); + baseConf.setLedgerManagerFactoryClass(TestLedgerManagerFactory.class); + baseClientConf.setLedgerManagerFactoryClass(TestLedgerManagerFactory.class); + baseClientConf.setReadEntryTimeout(60000); + baseClientConf.setAddEntryTimeout(60000); + this.digestType = DigestType.CRC32; + } + + @Test(timeout = 60000) + public void testRecoverBeforeWriteMetadata1() throws Exception { + rereadDuringRecovery(true, 1, false, false); + } + + @Test(timeout = 60000) + public void testRecoverBeforeWriteMetadata2() throws Exception { + rereadDuringRecovery(true, 3, false, false); + } + + @Test(timeout = 60000) + public void testRecoverBeforeWriteMetadata3() throws Exception { + rereadDuringRecovery(false, 1, false, false); + } + + @Test(timeout = 60000) + public void testRecoverBeforeWriteMetadata4() throws Exception { + rereadDuringRecovery(false, 3, false, false); + } + + @Test(timeout = 60000) + public void testRereadDuringRecovery1() throws Exception { + rereadDuringRecovery(true, 1, true, false); + } + + @Test(timeout = 60000) + public void testRereadDuringRecovery2() throws Exception { + rereadDuringRecovery(true, 3, true, false); + } + + @Test(timeout = 60000) + public void testRereadDuringRecovery3() throws Exception { + rereadDuringRecovery(false, 1, true, false); + } + + @Test(timeout = 60000) + public void testRereadDuringRecovery4() throws Exception { + rereadDuringRecovery(false, 3, true, false); + } + + @Test(timeout = 60000) + public void testConcurrentRecovery1() throws Exception { + rereadDuringRecovery(true, 1, true, false); + } + + @Test(timeout = 60000) + public void testConcurrentRecovery2() throws Exception { + rereadDuringRecovery(true, 3, true, false); + } + + @Test(timeout = 60000) + public void testConcurrentRecovery3() throws Exception { + rereadDuringRecovery(false, 1, true, false); + } + + @Test(timeout = 60000) + public void testConcurrentRecovery4() throws Exception { + rereadDuringRecovery(false, 3, true, false); + } + + private void rereadDuringRecovery(boolean parallelRead, int batchSize, + boolean updateMetadata, boolean close) throws Exception { + ClientConfiguration newConf = new ClientConfiguration(); + newConf.addConfiguration(baseClientConf); + newConf.setEnableParallelRecoveryRead(parallelRead); + newConf.setRecoveryReadBatchSize(batchSize); + BookKeeper newBk = new BookKeeper(newConf); + + TestLedgerManager tlm = (TestLedgerManager) newBk.getUnderlyingLedgerManager(); + + final LedgerHandle lh = newBk.createLedger(numBookies, 2, 2, digestType, "".getBytes()); + CountDownLatch latch1 = new CountDownLatch(1); + CountDownLatch latch2 = new CountDownLatch(1); + sleepBookie(lh.getLedgerMetadata().currentEnsemble.get(0), latch1); + sleepBookie(lh.getLedgerMetadata().currentEnsemble.get(1), latch2); + + int numEntries = (numBookies * 3) + 1; + final AtomicInteger numPendingAdds = new AtomicInteger(numEntries); + final CountDownLatch addDone = new CountDownLatch(1); + for (int i = 0; i < numEntries; i++) { + lh.asyncAddEntry(("" + i).getBytes(), new org.apache.bookkeeper.client.AsyncCallback.AddCallback() { + @Override + public void addComplete(int rc, LedgerHandle lh, long entryId, Object ctx) { + if (BKException.Code.OK != rc) { + addDone.countDown(); + return; + } + if (numPendingAdds.decrementAndGet() == 0) { + addDone.countDown(); + } + } + }, null); + } + latch1.countDown(); + latch2.countDown(); + addDone.await(10, TimeUnit.SECONDS); + assertEquals(0, numPendingAdds.get()); + + LOG.info("Added {} entries to ledger {}.", numEntries, lh.getId()); + + long ledgerLenth = lh.getLength(); + + LedgerHandle recoverLh = newBk.openLedgerNoRecovery(lh.getId(), digestType, "".getBytes()); + assertEquals(BookieProtocol.INVALID_ENTRY_ID, recoverLh.getLastAddPushed()); + assertEquals(BookieProtocol.INVALID_ENTRY_ID, recoverLh.getLastAddConfirmed()); + assertEquals(0, recoverLh.getLength()); + + LOG.info("OpenLedgerNoRecovery {}.", lh.getId()); + + final CountDownLatch metadataLatch = new CountDownLatch(1); + + tlm.setLatch(metadataLatch); + + final CountDownLatch recoverLatch = new CountDownLatch(1); + final AtomicBoolean success = new AtomicBoolean(false); + recoverLh.recover(new GenericCallback<Void>() { + @Override + public void operationComplete(int rc, Void result) { + LOG.info("Recovering ledger {} completed : {}.", lh.getId(), rc); + success.set(BKException.Code.OK == rc); + recoverLatch.countDown(); + } + }); + + // clear the metadata latch + tlm.setLatch(null); + + if (updateMetadata) { + if (close) { + LOG.info("OpenLedger {} to close.", lh.getId()); + LedgerHandle newRecoverLh = newBk.openLedger(lh.getId(), digestType, "".getBytes()); + newRecoverLh.close(); + } else { + LOG.info("OpenLedgerNoRecovery {} again.", lh.getId()); + LedgerHandle newRecoverLh = newBk.openLedgerNoRecovery(lh.getId(), digestType, "".getBytes()); + assertEquals(BookieProtocol.INVALID_ENTRY_ID, newRecoverLh.getLastAddPushed()); + assertEquals(BookieProtocol.INVALID_ENTRY_ID, newRecoverLh.getLastAddConfirmed()); + // mark the ledger as in recovery to update version. + newRecoverLh.getLedgerMetadata().markLedgerInRecovery(); + final CountDownLatch updateLatch = new CountDownLatch(1); + final AtomicInteger updateResult = new AtomicInteger(0x12345); + newRecoverLh.writeLedgerConfig(new GenericCallback<Void>() { + @Override + public void operationComplete(int rc, Void result) { + updateResult.set(rc); + updateLatch.countDown(); + } + }); + updateLatch.await(); + assertEquals(BKException.Code.OK, updateResult.get()); + newRecoverLh.close(); + LOG.info("Updated ledger manager {}.", newRecoverLh.getLedgerMetadata()); + } + } + + // resume metadata operation on recoverLh + metadataLatch.countDown(); + + LOG.info("Resume metadata update."); + + // wait until recover completed + recoverLatch.await(20, TimeUnit.SECONDS); + assertTrue(success.get()); + assertEquals(numEntries - 1, recoverLh.getLastAddPushed()); + assertEquals(numEntries - 1, recoverLh.getLastAddConfirmed()); + assertEquals(ledgerLenth, recoverLh.getLength()); + assertTrue(recoverLh.getLedgerMetadata().isClosed()); + + Enumeration<LedgerEntry> enumeration = recoverLh.readEntries(0, numEntries - 1); + int numReads = 0; + while (enumeration.hasMoreElements()) { + LedgerEntry entry = enumeration.nextElement(); + assertEquals((long) numReads, entry.getEntryId()); + assertEquals(numReads, Integer.parseInt(new String(entry.getEntry()))); + ++numReads; + } + assertEquals(numEntries, numReads); + + recoverLh.close(); + newBk.close(); + } + + @Test(timeout = 60000) + public void testRecoveryOnEntryGap() throws Exception { + byte[] passwd = "recovery-on-entry-gap".getBytes(UTF_8); + LedgerHandle lh = bkc.createLedger(1, 1, 1, DigestType.CRC32, passwd); + for (int i = 0; i < 10; i++) { + lh.addEntry(("recovery-on-entry-gap-" + i).getBytes(UTF_8)); + } + + // simulate ledger writer failure on concurrent writes causing gaps + + long entryId = 14; + long lac = 8; + byte[] data = "recovery-on-entry-gap-gap".getBytes(UTF_8); + ByteBuf toSend = + lh.macManager.computeDigestAndPackageForSending( + entryId, lac, lh.getLength() + 100, Unpooled.wrappedBuffer(data, 0, data.length)); + final CountDownLatch addLatch = new CountDownLatch(1); + final AtomicBoolean addSuccess = new AtomicBoolean(false); + LOG.info("Add entry {} with lac = {}", entryId, lac); + lh.bk.bookieClient.addEntry(lh.metadata.currentEnsemble.get(0), lh.getId(), lh.ledgerKey, entryId, toSend, + new WriteCallback() { + @Override + public void writeComplete(int rc, long ledgerId, long entryId, BookieSocketAddress addr, Object ctx) { + addSuccess.set(BKException.Code.OK == rc); + addLatch.countDown(); + } + }, 0, BookieProtocol.FLAG_NONE); + addLatch.await(); + assertTrue("add entry 14 should succeed", addSuccess.get()); + + ClientConfiguration newConf = new ClientConfiguration(); + newConf.addConfiguration(baseClientConf); + newConf.setEnableParallelRecoveryRead(true); + newConf.setRecoveryReadBatchSize(10); + + BookKeeper newBk = new BookKeeper(newConf); + + final LedgerHandle recoverLh = + newBk.openLedgerNoRecovery(lh.getId(), DigestType.CRC32, passwd); + + assertEquals("wrong lac found", 8L, recoverLh.getLastAddConfirmed()); + + final CountDownLatch recoverLatch = new CountDownLatch(1); + final AtomicLong newLac = new AtomicLong(-1); + final AtomicBoolean isMetadataClosed = new AtomicBoolean(false); + final AtomicInteger numSuccessCalls = new AtomicInteger(0); + final AtomicInteger numFailureCalls = new AtomicInteger(0); + recoverLh.recover(new GenericCallback<Void>() { + @Override + public void operationComplete(int rc, Void result) { + if (BKException.Code.OK == rc) { + newLac.set(recoverLh.getLastAddConfirmed()); + isMetadataClosed.set(recoverLh.getLedgerMetadata().isClosed()); + numSuccessCalls.incrementAndGet(); + } else { + numFailureCalls.incrementAndGet(); + } + recoverLatch.countDown(); + } + }); + recoverLatch.await(); + assertEquals("wrong lac found", 9L, newLac.get()); + assertTrue("metadata isn't closed after recovery", isMetadataClosed.get()); + Thread.sleep(5000); + assertEquals("recovery callback should be triggered only once", 1, numSuccessCalls.get()); + assertEquals("recovery callback should be triggered only once", 0, numFailureCalls.get()); + } + + static class DelayResponseBookie extends Bookie { + + static final class WriteCallbackEntry { + + private final WriteCallback cb; + private final int rc; + private final long ledgerId; + private final long entryId; + private final BookieSocketAddress addr; + private final Object ctx; + + WriteCallbackEntry(WriteCallback cb, + int rc, long ledgerId, long entryId, + BookieSocketAddress addr, Object ctx) { + this.cb = cb; + this.rc = rc; + this.ledgerId = ledgerId; + this.entryId = entryId; + this.addr = addr; + this.ctx = ctx; + } + + public void callback() { + cb.writeComplete(rc, ledgerId, entryId, addr, ctx); + } + } + + private final AtomicBoolean delayAddResponse = new AtomicBoolean(false); + private final AtomicBoolean delayReadResponse = new AtomicBoolean(false); + private final AtomicLong delayReadOnEntry = new AtomicLong(-1234L); + private volatile CountDownLatch delayReadLatch = null; + private final LinkedBlockingQueue<WriteCallbackEntry> delayQueue = + new LinkedBlockingQueue<WriteCallbackEntry>(); + + public DelayResponseBookie(ServerConfiguration conf) + throws IOException, KeeperException, InterruptedException, BookieException { + super(conf); + } + + @Override + public void addEntry(ByteBuf entry, final WriteCallback cb, Object ctx, byte[] masterKey) + throws IOException, BookieException { + super.addEntry(entry, new WriteCallback() { + @Override + public void writeComplete(int rc, long ledgerId, long entryId, + BookieSocketAddress addr, Object ctx) { + if (delayAddResponse.get()) { + delayQueue.add(new WriteCallbackEntry(cb, rc, ledgerId, entryId, addr, ctx)); + } else { + cb.writeComplete(rc, ledgerId, entryId, addr, ctx); + } + } + }, ctx, masterKey); + } + + @Override + public ByteBuf readEntry(long ledgerId, long entryId) throws IOException, NoLedgerException { + LOG.info("ReadEntry {} - {}", ledgerId, entryId); + if (delayReadResponse.get() && delayReadOnEntry.get() == entryId) { + CountDownLatch latch = delayReadLatch; + if (null != latch) { + try { + latch.await(); + } catch (InterruptedException e) { + // no-op + } + } + } + return super.readEntry(ledgerId, entryId); + } + + void delayAdd(boolean delayed) { + this.delayAddResponse.set(delayed); + } + + void delayRead(boolean delayed, long entryId, CountDownLatch delayReadLatch) { + this.delayReadResponse.set(delayed); + this.delayReadOnEntry.set(entryId); + this.delayReadLatch = delayReadLatch; + } + + } + + @Test(timeout = 60000) + public void testRecoveryWhenClosingLedgerHandle() throws Exception { + byte[] passwd = "recovery-when-closing-ledger-handle".getBytes(UTF_8); + + ClientConfiguration newConf = new ClientConfiguration(); + newConf.addConfiguration(baseClientConf); + newConf.setEnableParallelRecoveryRead(true); + newConf.setRecoveryReadBatchSize(1); + newConf.setAddEntryTimeout(9999999); + newConf.setReadEntryTimeout(9999999); + + final BookKeeper newBk0 = new BookKeeper(newConf); + final LedgerHandle lh0 = newBk0.createLedger(1, 1, 1, digestType, passwd); + + final BookKeeper newBk1 = new BookKeeper(newConf); + final LedgerHandle lh1 = newBk1.openLedgerNoRecovery(lh0.getId(), digestType, passwd); + final TestLedgerManager tlm1 = (TestLedgerManager) newBk1.getUnderlyingLedgerManager(); + + final BookKeeper readBk = new BookKeeper(newConf); + final LedgerHandle readLh = readBk.openLedgerNoRecovery(lh0.getId(), digestType, passwd); + + LOG.info("Create ledger {}", lh0.getId()); + + // 0) place the bookie with a fake bookie + BookieSocketAddress address = lh0.getLedgerMetadata().currentEnsemble.get(0); + ServerConfiguration conf = killBookie(address); + conf.setLedgerStorageClass(InterleavedLedgerStorage.class.getName()); + DelayResponseBookie fakeBookie = new DelayResponseBookie(conf); + bs.add(startBookie(conf, fakeBookie)); + bsConfs.add(conf); + + // 1) bk0 write two entries + lh0.addEntry("entry-0".getBytes(UTF_8)); + lh0.addEntry("entry-1".getBytes(UTF_8)); + + // 2) readBk read last add confirmed + long lac = readLh.readLastConfirmed(); + assertEquals(0L, lac); + lac = lh1.readLastConfirmed(); + assertEquals(0L, lac); + + final CountDownLatch addLatch = new CountDownLatch(3); + final AtomicInteger numAddFailures = new AtomicInteger(0); + // 3) bk0 write more entries in parallel + fakeBookie.delayAdd(true); + for (int i = 2; i < 5; i++) { + lh0.asyncAddEntry(("entry-" + i).getBytes(UTF_8), new AsyncCallback.AddCallback() { + @Override + public void addComplete(int rc, LedgerHandle lh, long entryId, Object ctx) { + if (BKException.Code.OK != rc) { + numAddFailures.incrementAndGet(); + } + addLatch.countDown(); + } + }, null); + } + while (fakeBookie.delayQueue.size() < 3) { + // wait until all add requests are queued + Thread.sleep(100); + } + + // 4) lac moved to 1L + lac = readLh.readLastConfirmed(); + assertEquals(1L, lac); + lac = lh1.readLastConfirmed(); + assertEquals(1L, lac); + + // 5) bk1 is doing recovery, but the metadata update is delayed + final CountDownLatch readLatch = new CountDownLatch(1); + fakeBookie.delayAdd(false); + fakeBookie.delayRead(true, 3L, readLatch); + final CountDownLatch metadataLatch = new CountDownLatch(1); + tlm1.setLatch(metadataLatch); + final CountDownLatch recoverLatch = new CountDownLatch(1); + final AtomicBoolean recoverSuccess = new AtomicBoolean(false); + lh1.recover(new GenericCallback<Void>() { + @Override + public void operationComplete(int rc, Void result) { + LOG.info("Recovering ledger {} completed : {}", lh1.getId(), rc); + recoverSuccess.set(BKException.Code.OK == rc); + recoverLatch.countDown(); + } + }); + Thread.sleep(2000); + readLatch.countDown(); + + // we don't expected lac being updated before we successfully marked the ledger in recovery + lac = readLh.readLastConfirmed(); + assertEquals(1L, lac); + + // 6) bk0 closes ledger before bk1 marks in recovery + lh0.close(); + assertEquals(1L, lh0.getLastAddConfirmed()); + + // 7) bk1 proceed recovery and succeed + metadataLatch.countDown(); + recoverLatch.await(); + assertTrue(recoverSuccess.get()); + assertEquals(1L, lh1.getLastAddConfirmed()); + + // 8) make sure we won't see lac advanced during ledger is closed by bk0 and recovered by bk1 + final AtomicLong lacHolder = new AtomicLong(-1234L); + final AtomicInteger rcHolder = new AtomicInteger(-1234); + final CountDownLatch doneLatch = new CountDownLatch(1); + new ReadLastConfirmedOp(readLh, new ReadLastConfirmedOp.LastConfirmedDataCallback() { + @Override + public void readLastConfirmedDataComplete(int rc, DigestManager.RecoveryData data) { + rcHolder.set(rc); + lacHolder.set(data.lastAddConfirmed); + doneLatch.countDown(); + } + }).initiate(); + doneLatch.await(); + assertEquals(BKException.Code.OK, rcHolder.get()); + assertEquals(1L, lacHolder.get()); + + newBk0.close(); + newBk1.close(); + readBk.close(); + } + +}
