| /** |
| * |
| * 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 org.apache.bookkeeper.client.BookKeeperClientStats.ADD_OP; |
| import static org.apache.bookkeeper.client.BookKeeperClientStats.ADD_OP_UR; |
| import static org.apache.bookkeeper.client.BookKeeperClientStats.CLIENT_SCOPE; |
| import static org.apache.bookkeeper.client.BookKeeperClientStats.READ_OP_DM; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| import com.google.common.collect.Lists; |
| import io.netty.buffer.AbstractByteBufAllocator; |
| import io.netty.buffer.ByteBuf; |
| import io.netty.buffer.PooledByteBufAllocator; |
| import io.netty.buffer.Unpooled; |
| import io.netty.buffer.UnpooledByteBufAllocator; |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Enumeration; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.UUID; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.TimeUnit; |
| import java.util.stream.Collectors; |
| import org.apache.bookkeeper.bookie.BookieException; |
| import org.apache.bookkeeper.bookie.BookieImpl; |
| import org.apache.bookkeeper.client.AsyncCallback.AddCallback; |
| import org.apache.bookkeeper.client.BKException.BKLedgerClosedException; |
| import org.apache.bookkeeper.client.BookKeeper.DigestType; |
| import org.apache.bookkeeper.client.api.LedgerEntries; |
| import org.apache.bookkeeper.client.api.ReadHandle; |
| import org.apache.bookkeeper.client.api.WriteAdvHandle; |
| import org.apache.bookkeeper.conf.ServerConfiguration; |
| import org.apache.bookkeeper.meta.LongHierarchicalLedgerManagerFactory; |
| import org.apache.bookkeeper.net.BookieId; |
| import org.apache.bookkeeper.test.BookKeeperClusterTestCase; |
| import org.apache.commons.lang3.tuple.Pair; |
| import org.apache.zookeeper.KeeperException; |
| import org.junit.Assert; |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.powermock.reflect.Whitebox; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * Testing ledger write entry cases. |
| */ |
| public class BookieWriteLedgerTest extends |
| BookKeeperClusterTestCase implements AddCallback { |
| |
| private static final Logger LOG = LoggerFactory |
| .getLogger(BookieWriteLedgerTest.class); |
| |
| byte[] ledgerPassword = "aaa".getBytes(); |
| LedgerHandle lh, lh2; |
| Enumeration<LedgerEntry> ls; |
| |
| // test related variables |
| int numEntriesToWrite = 100; |
| int maxInt = Integer.MAX_VALUE; |
| Random rng; // Random Number Generator |
| ArrayList<byte[]> entries1; // generated entries |
| ArrayList<byte[]> entries2; // generated entries |
| |
| private final DigestType digestType; |
| |
| private static class SyncObj { |
| volatile int counter; |
| volatile int rc; |
| |
| public SyncObj() { |
| counter = 0; |
| } |
| } |
| |
| @Override |
| @Before |
| public void setUp() throws Exception { |
| super.setUp(); |
| rng = new Random(0); // Initialize the Random |
| // Number Generator |
| entries1 = new ArrayList<byte[]>(); // initialize the entries list |
| entries2 = new ArrayList<byte[]>(); // initialize the entries list |
| } |
| |
| public BookieWriteLedgerTest() { |
| super(5, 180); |
| this.digestType = DigestType.CRC32; |
| String ledgerManagerFactory = "org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory"; |
| // set ledger manager |
| baseConf.setLedgerManagerFactoryClassName(ledgerManagerFactory); |
| /* |
| * 'testLedgerCreateAdvWithLedgerIdInLoop2' testcase relies on skipListSizeLimit, |
| * so setting it to some small value for making that testcase lite. |
| */ |
| baseConf.setSkipListSizeLimit(4 * 1024 * 1024); |
| baseClientConf.setLedgerManagerFactoryClassName(ledgerManagerFactory); |
| } |
| |
| /** |
| * Verify write when few bookie failures in last ensemble and forcing |
| * ensemble reformation. |
| */ |
| @Test |
| public void testWithMultipleBookieFailuresInLastEnsemble() throws Exception { |
| // Create a ledger |
| lh = bkc.createLedger(5, 4, digestType, ledgerPassword); |
| LOG.info("Ledger ID: " + lh.getId()); |
| for (int i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| // Start three more bookies |
| startNewBookie(); |
| startNewBookie(); |
| startNewBookie(); |
| |
| // Shutdown three bookies in the last ensemble and continue writing |
| List<BookieId> ensemble = lh.getLedgerMetadata() |
| .getAllEnsembles().entrySet().iterator().next().getValue(); |
| killBookie(ensemble.get(0)); |
| killBookie(ensemble.get(1)); |
| killBookie(ensemble.get(2)); |
| |
| int i = numEntriesToWrite; |
| numEntriesToWrite = numEntriesToWrite + 50; |
| for (; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| readEntries(lh, entries1); |
| lh.close(); |
| } |
| |
| /** |
| * Verify write and Read durability stats. |
| */ |
| @Test |
| public void testWriteAndReadStats() throws Exception { |
| // Create a ledger |
| lh = bkc.createLedger(3, 3, 2, digestType, ledgerPassword); |
| |
| // write-batch-1 |
| for (int i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| assertTrue( |
| "Stats should have captured a new writes", |
| bkc.getTestStatsProvider().getOpStatsLogger( |
| CLIENT_SCOPE + "." + ADD_OP) |
| .getSuccessCount() > 0); |
| |
| CountDownLatch sleepLatch1 = new CountDownLatch(1); |
| CountDownLatch sleepLatch2 = new CountDownLatch(1); |
| List<BookieId> ensemble = lh.getLedgerMetadata() |
| .getAllEnsembles().entrySet().iterator().next().getValue(); |
| |
| sleepBookie(ensemble.get(0), sleepLatch1); |
| |
| int i = numEntriesToWrite; |
| numEntriesToWrite = numEntriesToWrite + 50; |
| |
| // write-batch-2 |
| |
| for (; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| |
| // Let the second bookie go to sleep. This forces write timeout and ensemble change |
| // Which will be enough time to receive delayed write failures on the write-batch-2 |
| |
| sleepBookie(ensemble.get(1), sleepLatch2); |
| i = numEntriesToWrite; |
| numEntriesToWrite = numEntriesToWrite + 50; |
| |
| // write-batch-3 |
| |
| for (; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| |
| assertTrue( |
| "Stats should have captured a new UnderReplication during write", |
| bkc.getTestStatsProvider().getCounter( |
| CLIENT_SCOPE + "." + ADD_OP_UR) |
| .get() > 0); |
| |
| sleepLatch1.countDown(); |
| sleepLatch2.countDown(); |
| |
| // Replace the bookie with a fake bookie |
| ServerConfiguration conf = killBookie(ensemble.get(0)); |
| BookieWriteLedgerTest.CorruptReadBookie corruptBookie = new BookieWriteLedgerTest.CorruptReadBookie(conf); |
| bs.add(startBookie(conf, corruptBookie)); |
| bsConfs.add(conf); |
| |
| i = numEntriesToWrite; |
| numEntriesToWrite = numEntriesToWrite + 50; |
| |
| // write-batch-4 |
| |
| for (; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| |
| readEntries(lh, entries1); |
| assertTrue( |
| "Stats should have captured DigestMismatch on Read", |
| bkc.getTestStatsProvider().getCounter( |
| CLIENT_SCOPE + "." + READ_OP_DM) |
| .get() > 0); |
| lh.close(); |
| } |
| /** |
| * Verty delayedWriteError causes ensemble changes. |
| */ |
| @Test |
| public void testDelayedWriteEnsembleChange() throws Exception { |
| // Create a ledger |
| lh = bkc.createLedger(3, 3, 2, digestType, ledgerPassword); |
| baseClientConf.setAddEntryTimeout(1); |
| |
| int numEntriesToWrite = 10; |
| // write-batch-1 |
| for (int i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| |
| CountDownLatch sleepLatch1 = new CountDownLatch(1); |
| |
| // get bookie at index-0 |
| BookieId bookie1 = lh.getCurrentEnsemble().get(0); |
| sleepBookie(bookie1, sleepLatch1); |
| |
| int i = numEntriesToWrite; |
| numEntriesToWrite = numEntriesToWrite + 10; |
| |
| // write-batch-2 |
| |
| for (; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| // Sleep to receive delayed error on the write directed to the sleeping bookie |
| Thread.sleep(baseClientConf.getAddEntryTimeout() * 1000 * 2); |
| assertTrue( |
| "Stats should have captured a new UnderReplication during write", |
| bkc.getTestStatsProvider().getCounter( |
| CLIENT_SCOPE + "." + ADD_OP_UR) |
| .get() > 0); |
| |
| i = numEntriesToWrite; |
| numEntriesToWrite = numEntriesToWrite + 10; |
| |
| // write-batch-3 |
| for (; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| |
| sleepLatch1.countDown(); |
| // get the bookie at index-0 again, this must be different. |
| BookieId bookie2 = lh.getCurrentEnsemble().get(0); |
| |
| assertFalse( |
| "Delayed write error must have forced ensemble change", |
| bookie1.equals(bookie2)); |
| lh.close(); |
| } |
| /** |
| * Verify the functionality Ledgers with different digests. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerDigestTest() throws Exception { |
| for (DigestType type: DigestType.values()) { |
| lh = bkc.createLedger(5, 3, 2, type, ledgerPassword); |
| |
| for (int i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(entry.array()); |
| } |
| |
| readEntries(lh, entries1); |
| |
| long lid = lh.getId(); |
| lh.close(); |
| bkc.deleteLedger(lid); |
| entries1.clear(); |
| } |
| } |
| |
| /** |
| * Verify the functionality of Advanced Ledger which returns |
| * LedgerHandleAdv. LedgerHandleAdv takes entryId for addEntry, and let |
| * user manage entryId allocation. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerCreateAdv() throws Exception { |
| // Create a ledger |
| lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| for (int i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(i, entry.array()); |
| } |
| // Start one more bookies |
| startNewBookie(); |
| |
| // Shutdown one bookie in the last ensemble and continue writing |
| List<BookieId> ensemble = lh.getLedgerMetadata().getAllEnsembles().entrySet().iterator().next() |
| .getValue(); |
| killBookie(ensemble.get(0)); |
| |
| int i = numEntriesToWrite; |
| numEntriesToWrite = numEntriesToWrite + 50; |
| for (; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(i, entry.array()); |
| } |
| |
| readEntries(lh, entries1); |
| lh.close(); |
| } |
| |
| /** |
| * Verify that attempts to use addEntry() variant that does not require specifying entry id |
| * on LedgerHandleAdv results in error. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerCreateAdvAndWriteNonAdv() throws Exception { |
| long ledgerId = 0xABCDEF; |
| lh = bkc.createLedgerAdv(ledgerId, 3, 3, 2, digestType, ledgerPassword, null); |
| |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| try { |
| lh.addEntry(entry.array()); |
| fail("expected IllegalOpException"); |
| } catch (BKException.BKIllegalOpException e) { |
| // pass, expected |
| } finally { |
| lh.close(); |
| bkc.deleteLedger(ledgerId); |
| } |
| } |
| |
| /** |
| * Verify that LedgerHandleAdv cannnot handle addEntry without the entryId. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testNoAddEntryLedgerCreateAdv() throws Exception { |
| |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| assertTrue(lh instanceof LedgerHandleAdv); |
| |
| try { |
| lh.addEntry(entry.array()); |
| fail("using LedgerHandleAdv addEntry without entryId is forbidden"); |
| } catch (BKException e) { |
| assertEquals(e.getCode(), BKException.Code.IllegalOpException); |
| } |
| |
| try { |
| lh.addEntry(entry.array(), 0, 4); |
| fail("using LedgerHandleAdv addEntry without entryId is forbidden"); |
| } catch (BKException e) { |
| assertEquals(e.getCode(), BKException.Code.IllegalOpException); |
| } |
| |
| try { |
| CompletableFuture<Object> done = new CompletableFuture<>(); |
| lh.asyncAddEntry(Unpooled.wrappedBuffer(entry.array()), |
| (int rc, LedgerHandle lh1, long entryId, Object ctx) -> { |
| SyncCallbackUtils.finish(rc, null, done); |
| }, null); |
| done.get(); |
| } catch (ExecutionException ee) { |
| assertTrue(ee.getCause() instanceof BKException); |
| BKException e = (BKException) ee.getCause(); |
| assertEquals(e.getCode(), BKException.Code.IllegalOpException); |
| } |
| |
| try { |
| CompletableFuture<Object> done = new CompletableFuture<>(); |
| lh.asyncAddEntry(entry.array(), |
| (int rc, LedgerHandle lh1, long entryId, Object ctx) -> { |
| SyncCallbackUtils.finish(rc, null, done); |
| }, null); |
| done.get(); |
| } catch (ExecutionException ee) { |
| assertTrue(ee.getCause() instanceof BKException); |
| BKException e = (BKException) ee.getCause(); |
| assertEquals(e.getCode(), BKException.Code.IllegalOpException); |
| } |
| |
| try { |
| CompletableFuture<Object> done = new CompletableFuture<>(); |
| lh.asyncAddEntry(entry.array(), 0, 4, |
| (int rc, LedgerHandle lh1, long entryId, Object ctx) -> { |
| SyncCallbackUtils.finish(rc, null, done); |
| }, null); |
| done.get(); |
| } catch (ExecutionException ee) { |
| assertTrue(ee.getCause() instanceof BKException); |
| BKException e = (BKException) ee.getCause(); |
| assertEquals(e.getCode(), BKException.Code.IllegalOpException); |
| } |
| lh.close(); |
| } |
| |
| /** |
| * Verify the functionality of Advanced Ledger which accepts ledgerId as input and returns |
| * LedgerHandleAdv. LedgerHandleAdv takes entryId for addEntry, and let |
| * user manage entryId allocation. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerCreateAdvWithLedgerId() throws Exception { |
| // Create a ledger |
| long ledgerId = 0xABCDEF; |
| lh = bkc.createLedgerAdv(ledgerId, 5, 3, 2, digestType, ledgerPassword, null); |
| for (int i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(i, entry.array()); |
| } |
| // Start one more bookies |
| startNewBookie(); |
| |
| // Shutdown one bookie in the last ensemble and continue writing |
| List<BookieId> ensemble = lh.getLedgerMetadata().getAllEnsembles().entrySet().iterator().next() |
| .getValue(); |
| killBookie(ensemble.get(0)); |
| |
| int i = numEntriesToWrite; |
| numEntriesToWrite = numEntriesToWrite + 50; |
| for (; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(i, entry.array()); |
| } |
| |
| readEntries(lh, entries1); |
| lh.close(); |
| bkc.deleteLedger(ledgerId); |
| } |
| |
| /** |
| * Verify the functionality of Ledger create which accepts customMetadata as input. |
| * Also verifies that the data written is read back properly. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerCreateWithCustomMetadata() throws Exception { |
| // Create a ledger |
| long ledgerId; |
| int maxLedgers = 10; |
| for (int i = 0; i < maxLedgers; i++) { |
| Map<String, byte[]> inputCustomMetadataMap = new HashMap<String, byte[]>(); |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| // each ledger has different number of key, value pairs. |
| for (int j = 0; j < i; j++) { |
| inputCustomMetadataMap.put("key" + j, UUID.randomUUID().toString().getBytes()); |
| } |
| |
| if (i < maxLedgers / 2) { |
| // 0 to 4 test with createLedger interface |
| lh = bkc.createLedger(5, 3, 2, digestType, ledgerPassword, inputCustomMetadataMap); |
| ledgerId = lh.getId(); |
| lh.addEntry(entry.array()); |
| } else { |
| // 5 to 9 test with createLedgerAdv interface |
| lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword, inputCustomMetadataMap); |
| ledgerId = lh.getId(); |
| lh.addEntry(0, entry.array()); |
| } |
| lh.close(); |
| |
| // now reopen the ledger; this should fetch all the metadata stored on zk |
| // and the customMetadata written and read should match |
| lh = bkc.openLedger(ledgerId, digestType, ledgerPassword); |
| Map<String, byte[]> outputCustomMetadataMap = lh.getCustomMetadata(); |
| assertTrue("Can't retrieve proper Custom Data", |
| areByteArrayValMapsEqual(inputCustomMetadataMap, outputCustomMetadataMap)); |
| lh.close(); |
| bkc.deleteLedger(ledgerId); |
| } |
| } |
| |
| /** |
| * Routine to compare two {@code Map<String, byte[]>}; Since the values in the map are {@code byte[]}, we can't use |
| * {@code Map.equals}. |
| * @param first |
| * The first map |
| * @param second |
| * The second map to compare with |
| * @return true if the 2 maps contain the exact set of {@code <K,V>} pairs. |
| */ |
| public static boolean areByteArrayValMapsEqual(Map<String, byte[]> first, Map<String, byte[]> second) { |
| if (first == null && second == null) { |
| return true; |
| } |
| |
| // above check confirms that both are not null; |
| // if one is null the other isn't; so they must |
| // be different |
| if (first == null || second == null) { |
| return false; |
| } |
| |
| if (first.size() != second.size()) { |
| return false; |
| } |
| for (Map.Entry<String, byte[]> entry : first.entrySet()) { |
| if (!Arrays.equals(entry.getValue(), second.get(entry.getKey()))) { |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| /* |
| * Verify the functionality of Advanced Ledger which accepts ledgerId as |
| * input and returns LedgerHandleAdv. LedgerHandleAdv takes entryId for |
| * addEntry, and let user manage entryId allocation. |
| * This testcase is mainly added for covering missing code coverage branches |
| * in LedgerHandleAdv |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerHandleAdvFunctionality() throws Exception { |
| // Create a ledger |
| long ledgerId = 0xABCDEF; |
| lh = bkc.createLedgerAdv(ledgerId, 5, 3, 2, digestType, ledgerPassword, null); |
| numEntriesToWrite = 3; |
| |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| entries1.add(entry.array()); |
| lh.addEntry(0, entry.array()); |
| |
| // here asyncAddEntry(final long entryId, final byte[] data, final |
| // AddCallback cb, final Object ctx) method is |
| // called which is not covered in any other testcase |
| entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| entries1.add(entry.array()); |
| CountDownLatch latch = new CountDownLatch(1); |
| final int[] returnedRC = new int[1]; |
| lh.asyncAddEntry(1, entry.array(), new AddCallback() { |
| @Override |
| public void addComplete(int rc, LedgerHandle lh, long entryId, Object ctx) { |
| CountDownLatch latch = (CountDownLatch) ctx; |
| returnedRC[0] = rc; |
| latch.countDown(); |
| } |
| }, latch); |
| latch.await(); |
| assertTrue("Returned code is expected to be OK", returnedRC[0] == BKException.Code.OK); |
| |
| // here addEntry is called with incorrect offset and length |
| entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| try { |
| lh.addEntry(2, entry.array(), -3, 9); |
| fail("AddEntry is called with negative offset and incorrect length," |
| + "so it is expected to throw RuntimeException/IndexOutOfBoundsException"); |
| } catch (RuntimeException exception) { |
| // expected RuntimeException/IndexOutOfBoundsException |
| } |
| |
| // here addEntry is called with corrected offset and length and it is |
| // supposed to succeed |
| entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| entries1.add(entry.array()); |
| lh.addEntry(2, entry.array()); |
| |
| // LedgerHandle is closed for write |
| lh.close(); |
| |
| // here addEntry is called even after the close of the LedgerHandle, so |
| // it is expected to throw exception |
| entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| entries1.add(entry.array()); |
| try { |
| lh.addEntry(3, entry.array()); |
| fail("AddEntry is called after the close of LedgerHandle," |
| + "so it is expected to throw BKLedgerClosedException"); |
| } catch (BKLedgerClosedException exception) { |
| } |
| |
| readEntries(lh, entries1); |
| bkc.deleteLedger(ledgerId); |
| } |
| |
| /** |
| * In a loop create/write/delete the ledger with same ledgerId through |
| * the functionality of Advanced Ledger which accepts ledgerId as input. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerCreateAdvWithLedgerIdInLoop() throws Exception { |
| int ledgerCount = 40; |
| |
| long maxId = 9999999999L; |
| if (baseConf.getLedgerManagerFactoryClass().equals(LongHierarchicalLedgerManagerFactory.class)) { |
| // since LongHierarchicalLedgerManager supports ledgerIds of decimal length upto 19 digits but other |
| // LedgerManagers only upto 10 decimals |
| maxId = Long.MAX_VALUE; |
| } |
| |
| rng.longs(ledgerCount, 0, maxId) // generate a stream of ledger ids |
| .mapToObj(ledgerId -> { // create a ledger for each ledger id |
| LOG.info("Creating adv ledger with id {}", ledgerId); |
| return bkc.newCreateLedgerOp() |
| .withEnsembleSize(1).withWriteQuorumSize(1).withAckQuorumSize(1) |
| .withDigestType(org.apache.bookkeeper.client.api.DigestType.CRC32) |
| .withPassword(ledgerPassword).makeAdv().withLedgerId(ledgerId) |
| .execute() |
| .thenApply(writer -> { // Add entries to ledger when created |
| LOG.info("Writing stream of {} entries to {}", |
| numEntriesToWrite, ledgerId); |
| List<ByteBuf> entries = rng.ints(numEntriesToWrite, 0, maxInt) |
| .mapToObj(i -> { |
| ByteBuf entry = Unpooled.buffer(4); |
| entry.retain(); |
| entry.writeInt(i); |
| return entry; |
| }) |
| .collect(Collectors.toList()); |
| CompletableFuture<?> lastRequest = null; |
| int i = 0; |
| for (ByteBuf entry : entries) { |
| long entryId = i++; |
| LOG.info("Writing {}:{} as {}", |
| ledgerId, entryId, entry.slice().readInt()); |
| lastRequest = writer.writeAsync(entryId, entry); |
| } |
| lastRequest.join(); |
| return Pair.of(writer, entries); |
| }); |
| }) |
| .parallel().map(CompletableFuture::join) // wait for all creations and adds in parallel |
| .forEach(e -> { // check that each set of adds succeeded |
| try { |
| WriteAdvHandle handle = e.getLeft(); |
| List<ByteBuf> entries = e.getRight(); |
| // Read and verify |
| LOG.info("Read entries for ledger: {}", handle.getId()); |
| readEntries(handle, entries); |
| entries.forEach(ByteBuf::release); |
| handle.close(); |
| bkc.deleteLedger(handle.getId()); |
| } catch (InterruptedException ie) { |
| Thread.currentThread().interrupt(); |
| Assert.fail("Test interrupted"); |
| } catch (Exception ex) { |
| LOG.info("Readback failed with exception", ex); |
| Assert.fail("Readback failed " + ex.getMessage()); |
| } |
| }); |
| } |
| |
| /** |
| * In a loop create/write/read/delete the ledger with ledgerId through the |
| * functionality of Advanced Ledger which accepts ledgerId as input. |
| * In this testcase (other testcases don't cover these conditions, hence new |
| * testcase is added), we create entries which are greater than |
| * SKIP_LIST_MAX_ALLOC_ENTRY size and tried to addEntries so that the total |
| * length of data written in this testcase is much greater than |
| * SKIP_LIST_SIZE_LIMIT, so that entries will be flushed from EntryMemTable |
| * to persistent storage |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerCreateAdvWithLedgerIdInLoop2() throws Exception { |
| |
| assertTrue("Here we are expecting Bookies are configured to use SortedLedgerStorage", |
| baseConf.getSortedLedgerStorageEnabled()); |
| |
| long ledgerId; |
| int ledgerCount = 10; |
| |
| List<List<byte[]>> entryList = new ArrayList<List<byte[]>>(); |
| LedgerHandle[] lhArray = new LedgerHandle[ledgerCount]; |
| long skipListSizeLimit = baseConf.getSkipListSizeLimit(); |
| int skipListArenaMaxAllocSize = baseConf.getSkipListArenaMaxAllocSize(); |
| |
| List<byte[]> tmpEntry; |
| for (int lc = 0; lc < ledgerCount; lc++) { |
| tmpEntry = new ArrayList<byte[]>(); |
| |
| ledgerId = rng.nextLong(); |
| ledgerId &= Long.MAX_VALUE; |
| if (!baseConf.getLedgerManagerFactoryClass().equals(LongHierarchicalLedgerManagerFactory.class)) { |
| // since LongHierarchicalLedgerManager supports ledgerIds of |
| // decimal length upto 19 digits but other |
| // LedgerManagers only upto 10 decimals |
| ledgerId %= 9999999999L; |
| } |
| |
| LOG.debug("Iteration: {} LedgerId: {}", lc, ledgerId); |
| lh = bkc.createLedgerAdv(ledgerId, 5, 3, 2, digestType, ledgerPassword, null); |
| lhArray[lc] = lh; |
| |
| long ledgerLength = 0; |
| int i = 0; |
| while (ledgerLength < ((4 * skipListSizeLimit) / ledgerCount)) { |
| int length; |
| if (rng.nextBoolean()) { |
| length = Math.abs(rng.nextInt()) % (skipListArenaMaxAllocSize); |
| } else { |
| // here we want length to be random no. in the range of skipListArenaMaxAllocSize and |
| // 4*skipListArenaMaxAllocSize |
| length = (Math.abs(rng.nextInt()) % (skipListArenaMaxAllocSize * 3)) + skipListArenaMaxAllocSize; |
| } |
| byte[] data = new byte[length]; |
| rng.nextBytes(data); |
| tmpEntry.add(data); |
| lh.addEntry(i, data); |
| ledgerLength += length; |
| i++; |
| } |
| entryList.add(tmpEntry); |
| } |
| for (int lc = 0; lc < ledgerCount; lc++) { |
| // Read and verify |
| long lid = lhArray[lc].getId(); |
| LOG.debug("readEntries for lc: {} ledgerId: {} ", lc, lhArray[lc].getId()); |
| readEntriesAndValidateDataArray(lhArray[lc], entryList.get(lc)); |
| lhArray[lc].close(); |
| bkc.deleteLedger(lid); |
| } |
| } |
| |
| /** |
| * Verify asynchronous writing when few bookie failures in last ensemble. |
| */ |
| @Test |
| public void testAsyncWritesWithMultipleFailuresInLastEnsemble() |
| throws Exception { |
| // Create ledgers |
| lh = bkc.createLedger(5, 4, digestType, ledgerPassword); |
| lh2 = bkc.createLedger(5, 4, digestType, ledgerPassword); |
| |
| LOG.info("Ledger ID-1: " + lh.getId()); |
| LOG.info("Ledger ID-2: " + lh2.getId()); |
| for (int i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| entries2.add(entry.array()); |
| lh.addEntry(entry.array()); |
| lh2.addEntry(entry.array()); |
| } |
| // Start three more bookies |
| startNewBookie(); |
| startNewBookie(); |
| startNewBookie(); |
| |
| // Shutdown three bookies in the last ensemble and continue writing |
| List<BookieId> ensemble = lh.getLedgerMetadata() |
| .getAllEnsembles().entrySet().iterator().next().getValue(); |
| killBookie(ensemble.get(0)); |
| killBookie(ensemble.get(1)); |
| killBookie(ensemble.get(2)); |
| |
| // adding one more entry to both the ledgers async after multiple bookie |
| // failures. This will do asynchronously modifying the ledger metadata |
| // simultaneously. |
| numEntriesToWrite++; |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| entries1.add(entry.array()); |
| entries2.add(entry.array()); |
| |
| SyncObj syncObj1 = new SyncObj(); |
| SyncObj syncObj2 = new SyncObj(); |
| lh.asyncAddEntry(entry.array(), this, syncObj1); |
| lh2.asyncAddEntry(entry.array(), this, syncObj2); |
| |
| // wait for all entries to be acknowledged for the first ledger |
| synchronized (syncObj1) { |
| while (syncObj1.counter < 1) { |
| LOG.debug("Entries counter = " + syncObj1.counter); |
| syncObj1.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj1.rc); |
| } |
| // wait for all entries to be acknowledged for the second ledger |
| synchronized (syncObj2) { |
| while (syncObj2.counter < 1) { |
| LOG.debug("Entries counter = " + syncObj2.counter); |
| syncObj2.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj2.rc); |
| } |
| |
| // reading ledger till the last entry |
| readEntries(lh, entries1); |
| readEntries(lh2, entries2); |
| lh.close(); |
| lh2.close(); |
| } |
| |
| /** |
| * Verify Advanced asynchronous writing with entryIds in reverse order. |
| */ |
| @Test |
| public void testLedgerCreateAdvWithAsyncWritesWithBookieFailures() throws Exception { |
| // Create ledgers |
| lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| lh2 = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| |
| LOG.info("Ledger ID-1: " + lh.getId()); |
| LOG.info("Ledger ID-2: " + lh2.getId()); |
| SyncObj syncObj1 = new SyncObj(); |
| SyncObj syncObj2 = new SyncObj(); |
| for (int i = numEntriesToWrite - 1; i >= 0; i--) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| try { |
| entries1.add(0, entry.array()); |
| entries2.add(0, entry.array()); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| lh.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj1); |
| lh2.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj2); |
| } |
| // Start One more bookie and shutdown one from last ensemble before reading |
| startNewBookie(); |
| List<BookieId> ensemble = lh.getLedgerMetadata().getAllEnsembles().entrySet().iterator().next() |
| .getValue(); |
| killBookie(ensemble.get(0)); |
| |
| // Wait for all entries to be acknowledged for the first ledger |
| synchronized (syncObj1) { |
| while (syncObj1.counter < numEntriesToWrite) { |
| syncObj1.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj1.rc); |
| } |
| // Wait for all entries to be acknowledged for the second ledger |
| synchronized (syncObj2) { |
| while (syncObj2.counter < numEntriesToWrite) { |
| syncObj2.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj2.rc); |
| } |
| |
| // Reading ledger till the last entry |
| readEntries(lh, entries1); |
| readEntries(lh2, entries2); |
| lh.close(); |
| lh2.close(); |
| } |
| |
| /** |
| * LedgerHandleAdv out of order writers with ensemble changes. |
| * Verify that entry that was written to old ensemble will be |
| * written to new enseble too after ensemble change. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerHandleAdvOutOfOrderWriteAndFrocedEnsembleChange() throws Exception { |
| // Create a ledger |
| long ledgerId = 0xABCDEF; |
| SyncObj syncObj1 = new SyncObj(); |
| ByteBuffer entry; |
| lh = bkc.createLedgerAdv(ledgerId, 3, 3, 3, digestType, ledgerPassword, null); |
| entry = ByteBuffer.allocate(4); |
| // Add entries 0-4 |
| for (int i = 0; i < 5; i++) { |
| entry.rewind(); |
| entry.putInt(rng.nextInt(maxInt)); |
| lh.addEntry(i, entry.array()); |
| } |
| |
| // Add 10 as Async Entry, which goes to first ensemble |
| ByteBuffer entry1 = ByteBuffer.allocate(4); |
| entry1.putInt(rng.nextInt(maxInt)); |
| lh.asyncAddEntry(10, entry1.array(), 0, entry1.capacity(), this, syncObj1); |
| |
| // Make sure entry-10 goes to the bookies and gets response. |
| java.util.Queue<PendingAddOp> myPendingAddOps = Whitebox.getInternalState(lh, "pendingAddOps"); |
| PendingAddOp addOp = null; |
| boolean pendingAddOpReceived = false; |
| |
| while (!pendingAddOpReceived) { |
| addOp = myPendingAddOps.peek(); |
| if (addOp.entryId == 10 && addOp.completed) { |
| pendingAddOpReceived = true; |
| } else { |
| Thread.sleep(1000); |
| } |
| } |
| |
| CountDownLatch sleepLatch1 = new CountDownLatch(1); |
| List<BookieId> ensemble; |
| |
| ensemble = lh.getLedgerMetadata().getAllEnsembles().entrySet().iterator().next().getValue(); |
| |
| // Put all 3 bookies to sleep and start 3 new ones |
| sleepBookie(ensemble.get(0), sleepLatch1); |
| sleepBookie(ensemble.get(1), sleepLatch1); |
| sleepBookie(ensemble.get(2), sleepLatch1); |
| startNewBookie(); |
| startNewBookie(); |
| startNewBookie(); |
| |
| // Original bookies are in sleep, new bookies added. |
| // Now add entries 5-9 which forces ensemble changes |
| // So at this point entries 0-4, 10 went to first |
| // ensemble, 5-9 will go to new ensemble. |
| for (int i = 5; i < 10; i++) { |
| entry.rewind(); |
| entry.putInt(rng.nextInt(maxInt)); |
| lh.addEntry(i, entry.array()); |
| } |
| |
| // Wakeup all 3 bookies that went to sleep |
| sleepLatch1.countDown(); |
| |
| // Wait for all entries to be acknowledged for the first ledger |
| synchronized (syncObj1) { |
| while (syncObj1.counter < 1) { |
| syncObj1.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj1.rc); |
| } |
| |
| // Close write handle |
| lh.close(); |
| |
| // Open read handle |
| lh = bkc.openLedger(ledgerId, digestType, ledgerPassword); |
| |
| // Make sure to read all 10 entries. |
| for (int i = 0; i < 11; i++) { |
| lh.readEntries(i, i); |
| } |
| lh.close(); |
| bkc.deleteLedger(ledgerId); |
| } |
| |
| /** |
| * Verify Advanced asynchronous writing with entryIds in pseudo random order with bookie failures between writes. |
| */ |
| @Test |
| public void testLedgerCreateAdvWithRandomAsyncWritesWithBookieFailuresBetweenWrites() throws Exception { |
| // Create ledgers |
| lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| lh2 = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| |
| LOG.info("Ledger ID-1: " + lh.getId()); |
| LOG.info("Ledger ID-2: " + lh2.getId()); |
| SyncObj syncObj1 = new SyncObj(); |
| SyncObj syncObj2 = new SyncObj(); |
| int batchSize = 5; |
| int i, j; |
| |
| // Fill the result buffers first |
| for (i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| try { |
| entries1.add(0, entry.array()); |
| entries2.add(0, entry.array()); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| for (i = 0; i < batchSize; i++) { |
| for (j = i; j < numEntriesToWrite; j = j + batchSize) { |
| byte[] entry1 = entries1.get(j); |
| byte[] entry2 = entries2.get(j); |
| lh.asyncAddEntry(j, entry1, 0, entry1.length, this, syncObj1); |
| lh2.asyncAddEntry(j, entry2, 0, entry2.length, this, syncObj2); |
| if (j == numEntriesToWrite / 2) { |
| // Start One more bookie and shutdown one from last ensemble at half-way |
| startNewBookie(); |
| List<BookieId> ensemble = lh.getLedgerMetadata().getAllEnsembles().entrySet() |
| .iterator().next().getValue(); |
| killBookie(ensemble.get(0)); |
| } |
| } |
| } |
| |
| // Wait for all entries to be acknowledged for the first ledger |
| synchronized (syncObj1) { |
| while (syncObj1.counter < numEntriesToWrite) { |
| syncObj1.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj1.rc); |
| } |
| // Wait for all entries to be acknowledged for the second ledger |
| synchronized (syncObj2) { |
| while (syncObj2.counter < numEntriesToWrite) { |
| syncObj2.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj2.rc); |
| } |
| |
| // Reading ledger till the last entry |
| readEntries(lh, entries1); |
| readEntries(lh2, entries2); |
| lh.close(); |
| lh2.close(); |
| } |
| |
| /** |
| * Verify Advanced asynchronous writing with entryIds in pseudo random order. |
| */ |
| @Test |
| public void testLedgerCreateAdvWithRandomAsyncWritesWithBookieFailures() throws Exception { |
| // Create ledgers |
| lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| lh2 = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| |
| LOG.info("Ledger ID-1: " + lh.getId()); |
| LOG.info("Ledger ID-2: " + lh2.getId()); |
| SyncObj syncObj1 = new SyncObj(); |
| SyncObj syncObj2 = new SyncObj(); |
| int batchSize = 5; |
| int i, j; |
| |
| // Fill the result buffers first |
| for (i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| try { |
| entries1.add(0, entry.array()); |
| entries2.add(0, entry.array()); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| for (i = 0; i < batchSize; i++) { |
| for (j = i; j < numEntriesToWrite; j = j + batchSize) { |
| byte[] entry1 = entries1.get(j); |
| byte[] entry2 = entries2.get(j); |
| lh.asyncAddEntry(j, entry1, 0, entry1.length, this, syncObj1); |
| lh2.asyncAddEntry(j, entry2, 0, entry2.length, this, syncObj2); |
| } |
| } |
| // Start One more bookie and shutdown one from last ensemble before reading |
| startNewBookie(); |
| List<BookieId> ensemble = lh.getLedgerMetadata().getAllEnsembles().entrySet().iterator().next() |
| .getValue(); |
| killBookie(ensemble.get(0)); |
| |
| // Wait for all entries to be acknowledged for the first ledger |
| synchronized (syncObj1) { |
| while (syncObj1.counter < numEntriesToWrite) { |
| syncObj1.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj1.rc); |
| } |
| // Wait for all entries to be acknowledged for the second ledger |
| synchronized (syncObj2) { |
| while (syncObj2.counter < numEntriesToWrite) { |
| syncObj2.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj2.rc); |
| } |
| |
| // Reading ledger till the last entry |
| readEntries(lh, entries1); |
| readEntries(lh2, entries2); |
| lh.close(); |
| lh2.close(); |
| } |
| |
| /** |
| * Skips few entries before closing the ledger and assert that the |
| * lastAddConfirmed is right before our skipEntryId. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerCreateAdvWithSkipEntries() throws Exception { |
| long ledgerId; |
| SyncObj syncObj1 = new SyncObj(); |
| |
| // Create a ledger |
| lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| // Save ledgerId to reopen the ledger |
| ledgerId = lh.getId(); |
| LOG.info("Ledger ID: " + ledgerId); |
| int skipEntryId = rng.nextInt(numEntriesToWrite - 1); |
| for (int i = numEntriesToWrite - 1; i >= 0; i--) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| try { |
| entries1.add(0, entry.array()); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| if (i == skipEntryId) { |
| LOG.info("Skipping entry:{}", skipEntryId); |
| continue; |
| } |
| lh.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj1); |
| } |
| // wait for all entries to be acknowledged for the first ledger |
| synchronized (syncObj1) { |
| while (syncObj1.counter < skipEntryId) { |
| syncObj1.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj1.rc); |
| } |
| // Close the ledger |
| lh.close(); |
| // Open the ledger |
| lh = bkc.openLedger(ledgerId, digestType, ledgerPassword); |
| assertEquals(lh.lastAddConfirmed, skipEntryId - 1); |
| lh.close(); |
| } |
| |
| /** |
| * Verify the functionality LedgerHandleAdv addEntry with duplicate entryIds. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerCreateAdvSyncAddDuplicateEntryIds() throws Exception { |
| // Create a ledger |
| lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| LOG.info("Ledger ID: " + lh.getId()); |
| for (int i = 0; i < numEntriesToWrite; i++) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| |
| entries1.add(entry.array()); |
| lh.addEntry(i, entry.array()); |
| entry.position(0); |
| } |
| readEntries(lh, entries1); |
| |
| int dupEntryId = rng.nextInt(numEntriesToWrite - 1); |
| |
| try { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| lh.addEntry(dupEntryId, entry.array()); |
| fail("Expected exception not thrown"); |
| } catch (BKException e) { |
| // This test expects DuplicateEntryIdException |
| assertEquals(e.getCode(), BKException.Code.DuplicateEntryIdException); |
| } |
| lh.close(); |
| } |
| |
| /** |
| * Verify the functionality LedgerHandleAdv asyncAddEntry with duplicate |
| * entryIds. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testLedgerCreateAdvSyncAsyncAddDuplicateEntryIds() throws Exception { |
| long ledgerId; |
| SyncObj syncObj1 = new SyncObj(); |
| SyncObj syncObj2 = new SyncObj(); |
| |
| // Create a ledger |
| lh = bkc.createLedgerAdv(5, 3, 2, digestType, ledgerPassword); |
| // Save ledgerId to reopen the ledger |
| ledgerId = lh.getId(); |
| LOG.info("Ledger ID: " + ledgerId); |
| for (int i = numEntriesToWrite - 1; i >= 0; i--) { |
| ByteBuffer entry = ByteBuffer.allocate(4); |
| entry.putInt(rng.nextInt(maxInt)); |
| entry.position(0); |
| try { |
| entries1.add(0, entry.array()); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| lh.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj1); |
| if (rng.nextBoolean()) { |
| // Attempt to write the same entry |
| lh.asyncAddEntry(i, entry.array(), 0, entry.capacity(), this, syncObj2); |
| synchronized (syncObj2) { |
| while (syncObj2.counter < 1) { |
| syncObj2.wait(); |
| } |
| assertEquals(BKException.Code.DuplicateEntryIdException, syncObj2.rc); |
| } |
| } |
| } |
| // Wait for all entries to be acknowledged for the first ledger |
| synchronized (syncObj1) { |
| while (syncObj1.counter < numEntriesToWrite) { |
| syncObj1.wait(); |
| } |
| assertEquals(BKException.Code.OK, syncObj1.rc); |
| } |
| // Close the ledger |
| lh.close(); |
| } |
| |
| @Test |
| @SuppressWarnings("unchecked") |
| public void testLedgerCreateAdvByteBufRefCnt() throws Exception { |
| long ledgerId = rng.nextLong(); |
| ledgerId &= Long.MAX_VALUE; |
| if (!baseConf.getLedgerManagerFactoryClass().equals(LongHierarchicalLedgerManagerFactory.class)) { |
| // since LongHierarchicalLedgerManager supports ledgerIds of |
| // decimal length upto 19 digits but other |
| // LedgerManagers only upto 10 decimals |
| ledgerId %= 9999999999L; |
| } |
| |
| final LedgerHandle lh = bkc.createLedgerAdv(ledgerId, 5, 3, 2, digestType, ledgerPassword, null); |
| |
| final List<AbstractByteBufAllocator> allocs = Lists.newArrayList( |
| new PooledByteBufAllocator(true), |
| new PooledByteBufAllocator(false), |
| new UnpooledByteBufAllocator(true), |
| new UnpooledByteBufAllocator(false)); |
| |
| long entryId = 0; |
| for (AbstractByteBufAllocator alloc: allocs) { |
| final ByteBuf data = alloc.buffer(10); |
| data.writeBytes(("fragment0" + entryId).getBytes()); |
| assertEquals("ref count on ByteBuf should be 1", 1, data.refCnt()); |
| |
| CompletableFuture<Integer> cf = new CompletableFuture<>(); |
| lh.asyncAddEntry(entryId, data, (rc, handle, eId, qwcLatency, ctx) -> { |
| CompletableFuture<Integer> future = (CompletableFuture<Integer>) ctx; |
| future.complete(rc); |
| }, cf); |
| |
| int rc = cf.get(); |
| assertEquals("rc code is OK", BKException.Code.OK, rc); |
| |
| for (int i = 0; i < 10; i++) { |
| if (data.refCnt() == 0) { |
| break; |
| } |
| TimeUnit.MILLISECONDS.sleep(250); // recycler runs asynchronously |
| } |
| assertEquals("writing entry with id " + entryId + ", ref count on ByteBuf should be 0 ", |
| 0, data.refCnt()); |
| |
| org.apache.bookkeeper.client.api.LedgerEntry e = lh.read(entryId, entryId).getEntry(entryId); |
| assertEquals("entry data is correct", "fragment0" + entryId, new String(e.getEntryBytes())); |
| entryId++; |
| } |
| |
| bkc.deleteLedger(lh.ledgerId); |
| } |
| |
| @Test |
| @SuppressWarnings("unchecked") |
| public void testLedgerCreateByteBufRefCnt() throws Exception { |
| final LedgerHandle lh = bkc.createLedger(5, 3, 2, digestType, ledgerPassword, null); |
| |
| final List<AbstractByteBufAllocator> allocs = Lists.newArrayList( |
| new PooledByteBufAllocator(true), |
| new PooledByteBufAllocator(false), |
| new UnpooledByteBufAllocator(true), |
| new UnpooledByteBufAllocator(false)); |
| |
| int entryId = 0; |
| for (AbstractByteBufAllocator alloc: allocs) { |
| final ByteBuf data = alloc.buffer(10); |
| data.writeBytes(("fragment0" + entryId).getBytes()); |
| assertEquals("ref count on ByteBuf should be 1", 1, data.refCnt()); |
| |
| CompletableFuture<Integer> cf = new CompletableFuture<>(); |
| lh.asyncAddEntry(data, (rc, handle, eId, ctx) -> { |
| CompletableFuture<Integer> future = (CompletableFuture<Integer>) ctx; |
| future.complete(rc); |
| }, cf); |
| |
| int rc = cf.get(); |
| assertEquals("rc code is OK", BKException.Code.OK, rc); |
| |
| for (int i = 0; i < 10; i++) { |
| if (data.refCnt() == 0) { |
| break; |
| } |
| TimeUnit.MILLISECONDS.sleep(250); // recycler runs asynchronously |
| } |
| assertEquals("writing entry with id " + entryId + ", ref count on ByteBuf should be 0 ", |
| 0, data.refCnt()); |
| |
| org.apache.bookkeeper.client.api.LedgerEntry e = lh.read(entryId, entryId).getEntry(entryId); |
| assertEquals("entry data is correct", "fragment0" + entryId, new String(e.getEntryBytes())); |
| entryId++; |
| } |
| |
| bkc.deleteLedger(lh.ledgerId); |
| } |
| |
| private void readEntries(LedgerHandle lh, List<byte[]> entries) throws InterruptedException, BKException { |
| ls = lh.readEntries(0, numEntriesToWrite - 1); |
| int index = 0; |
| while (ls.hasMoreElements()) { |
| ByteBuffer origbb = ByteBuffer.wrap(entries.get(index++)); |
| Integer origEntry = origbb.getInt(); |
| ByteBuffer result = ByteBuffer.wrap(ls.nextElement().getEntry()); |
| LOG.debug("Length of result: " + result.capacity()); |
| LOG.debug("Original entry: " + origEntry); |
| Integer retrEntry = result.getInt(); |
| LOG.debug("Retrieved entry: " + retrEntry); |
| assertTrue("Checking entry " + index + " for equality", origEntry |
| .equals(retrEntry)); |
| } |
| } |
| |
| private void readEntries(ReadHandle reader, List<ByteBuf> entries) throws Exception { |
| assertEquals("Not enough entries in ledger " + reader.getId(), |
| reader.getLastAddConfirmed(), entries.size() - 1); |
| try (LedgerEntries readEntries = reader.read(0, reader.getLastAddConfirmed())) { |
| int i = 0; |
| for (org.apache.bookkeeper.client.api.LedgerEntry e : readEntries) { |
| int entryId = i++; |
| ByteBuf origEntry = entries.get(entryId); |
| ByteBuf readEntry = e.getEntryBuffer(); |
| assertEquals("Unexpected contents in " + reader.getId() + ":" + entryId, origEntry, readEntry); |
| } |
| } |
| } |
| |
| private void readEntriesAndValidateDataArray(LedgerHandle lh, List<byte[]> entries) |
| throws InterruptedException, BKException { |
| ls = lh.readEntries(0, entries.size() - 1); |
| int index = 0; |
| while (ls.hasMoreElements()) { |
| byte[] originalData = entries.get(index++); |
| byte[] receivedData = ls.nextElement().getEntry(); |
| LOG.debug("Length of originalData: {}", originalData.length); |
| LOG.debug("Length of receivedData: {}", receivedData.length); |
| assertEquals( |
| String.format("LedgerID: %d EntryID: %d OriginalDataLength: %d ReceivedDataLength: %d", lh.getId(), |
| (index - 1), originalData.length, receivedData.length), |
| originalData.length, receivedData.length); |
| Assert.assertArrayEquals( |
| String.format("Checking LedgerID: %d EntryID: %d for equality", lh.getId(), (index - 1)), |
| originalData, receivedData); |
| } |
| } |
| |
| @Override |
| public void addComplete(int rc, LedgerHandle lh, long entryId, Object ctx) { |
| SyncObj x = (SyncObj) ctx; |
| synchronized (x) { |
| x.rc = rc; |
| x.counter++; |
| x.notify(); |
| } |
| } |
| |
| static class CorruptReadBookie extends BookieImpl { |
| |
| static final Logger LOG = LoggerFactory.getLogger(CorruptReadBookie.class); |
| ByteBuf localBuf; |
| |
| public CorruptReadBookie(ServerConfiguration conf) |
| throws IOException, KeeperException, InterruptedException, BookieException { |
| super(conf); |
| } |
| |
| @Override |
| public ByteBuf readEntry(long ledgerId, long entryId) throws IOException, NoLedgerException { |
| localBuf = super.readEntry(ledgerId, entryId); |
| |
| int capacity = 0; |
| while (capacity < localBuf.capacity()) { |
| localBuf.setByte(capacity, 0); |
| capacity++; |
| } |
| return localBuf; |
| } |
| |
| } |
| } |
