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.bookie;
  
  import static org.junit.Assert.assertArrayEquals;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertNotNull;
  import static org.junit.Assert.assertTrue;
  
  import java.util.Enumeration;
  import org.apache.bookkeeper.client.BookKeeper.DigestType;
  import org.apache.bookkeeper.client.LedgerEntry;
  import org.apache.bookkeeper.client.LedgerHandle;
  import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
  import org.junit.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * This class tests that index corruption cases.
   */
  public class IndexCorruptionTest extends BookKeeperClusterTestCase {
      private static final Logger LOG = LoggerFactory.getLogger(IndexCorruptionTest.class);
  
      DigestType digestType;
  
      int pageSize = 1024;
  
      public IndexCorruptionTest() {
          super(1);
          this.digestType = DigestType.CRC32;
          baseConf.setPageSize(pageSize);
      }
  
      @Test
      public void testNoSuchLedger() throws Exception {
          if (LOG.isDebugEnabled()) {
              LOG.debug("Testing NoSuchLedger");
          }
  
          SyncThread syncThread = ((BookieImpl) serverByIndex(0).getBookie()).syncThread;
          syncThread.suspendSync();
          // Create a ledger
          LedgerHandle lh = bkc.createLedger(1, 1, digestType, "".getBytes());
  
          // Close the ledger which cause a readEntry(0) call
          LedgerHandle newLh = bkc.openLedger(lh.getId(), digestType, "".getBytes());
  
          // Create a new ledger to write entries
          String dummyMsg = "NoSuchLedger";
          int numMsgs = 3;
          LedgerHandle wlh = bkc.createLedger(1, 1, digestType, "".getBytes());
          for (int i = 0; i < numMsgs; i++) {
              wlh.addEntry(dummyMsg.getBytes());
          }
  
          syncThread.resumeSync();
  
          // trigger sync
          Thread.sleep(2 * baseConf.getFlushInterval());
  
          // restart bookies
          restartBookies();
  
          Enumeration<LedgerEntry> seq = wlh.readEntries(0, numMsgs - 1);
          assertTrue("Enumeration of ledger entries has no element", seq.hasMoreElements());
          int entryId = 0;
          while (seq.hasMoreElements()) {
              LedgerEntry e = seq.nextElement();
              assertEquals(entryId, e.getEntryId());
  
              assertArrayEquals(dummyMsg.getBytes(), e.getEntry());
              ++entryId;
          }
          assertEquals(entryId, numMsgs);
      }
  
      @Test
      public void testEmptyIndexPage() throws Exception {
          if (LOG.isDebugEnabled()) {
              LOG.debug("Testing EmptyIndexPage");
          }
  
          SyncThread syncThread = ((BookieImpl) serverByIndex(0).getBookie()).syncThread;
          assertNotNull("Not found SyncThread.", syncThread);
  
          syncThread.suspendSync();
  
          // Create a ledger
          LedgerHandle lh1 = bkc.createLedger(1, 1, digestType, "".getBytes());
  
          String dummyMsg = "NoSuchLedger";
  
          // write two page entries to ledger 2
          int numMsgs = 2 * pageSize / 8;
          LedgerHandle lh2 = bkc.createLedger(1, 1, digestType, "".getBytes());
          for (int i = 0; i < numMsgs; i++) {
              lh2.addEntry(dummyMsg.getBytes());
          }
  
          syncThread.resumeSync();
  
          // trigger sync
          Thread.sleep(2 * baseConf.getFlushInterval());
  
          syncThread.suspendSync();
  
          // Close ledger 1 which cause a readEntry(0) call
          LedgerHandle newLh1 = bkc.openLedger(lh1.getId(), digestType, "".getBytes());
  
          // write another 3 entries to ledger 2
          for (int i = 0; i < 3; i++) {
              lh2.addEntry(dummyMsg.getBytes());
          }
  
          syncThread.resumeSync();
  
          // wait for sync again
          Thread.sleep(2 * baseConf.getFlushInterval());
  
          // restart bookies
          restartBookies();
  
          numMsgs += 3;
          Enumeration<LedgerEntry> seq = lh2.readEntries(0, numMsgs - 1);
          assertTrue("Enumeration of ledger entries has no element", seq.hasMoreElements());
          int entryId = 0;
          while (seq.hasMoreElements()) {
              LedgerEntry e = seq.nextElement();
              assertEquals(entryId, e.getEntryId());
  
              assertArrayEquals(dummyMsg.getBytes(), e.getEntry());
              ++entryId;
          }
          assertEquals(entryId, numMsgs);
      }
  }
  