content: /*
   * 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.replication;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertTrue;
  import static org.junit.Assert.fail;
  
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.List;
  import java.util.Map;
  import java.util.Random;
  import java.util.Set;
  import org.apache.bookkeeper.client.BKException;
  import org.apache.bookkeeper.client.BookKeeper.DigestType;
  import org.apache.bookkeeper.client.LedgerHandle;
  import org.apache.bookkeeper.meta.AbstractZkLedgerManagerFactory;
  import org.apache.bookkeeper.meta.LedgerManager;
  import org.apache.bookkeeper.meta.LedgerManagerFactory;
  import org.apache.bookkeeper.meta.ZkLayoutManager;
  import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
  import org.apache.bookkeeper.replication.ReplicationException.BKAuditException;
  import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
  import org.apache.bookkeeper.util.ZkUtils;
  import org.apache.zookeeper.KeeperException;
  import org.junit.After;
  import org.junit.Before;
  import org.junit.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Tests verifies bookie vs ledger mapping generating by the BookieLedgerIndexer.
   */
  public class BookieLedgerIndexTest extends BookKeeperClusterTestCase {
  
      // Depending on the taste, select the amount of logging
      // by decommenting one of the two lines below
      // private final static Logger LOG = Logger.getRootLogger();
      private static final Logger LOG = LoggerFactory
              .getLogger(BookieLedgerIndexTest.class);
  
      private Random rng; // Random Number Generator
      private ArrayList<byte[]> entries; // generated entries
      private final DigestType digestType = DigestType.CRC32;
      private int numberOfLedgers = 3;
      private List<Long> ledgerList;
      private LedgerManagerFactory newLedgerManagerFactory;
      private LedgerManager ledgerManager;
  
      public BookieLedgerIndexTest()
          throws IOException, KeeperException, InterruptedException {
          this("org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory");
      }
  
      BookieLedgerIndexTest(String ledgerManagerFactory)
              throws IOException, KeeperException, InterruptedException {
          super(3);
          LOG.info("Running test case using ledger manager : "
                  + ledgerManagerFactory);
          // set ledger manager name
          baseConf.setLedgerManagerFactoryClassName(ledgerManagerFactory);
          baseClientConf.setLedgerManagerFactoryClassName(ledgerManagerFactory);
      }
  
      @Before
      public void setUp() throws Exception {
          super.setUp();
          baseConf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
          rng = new Random(System.currentTimeMillis()); // Initialize the Random
          // Number Generator
          entries = new ArrayList<byte[]>(); // initialize the entries list
          ledgerList = new ArrayList<Long>(3);
          // initialize ledger manager
          newLedgerManagerFactory = AbstractZkLedgerManagerFactory.newLedgerManagerFactory(
              baseConf,
              new ZkLayoutManager(zkc,
                  ZKMetadataDriverBase.resolveZkLedgersRootPath(baseConf), ZkUtils.getACLs(baseConf)));
  
          ledgerManager = newLedgerManagerFactory.newLedgerManager();
      }
  
      @After
      public void tearDown() throws Exception {
          super.tearDown();
          if (null != newLedgerManagerFactory) {
              newLedgerManagerFactory.close();
              newLedgerManagerFactory = null;
          }
          if (null != ledgerManager) {
              ledgerManager.close();
              ledgerManager = null;
          }
      }
  
      /**
       * Verify the bookie-ledger mapping with minimum number of bookies and few
       * ledgers.
       */
      @Test
      public void testSimpleBookieLedgerMapping() throws Exception {
  
          for (int i = 0; i < numberOfLedgers; i++) {
              createAndAddEntriesToLedger().close();
          }
  
          BookieLedgerIndexer bookieLedgerIndex = new BookieLedgerIndexer(
                  ledgerManager);
  
          Map<String, Set<Long>> bookieToLedgerIndex = bookieLedgerIndex
                  .getBookieToLedgerIndex();
  
          assertEquals("Missed few bookies in the bookie-ledger mapping!", 3,
                  bookieToLedgerIndex.size());
          Collection<Set<Long>> bk2ledgerEntry = bookieToLedgerIndex.values();
          for (Set<Long> ledgers : bk2ledgerEntry) {
              assertEquals("Missed few ledgers in the bookie-ledger mapping!", 3,
                      ledgers.size());
              for (Long ledgerId : ledgers) {
                  assertTrue("Unknown ledger-bookie mapping", ledgerList
                          .contains(ledgerId));
              }
          }
      }
  
      /**
       * Verify ledger index with failed bookies and throws exception.
       */
      @SuppressWarnings("deprecation")
      @Test
      public void testWithoutZookeeper() throws Exception {
          // This test case is for ledger metadata that stored in ZooKeeper. As
          // far as MSLedgerManagerFactory, ledger metadata are stored in other
          // storage. So this test is not suitable for MSLedgerManagerFactory.
          if (newLedgerManagerFactory instanceof org.apache.bookkeeper.meta.MSLedgerManagerFactory) {
              return;
          }
  
          for (int i = 0; i < numberOfLedgers; i++) {
              createAndAddEntriesToLedger().close();
          }
  
          BookieLedgerIndexer bookieLedgerIndex = new BookieLedgerIndexer(
                  ledgerManager);
          stopZKCluster();
          try {
              bookieLedgerIndex.getBookieToLedgerIndex();
              fail("Must throw exception as zookeeper are not running!");
          } catch (BKAuditException bkAuditException) {
              // expected behaviour
          }
      }
  
      /**
       * Verify indexing with multiple ensemble reformation.
       */
      @Test
      public void testEnsembleReformation() throws Exception {
          try {
              LedgerHandle lh1 = createAndAddEntriesToLedger();
              LedgerHandle lh2 = createAndAddEntriesToLedger();
  
              startNewBookie();
              shutdownBookie(lastBookieIndex() - 1);
  
              // add few more entries after ensemble reformation
              for (int i = 0; i < 10; i++) {
                  ByteBuffer entry = ByteBuffer.allocate(4);
                  entry.putInt(rng.nextInt(Integer.MAX_VALUE));
                  entry.position(0);
  
                  entries.add(entry.array());
                  lh1.addEntry(entry.array());
                  lh2.addEntry(entry.array());
              }
  
              BookieLedgerIndexer bookieLedgerIndex = new BookieLedgerIndexer(
                      ledgerManager);
  
              Map<String, Set<Long>> bookieToLedgerIndex = bookieLedgerIndex
                      .getBookieToLedgerIndex();
              assertEquals("Missed few bookies in the bookie-ledger mapping!", 4,
                      bookieToLedgerIndex.size());
              Collection<Set<Long>> bk2ledgerEntry = bookieToLedgerIndex.values();
              for (Set<Long> ledgers : bk2ledgerEntry) {
                  assertEquals(
                          "Missed few ledgers in the bookie-ledger mapping!", 2,
                          ledgers.size());
                  for (Long ledgerNode : ledgers) {
                      assertTrue("Unknown ledger-bookie mapping", ledgerList
                              .contains(ledgerNode));
                  }
              }
          } catch (BKException e) {
              LOG.error("Test failed", e);
              fail("Test failed due to BookKeeper exception");
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
              LOG.error("Test failed", e);
              fail("Test failed due to interruption");
          }
      }
  
      private void shutdownBookie(int bkShutdownIndex) throws Exception {
          killBookie(bkShutdownIndex);
      }
  
      private LedgerHandle createAndAddEntriesToLedger() throws BKException,
              InterruptedException {
          int numEntriesToWrite = 20;
          // Create a ledger
          LedgerHandle lh = bkc.createLedger(digestType, "admin".getBytes());
          LOG.info("Ledger ID: " + lh.getId());
          for (int i = 0; i < numEntriesToWrite; i++) {
              ByteBuffer entry = ByteBuffer.allocate(4);
              entry.putInt(rng.nextInt(Integer.MAX_VALUE));
              entry.position(0);
  
              entries.add(entry.array());
              lh.addEntry(entry.array());
          }
          ledgerList.add(lh.getId());
          return lh;
      }
  }
  