| /* |
| * |
| * 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 java.nio.charset.StandardCharsets.UTF_8; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertTrue; |
| import io.netty.buffer.ByteBuf; |
| import io.netty.buffer.Unpooled; |
| import java.io.IOException; |
| import java.util.Enumeration; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.LinkedBlockingQueue; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicLong; |
| import org.apache.bookkeeper.bookie.BookieException; |
| import org.apache.bookkeeper.bookie.BookieImpl; |
| import org.apache.bookkeeper.bookie.InterleavedLedgerStorage; |
| import org.apache.bookkeeper.client.BookKeeper.DigestType; |
| import org.apache.bookkeeper.client.api.LedgerMetadata; |
| import org.apache.bookkeeper.client.api.WriteFlag; |
| import org.apache.bookkeeper.conf.ClientConfiguration; |
| import org.apache.bookkeeper.conf.ServerConfiguration; |
| import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory; |
| import org.apache.bookkeeper.meta.LedgerManager; |
| import org.apache.bookkeeper.meta.LedgerManagerFactory; |
| import org.apache.bookkeeper.meta.MetadataDrivers; |
| import org.apache.bookkeeper.meta.exceptions.Code; |
| import org.apache.bookkeeper.meta.exceptions.MetadataException; |
| import org.apache.bookkeeper.meta.zk.ZKMetadataBookieDriver; |
| import org.apache.bookkeeper.meta.zk.ZKMetadataClientDriver; |
| import org.apache.bookkeeper.net.BookieId; |
| 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.proto.checksum.DigestManager; |
| import org.apache.bookkeeper.test.BookKeeperClusterTestCase; |
| import org.apache.bookkeeper.util.ByteBufList; |
| import org.apache.bookkeeper.versioning.Version; |
| import org.apache.bookkeeper.versioning.Versioned; |
| import org.apache.commons.lang3.mutable.MutableInt; |
| import org.apache.zookeeper.AsyncCallback.VoidCallback; |
| import org.apache.zookeeper.KeeperException; |
| import org.junit.After; |
| import org.junit.Test; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * Test parallel ledger recovery. |
| */ |
| 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 CompletableFuture<Versioned<LedgerMetadata>> createLedgerMetadata( |
| long ledgerId, LedgerMetadata metadata) { |
| return lm.createLedgerMetadata(ledgerId, metadata); |
| } |
| |
| @Override |
| public CompletableFuture<Void> removeLedgerMetadata(long ledgerId, Version version) { |
| return lm.removeLedgerMetadata(ledgerId, version); |
| } |
| |
| @Override |
| public CompletableFuture<Versioned<LedgerMetadata>> readLedgerMetadata(long ledgerId) { |
| return lm.readLedgerMetadata(ledgerId); |
| } |
| |
| @Override |
| public LedgerRangeIterator getLedgerRanges(long zkOpTimeoutMs) { |
| return lm.getLedgerRanges(zkOpTimeoutMs); |
| } |
| |
| @Override |
| public CompletableFuture<Versioned<LedgerMetadata>> writeLedgerMetadata(long ledgerId, LedgerMetadata metadata, |
| Version currentVersion) { |
| final CountDownLatch cdl = waitLatch; |
| if (null != cdl) { |
| CompletableFuture<Versioned<LedgerMetadata>> promise = new CompletableFuture<>(); |
| executorService.submit(new Runnable() { |
| @Override |
| public void run() { |
| try { |
| cdl.await(); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| LOG.error("Interrupted on waiting latch : ", e); |
| } |
| lm.writeLedgerMetadata(ledgerId, metadata, currentVersion) |
| .whenComplete((metadata, exception) -> { |
| if (exception != null) { |
| promise.completeExceptionally(exception); |
| } else { |
| promise.complete(metadata); |
| } |
| }); |
| } |
| }); |
| return promise; |
| } else { |
| return lm.writeLedgerMetadata(ledgerId, metadata, currentVersion); |
| } |
| } |
| |
| @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()); |
| } |
| } |
| |
| static class TestMetadataClientDriver extends ZKMetadataClientDriver { |
| |
| @Override |
| public synchronized LedgerManagerFactory getLedgerManagerFactory() throws MetadataException { |
| if (null == lmFactory) { |
| try { |
| lmFactory = new TestLedgerManagerFactory() |
| .initialize(conf, layoutManager, TestLedgerManagerFactory.CUR_VERSION); |
| } catch (IOException e) { |
| throw new MetadataException(Code.METADATA_SERVICE_ERROR, e); |
| } |
| } |
| return lmFactory; |
| } |
| } |
| |
| static class TestMetadataBookieDriver extends ZKMetadataBookieDriver { |
| |
| @Override |
| public synchronized LedgerManagerFactory getLedgerManagerFactory() throws MetadataException { |
| if (null == lmFactory) { |
| try { |
| lmFactory = new TestLedgerManagerFactory() |
| .initialize(conf, layoutManager, TestLedgerManagerFactory.CUR_VERSION); |
| } catch (IOException e) { |
| throw new MetadataException(Code.METADATA_SERVICE_ERROR, e); |
| } |
| } |
| return lmFactory; |
| } |
| } |
| |
| final DigestType digestType; |
| |
| public ParallelLedgerRecoveryTest() throws Exception { |
| super(3); |
| |
| this.digestType = DigestType.CRC32; |
| } |
| |
| @Override |
| protected void startBKCluster(String metadataServiceUri) throws Exception { |
| MetadataDrivers.registerClientDriver("zk", TestMetadataClientDriver.class, true); |
| MetadataDrivers.registerBookieDriver("zk", TestMetadataBookieDriver.class, true); |
| baseConf.setLedgerManagerFactoryClass(TestLedgerManagerFactory.class); |
| baseClientConf.setLedgerManagerFactoryClass(TestLedgerManagerFactory.class); |
| baseClientConf.setReadEntryTimeout(60000); |
| baseClientConf.setAddEntryTimeout(60000); |
| |
| super.startBKCluster(metadataServiceUri); |
| } |
| |
| @After |
| @Override |
| public void tearDown() throws Exception { |
| try { |
| super.tearDown(); |
| } finally { |
| MetadataDrivers.registerClientDriver("zk", ZKMetadataClientDriver.class, true); |
| MetadataDrivers.registerBookieDriver("zk", ZKMetadataBookieDriver.class, true); |
| } |
| } |
| |
| @Test |
| public void testRecoverBeforeWriteMetadata1() throws Exception { |
| rereadDuringRecovery(true, 1, false, false); |
| } |
| |
| @Test |
| public void testRecoverBeforeWriteMetadata2() throws Exception { |
| rereadDuringRecovery(true, 3, false, false); |
| } |
| |
| @Test |
| public void testRecoverBeforeWriteMetadata3() throws Exception { |
| rereadDuringRecovery(false, 1, false, false); |
| } |
| |
| @Test |
| public void testRecoverBeforeWriteMetadata4() throws Exception { |
| rereadDuringRecovery(false, 3, false, false); |
| } |
| |
| @Test |
| public void testRereadDuringRecovery1() throws Exception { |
| rereadDuringRecovery(true, 1, true, false); |
| } |
| |
| @Test |
| public void testRereadDuringRecovery2() throws Exception { |
| rereadDuringRecovery(true, 3, true, false); |
| } |
| |
| @Test |
| public void testRereadDuringRecovery3() throws Exception { |
| rereadDuringRecovery(false, 1, true, false); |
| } |
| |
| @Test |
| public void testRereadDuringRecovery4() throws Exception { |
| rereadDuringRecovery(false, 3, true, false); |
| } |
| |
| @Test |
| public void testConcurrentRecovery1() throws Exception { |
| rereadDuringRecovery(true, 1, true, false); |
| } |
| |
| @Test |
| public void testConcurrentRecovery2() throws Exception { |
| rereadDuringRecovery(true, 3, true, false); |
| } |
| |
| @Test |
| public void testConcurrentRecovery3() throws Exception { |
| rereadDuringRecovery(false, 1, true, false); |
| } |
| |
| @Test |
| 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.getCurrentEnsemble().get(0), latch1); |
| sleepBookie(lh.getCurrentEnsemble().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); |
| ((ReadOnlyLedgerHandle) 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. |
| ClientUtil.transformMetadata(newBk.getClientCtx(), newRecoverLh.getId(), |
| (metadata) -> LedgerMetadataBuilder.from(metadata).withInRecoveryState().build()); |
| |
| 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 |
| 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); |
| ByteBufList 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); |
| |
| bkc.getBookieClient().addEntry(lh.getCurrentEnsemble().get(0), |
| lh.getId(), lh.ledgerKey, entryId, toSend, |
| new WriteCallback() { |
| @Override |
| public void writeComplete(int rc, long ledgerId, long entryId, |
| BookieId addr, Object ctx) { |
| addSuccess.set(BKException.Code.OK == rc); |
| addLatch.countDown(); |
| } |
| }, 0, BookieProtocol.FLAG_NONE, false, WriteFlag.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); |
| ((ReadOnlyLedgerHandle) 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 BookieImpl { |
| |
| static final class WriteCallbackEntry { |
| |
| private final WriteCallback cb; |
| private final int rc; |
| private final long ledgerId; |
| private final long entryId; |
| private final BookieId addr; |
| private final Object ctx; |
| |
| WriteCallbackEntry(WriteCallback cb, |
| int rc, long ledgerId, long entryId, |
| BookieId 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, boolean ackBeforeSync, final WriteCallback cb, Object ctx, byte[] masterKey) |
| throws IOException, BookieException, InterruptedException { |
| super.addEntry(entry, ackBeforeSync, new WriteCallback() { |
| @Override |
| public void writeComplete(int rc, long ledgerId, long entryId, |
| BookieId 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) { |
| Thread.currentThread().interrupt(); |
| // 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 |
| 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 |
| BookieId address = lh0.getCurrentEnsemble().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); |
| ((ReadOnlyLedgerHandle) 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(bkc.getBookieClient(), |
| readLh.distributionSchedule, |
| readLh.macManager, |
| readLh.ledgerId, |
| readLh.getLedgerMetadata().getAllEnsembles().lastEntry().getValue(), |
| readLh.ledgerKey, |
| new ReadLastConfirmedOp.LastConfirmedDataCallback() { |
| @Override |
| public void readLastConfirmedDataComplete(int rc, DigestManager.RecoveryData data) { |
| rcHolder.set(rc); |
| lacHolder.set(data.getLastAddConfirmed()); |
| doneLatch.countDown(); |
| } |
| }).initiate(); |
| doneLatch.await(); |
| assertEquals(BKException.Code.OK, rcHolder.get()); |
| assertEquals(1L, lacHolder.get()); |
| |
| newBk0.close(); |
| newBk1.close(); |
| readBk.close(); |
| } |
| |
| /** |
| * Validate ledger can recover with response: (Qw - Qa)+1. |
| * @throws Exception |
| */ |
| @Test |
| public void testRecoveryWithUnavailableBookie() throws Exception { |
| |
| byte[] passwd = "".getBytes(UTF_8); |
| ClientConfiguration newConf = new ClientConfiguration(); |
| newConf.addConfiguration(baseClientConf); |
| final BookKeeper readBk = new BookKeeper(newConf); |
| final BookKeeper newBk0 = new BookKeeper(newConf); |
| |
| /** |
| * Test Group-1 : Expected Response for recovery: Qr = (Qw - Qa)+1 = (3 |
| * -2) + 1 = 2 |
| */ |
| int ensembleSize = 3; |
| int writeQuorumSize = 3; |
| int ackQuormSize = 2; |
| LedgerHandle lh0 = newBk0.createLedger(ensembleSize, writeQuorumSize, ackQuormSize, DigestType.DUMMY, passwd); |
| LedgerHandle readLh = readBk.openLedgerNoRecovery(lh0.getId(), DigestType.DUMMY, passwd); |
| // Test 1: bookie response: OK, NO_SUCH_LEDGER_EXISTS, NOT_AVAILABLE |
| // Expected: Recovery successful Q(response) = 2 |
| int responseCode = readLACFromQuorum(readLh, BKException.Code.BookieHandleNotAvailableException, |
| BKException.Code.OK, BKException.Code.NoSuchLedgerExistsException); |
| assertEquals(responseCode, BKException.Code.OK); |
| // Test 2: bookie response: OK, NOT_AVAILABLE, NOT_AVAILABLE |
| // Expected: Recovery fail Q(response) = 1 |
| responseCode = readLACFromQuorum(readLh, BKException.Code.BookieHandleNotAvailableException, |
| BKException.Code.OK, BKException.Code.BookieHandleNotAvailableException); |
| assertEquals(responseCode, BKException.Code.BookieHandleNotAvailableException); |
| |
| /** |
| * Test Group-2 : Expected Response for recovery: Qr = (Qw - Qa)+1 = (2 |
| * -2) + 1 = 1 |
| */ |
| ensembleSize = 2; |
| writeQuorumSize = 2; |
| ackQuormSize = 2; |
| lh0 = newBk0.createLedger(ensembleSize, writeQuorumSize, ackQuormSize, DigestType.DUMMY, passwd); |
| readLh = readBk.openLedgerNoRecovery(lh0.getId(), DigestType.DUMMY, passwd); |
| // Test 1: bookie response: OK, NOT_AVAILABLE |
| // Expected: Recovery successful Q(response) = 1 |
| responseCode = readLACFromQuorum(readLh, BKException.Code.BookieHandleNotAvailableException, |
| BKException.Code.OK); |
| assertEquals(responseCode, BKException.Code.OK); |
| |
| // Test 1: bookie response: OK, NO_SUCH_LEDGER_EXISTS |
| // Expected: Recovery successful Q(response) = 2 |
| responseCode = readLACFromQuorum(readLh, BKException.Code.NoSuchLedgerExistsException, BKException.Code.OK); |
| assertEquals(responseCode, BKException.Code.OK); |
| |
| // Test 3: bookie response: NOT_AVAILABLE, NOT_AVAILABLE |
| // Expected: Recovery fail Q(response) = 0 |
| responseCode = readLACFromQuorum(readLh, BKException.Code.BookieHandleNotAvailableException, |
| BKException.Code.BookieHandleNotAvailableException); |
| assertEquals(responseCode, BKException.Code.BookieHandleNotAvailableException); |
| |
| newBk0.close(); |
| readBk.close(); |
| } |
| |
| private int readLACFromQuorum(LedgerHandle ledger, int... bookieLACResponse) throws Exception { |
| MutableInt responseCode = new MutableInt(100); |
| CountDownLatch responseLatch = new CountDownLatch(1); |
| ReadLastConfirmedOp readLCOp = new ReadLastConfirmedOp( |
| bkc.getBookieClient(), |
| ledger.getDistributionSchedule(), |
| ledger.getDigestManager(), |
| ledger.getId(), |
| ledger.getLedgerMetadata().getAllEnsembles().lastEntry().getValue(), |
| ledger.getLedgerKey(), |
| new ReadLastConfirmedOp.LastConfirmedDataCallback() { |
| @Override |
| public void readLastConfirmedDataComplete(int rc, DigestManager.RecoveryData data) { |
| System.out.println("response = " + rc); |
| responseCode.setValue(rc); |
| responseLatch.countDown(); |
| } |
| }); |
| byte[] lac = new byte[Long.SIZE * 3]; |
| ByteBuf data = Unpooled.wrappedBuffer(lac, 0, lac.length); |
| int writerIndex = data.writerIndex(); |
| data.resetWriterIndex(); |
| data.writeLong(ledger.getId()); |
| data.writeLong(0L); |
| data.writerIndex(writerIndex); |
| for (int i = 0; i < bookieLACResponse.length; i++) { |
| readLCOp.readEntryComplete(bookieLACResponse[i], 0, 0, data, i); |
| } |
| responseLatch.await(); |
| return responseCode.intValue(); |
| } |
| } |
