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.apache.bookkeeper.bookie.BookieException.Code.OK;
  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 io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  import io.netty.buffer.Unpooled;
  import java.io.File;
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicLong;
  import org.apache.bookkeeper.bookie.Bookie.NoLedgerException;
  import org.apache.bookkeeper.bookie.CheckpointSource.Checkpoint;
  import org.apache.bookkeeper.bookie.FileInfoBackingCache.CachedFileInfo;
  import org.apache.bookkeeper.conf.ServerConfiguration;
  import org.apache.bookkeeper.conf.TestBKConfiguration;
  import org.apache.bookkeeper.meta.LedgerManager;
  import org.apache.bookkeeper.net.BookieId;
  import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks;
  import org.apache.bookkeeper.stats.StatsLogger;
  import org.apache.bookkeeper.util.BookKeeperConstants;
  import org.apache.bookkeeper.util.IOUtils;
  import org.apache.bookkeeper.util.SnapshotMap;
  import org.apache.commons.io.FileUtils;
  import org.junit.After;
  import org.junit.Assert;
  import org.junit.Before;
  import org.junit.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * LedgerCache related test cases.
   */
  public class LedgerCacheTest {
      private static final Logger LOG = LoggerFactory.getLogger(LedgerCacheTest.class);
  
      SnapshotMap<Long, Boolean> activeLedgers;
      LedgerCache ledgerCache;
      Thread flushThread;
      ServerConfiguration conf;
      File txnDir, ledgerDir;
  
      private final List<File> tempDirs = new ArrayList<File>();
  
      private BookieImpl bookie;
  
      @Before
      public void setUp() throws Exception {
          txnDir = IOUtils.createTempDir("ledgercache", "txn");
          ledgerDir = IOUtils.createTempDir("ledgercache", "ledger");
          // create current dir
          new File(ledgerDir, BookKeeperConstants.CURRENT_DIR).mkdir();
  
          conf = TestBKConfiguration.newServerConfiguration();
          conf.setMetadataServiceUri(null);
          conf.setJournalDirName(txnDir.getPath());
          conf.setLedgerDirNames(new String[] { ledgerDir.getPath() });
          bookie = new TestBookieImpl(conf);
  
          activeLedgers = new SnapshotMap<Long, Boolean>();
          ledgerCache = ((InterleavedLedgerStorage) bookie.getLedgerStorage().getUnderlyingLedgerStorage()).ledgerCache;
      }
  
      @After
      public void tearDown() throws Exception {
          if (flushThread != null) {
              flushThread.interrupt();
              flushThread.join();
          }
          bookie.getLedgerStorage().shutdown();
          FileUtils.deleteDirectory(txnDir);
          FileUtils.deleteDirectory(ledgerDir);
          for (File dir : tempDirs) {
              FileUtils.deleteDirectory(dir);
          }
      }
  
      File createTempDir(String prefix, String suffix) throws IOException {
          File dir = IOUtils.createTempDir(prefix, suffix);
          tempDirs.add(dir);
          return dir;
      }
  
      private void newLedgerCache() throws IOException {
          if (ledgerCache != null) {
              ledgerCache.close();
          }
          ledgerCache = ((InterleavedLedgerStorage) bookie.getLedgerStorage().getUnderlyingLedgerStorage())
              .ledgerCache = new LedgerCacheImpl(conf, activeLedgers, bookie.getIndexDirsManager());
          flushThread = new Thread() {
                  public void run() {
                      while (true) {
                          try {
                              sleep(conf.getFlushInterval());
                              ledgerCache.flushLedger(true);
                          } catch (InterruptedException ie) {
                              // killed by teardown
                              Thread.currentThread().interrupt();
                              return;
                          } catch (Exception e) {
                              LOG.error("Exception in flush thread", e);
                          }
                      }
                  }
              };
          flushThread.start();
      }
  
      @Test
      public void testAddEntryException() throws IOException {
          // set page limitation
          conf.setPageLimit(10);
          // create a ledger cache
          newLedgerCache();
          /*
           * Populate ledger cache.
           */
          try {
              byte[] masterKey = "blah".getBytes();
              for (int i = 0; i < 100; i++) {
                  ledgerCache.setMasterKey((long) i, masterKey);
                  ledgerCache.putEntryOffset(i, 0, i * 8);
              }
          } catch (IOException e) {
              LOG.error("Got IOException.", e);
              fail("Failed to add entry.");
          }
      }
  
      @Test
      public void testLedgerEviction() throws Exception {
          int numEntries = 10;
          // limit open files & pages
          conf.setOpenFileLimit(1).setPageLimit(2)
              .setPageSize(8 * numEntries);
          // create ledger cache
          newLedgerCache();
          try {
              int numLedgers = 3;
              byte[] masterKey = "blah".getBytes();
              for (int i = 1; i <= numLedgers; i++) {
                  ledgerCache.setMasterKey((long) i, masterKey);
                  for (int j = 0; j < numEntries; j++) {
                      ledgerCache.putEntryOffset(i, j, i * numEntries + j);
                  }
              }
          } catch (Exception e) {
              LOG.error("Got Exception.", e);
              fail("Failed to add entry.");
          }
      }
  
      @Test
      public void testDeleteLedger() throws Exception {
          int numEntries = 10;
          // limit open files & pages
          conf.setOpenFileLimit(999).setPageLimit(2)
              .setPageSize(8 * numEntries);
          // create ledger cache
          newLedgerCache();
          try {
              int numLedgers = 2;
              byte[] masterKey = "blah".getBytes();
              for (int i = 1; i <= numLedgers; i++) {
                  ledgerCache.setMasterKey((long) i, masterKey);
                  for (int j = 0; j < numEntries; j++) {
                      ledgerCache.putEntryOffset(i, j, i * numEntries + j);
                  }
              }
              // ledger cache is exhausted
              // delete ledgers
              for (int i = 1; i <= numLedgers; i++) {
                  ledgerCache.deleteLedger((long) i);
              }
              // create num ledgers to add entries
              for (int i = numLedgers + 1; i <= 2 * numLedgers; i++) {
                  ledgerCache.setMasterKey((long) i, masterKey);
                  for (int j = 0; j < numEntries; j++) {
                      ledgerCache.putEntryOffset(i, j, i * numEntries + j);
                  }
              }
          } catch (Exception e) {
              LOG.error("Got Exception.", e);
              fail("Failed to add entry.");
          }
      }
  
      @Test
      public void testPageEviction() throws Exception {
          int numLedgers = 10;
          byte[] masterKey = "blah".getBytes();
          // limit page count
          conf.setOpenFileLimit(999999).setPageLimit(3);
          // create ledger cache
          newLedgerCache();
          try {
              // create several ledgers
              for (int i = 1; i <= numLedgers; i++) {
                  ledgerCache.setMasterKey((long) i, masterKey);
                  ledgerCache.putEntryOffset(i, 0, i * 8);
                  ledgerCache.putEntryOffset(i, 1, i * 8);
              }
  
              // flush all first to clean previous dirty ledgers
              ledgerCache.flushLedger(true);
              // flush all
              ledgerCache.flushLedger(true);
  
              // delete several ledgers
              for (int i = 1; i <= numLedgers / 2; i++) {
                  ledgerCache.deleteLedger(i);
              }
  
              // bookie restarts
              newLedgerCache();
  
              // simulate replaying journals to add entries again
              for (int i = 1; i <= numLedgers; i++) {
                  try {
                      ledgerCache.putEntryOffset(i, 1, i * 8);
                  } catch (NoLedgerException nsle) {
                      if (i <= numLedgers / 2) {
                          // it is ok
                      } else {
                          LOG.error("Error put entry offset : ", nsle);
                          fail("Should not reach here.");
                      }
                  }
              }
          } catch (Exception e) {
              LOG.error("Got Exception.", e);
              fail("Failed to add entry.");
          }
      }
  
      /**
       * Test Ledger Cache flush failure.
       */
      @Test
      public void testLedgerCacheFlushFailureOnDiskFull() throws Exception {
          File ledgerDir1 = createTempDir("bkTest", ".dir");
          File ledgerDir2 = createTempDir("bkTest", ".dir");
          ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
          conf.setLedgerDirNames(new String[] { ledgerDir1.getAbsolutePath(), ledgerDir2.getAbsolutePath() });
  
          BookieImpl bookie = new TestBookieImpl(conf);
          InterleavedLedgerStorage ledgerStorage =
              ((InterleavedLedgerStorage) bookie.getLedgerStorage().getUnderlyingLedgerStorage());
          LedgerCacheImpl ledgerCache = (LedgerCacheImpl) ledgerStorage.ledgerCache;
          // Create ledger index file
          ledgerStorage.setMasterKey(1, "key".getBytes());
  
          CachedFileInfo fileInfo = ledgerCache.getIndexPersistenceManager().getFileInfo(1L, null);
  
          // Add entries
          ledgerStorage.addEntry(generateEntry(1, 1));
          ledgerStorage.addEntry(generateEntry(1, 2));
          ledgerStorage.flush();
  
          ledgerStorage.addEntry(generateEntry(1, 3));
          // add the dir to failed dirs
          bookie.getIndexDirsManager().addToFilledDirs(
                  fileInfo.getLf().getParentFile().getParentFile().getParentFile());
          File before = fileInfo.getLf();
          // flush after disk is added as failed.
          ledgerStorage.flush();
          File after = fileInfo.getLf();
  
          assertFalse("After flush index file should be changed", before.equals(after));
          // Verify written entries
          Assert.assertEquals(generateEntry(1, 1), ledgerStorage.getEntry(1, 1));
          Assert.assertEquals(generateEntry(1, 2), ledgerStorage.getEntry(1, 2));
          Assert.assertEquals(generateEntry(1, 3), ledgerStorage.getEntry(1, 3));
      }
  
      /**
       * Test that if we are writing to more ledgers than there
       * are pages, then we will not flush the index before the
       * entries in the entrylogger have been persisted to disk.
       * {@link https://issues.apache.org/jira/browse/BOOKKEEPER-447}
       */
      @Test
      public void testIndexPageEvictionWriteOrder() throws Exception {
          final int numLedgers = 10;
          File journalDir = createTempDir("bookie", "journal");
          BookieImpl.checkDirectoryStructure(BookieImpl.getCurrentDirectory(journalDir));
  
          File ledgerDir = createTempDir("bookie", "ledger");
          BookieImpl.checkDirectoryStructure(BookieImpl.getCurrentDirectory(ledgerDir));
  
          ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
          conf.setMetadataServiceUri(null);
          conf.setJournalDirName(journalDir.getPath())
              .setLedgerDirNames(new String[] { ledgerDir.getPath() })
              .setFlushInterval(1000)
              .setPageLimit(1)
              .setLedgerStorageClass(InterleavedLedgerStorage.class.getName());
  
          Bookie b = new TestBookieImpl(conf);
          b.start();
          for (int i = 1; i <= numLedgers; i++) {
              ByteBuf packet = generateEntry(i, 1);
              b.addEntry(packet, false, new BookieImpl.NopWriteCallback(), null, "passwd".getBytes());
          }
  
          conf = TestBKConfiguration.newServerConfiguration();
          conf.setMetadataServiceUri(null);
          conf.setJournalDirName(journalDir.getPath())
              .setLedgerDirNames(new String[] { ledgerDir.getPath() });
  
          b = new TestBookieImpl(conf);
          for (int i = 1; i <= numLedgers; i++) {
              try {
                  b.readEntry(i, 1);
              } catch (Bookie.NoLedgerException nle) {
                  // this is fine, means the ledger was never written to the index cache
                  assertEquals("No ledger should only happen for the last ledger",
                               i, numLedgers);
              } catch (Bookie.NoEntryException nee) {
                  // this is fine, means the ledger was written to the index cache, but not
                  // the entry log
              } catch (IOException ioe) {
                  if (ioe.getCause() instanceof DefaultEntryLogger.EntryLookupException) {
                      // this is fine, means the ledger was not fully written to
                      // the entry log
                  } else {
                      LOG.info("Shouldn't have received IOException for entry {}", i, ioe);
                      fail("Shouldn't throw IOException, should say that entry is not found");
                  }
              }
          }
      }
  
  
      /**
       * {@link https://issues.apache.org/jira/browse/BOOKKEEPER-524}
       * Checks that getLedgerEntryPage does not throw an NPE in the
       * case getFromTable returns a null ledger entry page reference.
       * This NPE might kill the sync thread leaving a bookie with no
       * sync thread running.
       *
       * @throws IOException
       */
      @Test
      public void testSyncThreadNPE() throws IOException {
          newLedgerCache();
          try {
              ((LedgerCacheImpl) ledgerCache).getIndexPageManager().getLedgerEntryPageFromCache(0L, 0L, true);
          } catch (Exception e) {
              LOG.error("Exception when trying to get a ledger entry page", e);
              fail("Shouldn't have thrown an exception");
          }
      }
  
  
      /**
       * Test for race between putEntryOffset and flush.
       * {@link https://github.com/apache/bookkeeper/issues/1919}
       */
      @Test
      public void testPutEntryOffsetDeleteRace() throws Exception {
          newLedgerCache();
          final AtomicInteger rc = new AtomicInteger(0);
          final LinkedBlockingQueue<Long> putQ = new LinkedBlockingQueue<>(100);
          final LinkedBlockingQueue<Long> deleteQ = new LinkedBlockingQueue<>(100);
          final byte[] masterKey = "masterKey".getBytes();
          final long numLedgers = 1000;
          final int numPutters = 10;
          final int numDeleters = 10;
          final AtomicBoolean running = new AtomicBoolean(true);
          Thread newLedgerThread = new Thread() {
                  public void run() {
                      try {
                          for (long i = 0; i < numLedgers && rc.get() == 0; i++) {
                              ledgerCache.setMasterKey(i, masterKey);
  
                              ledgerCache.putEntryOffset(i, 1, 0);
                              deleteQ.put(i);
                              putQ.put(i);
                          }
                          for (int i = 0; i < numPutters; ++i) {
                              putQ.put(-1L);
                          }
                          for (int i = 0; i < numDeleters; ++i) {
                              deleteQ.put(-1L);
                          }
                      } catch (Throwable e) {
                          rc.set(-1);
                          LOG.error("Exception in new ledger thread", e);
                      }
                  }
              };
          newLedgerThread.start();
  
          Thread[] flushThreads = new Thread[numPutters];
          for (int i = 0; i < numPutters; ++i) {
              Thread flushThread = new Thread() {
                  public void run() {
                      try {
                          while (true) {
                              long id = putQ.take();
                              if (id == -1L) {
                                  break;
                              }
                              LOG.info("Putting {}", id);
                              try {
                                  ledgerCache.putEntryOffset(id, 2, 0);
                                  ledgerCache.deleteLedger(id);
                              } catch (NoLedgerException e) {
                                  // No problem
                              }
                          }
                      } catch (Throwable e) {
                          rc.set(-1);
                          LOG.error("Exception in put thread", e);
                      }
                  }
              };
              flushThread.start();
              flushThreads[i] = flushThread;
          }
  
          Thread[] deleteThreads = new Thread[numDeleters];
          for (int i = 0; i < numDeleters; ++i) {
              Thread deleteThread = new Thread() {
                  public void run() {
                      try {
                          while (true) {
                              long id = deleteQ.take();
                              if (id == -1L) {
                                  break;
                              }
                              LOG.info("Deleting {}", id);
                              try {
                                  ledgerCache.deleteLedger(id);
                              } catch (NoLedgerException e) {
                                  // No problem
                              }
                          }
                      } catch (Throwable e) {
                          rc.set(-1);
                          LOG.error("Exception in delete thread", e);
                      }
                  }
              };
              deleteThread.start();
              deleteThreads[i] = deleteThread;
          }
  
          newLedgerThread.join();
  
          for (Thread deleteThread : deleteThreads) {
              deleteThread.join();
          }
  
          running.set(false);
          for (Thread flushThread : flushThreads) {
              flushThread.join();
          }
  
          assertEquals("Should have been no errors", rc.get(), 0);
          for (long i = 0L; i < numLedgers; ++i) {
              boolean gotError = false;
              try {
                  LOG.error("Checking {}", i);
                  ledgerCache.getEntryOffset(i, 0);
              } catch (NoLedgerException e) {
                  gotError = true;
              }
              if (!gotError) {
                  LOG.error("Ledger {} is still around", i);
                  fail("Found ledger " + i + ", which should have been removed");
              }
          }
      }
  
      /**
       * Test for race between delete and flush.
       * {@link https://issues.apache.org/jira/browse/BOOKKEEPER-604}
       * {@link https://github.com/apache/bookkeeper/issues/1757}
       */
      @Test
      public void testFlushDeleteRace() throws Exception {
          newLedgerCache();
          final AtomicInteger rc = new AtomicInteger(0);
          final LinkedBlockingQueue<Long> ledgerQ = new LinkedBlockingQueue<>(100);
          final byte[] masterKey = "masterKey".getBytes();
          final long numLedgers = 1000;
          final int numFlushers = 10;
          final int numDeleters = 10;
          final AtomicBoolean running = new AtomicBoolean(true);
          Thread newLedgerThread = new Thread() {
                  public void run() {
                      try {
                          for (long i = 0; i < numLedgers && rc.get() == 0; i++) {
                              ledgerCache.setMasterKey(i, masterKey);
  
                              ledgerCache.putEntryOffset(i, 1, 0);
                              ledgerQ.put(i);
                          }
                          for (int i = 0; i < numDeleters; ++i) {
                              ledgerQ.put(-1L);
                          }
                      } catch (Throwable e) {
                          rc.set(-1);
                          LOG.error("Exception in new ledger thread", e);
                      }
                  }
              };
          newLedgerThread.start();
  
          Thread[] flushThreads = new Thread[numFlushers];
          for (int i = 0; i < numFlushers; ++i) {
              Thread flushThread = new Thread() {
                  public void run() {
                      try {
                          while (running.get()) {
                              ledgerCache.flushLedger(true);
                          }
                      } catch (Throwable e) {
                          rc.set(-1);
                          LOG.error("Exception in flush thread", e);
                      }
                      LOG.error("Shutting down flush thread");
                  }
              };
              flushThread.start();
              flushThreads[i] = flushThread;
          }
  
          Thread[] deleteThreads = new Thread[numDeleters];
          for (int i = 0; i < numDeleters; ++i) {
              Thread deleteThread = new Thread() {
                  public void run() {
                      try {
                          while (true) {
                              long id = ledgerQ.take();
                              if (id == -1L) {
                                  break;
                              }
                              LOG.info("Deleting {}", id);
                              ledgerCache.deleteLedger(id);
                          }
                      } catch (Throwable e) {
                          rc.set(-1);
                          LOG.error("Exception in delete thread", e);
                      }
                  }
              };
              deleteThread.start();
              deleteThreads[i] = deleteThread;
          }
  
          newLedgerThread.join();
  
          for (Thread deleteThread : deleteThreads) {
              deleteThread.join();
          }
  
          running.set(false);
          for (Thread flushThread : flushThreads) {
              flushThread.join();
          }
  
          assertEquals("Should have been no errors", rc.get(), 0);
          for (long i = 0L; i < numLedgers; ++i) {
              boolean gotError = false;
              try {
                  LOG.error("Checking {}", i);
                  ledgerCache.getEntryOffset(i, 0);
              } catch (NoLedgerException e) {
                  gotError = true;
              }
              if (!gotError) {
                  LOG.error("Ledger {} is still around", i);
                  fail("Found ledger " + i + ", which should have been removed");
              }
          }
      }
  
      // Mock SortedLedgerStorage to simulate flush failure (Dependency Fault Injection)
      static class FlushTestSortedLedgerStorage extends SortedLedgerStorage {
          final AtomicBoolean injectMemTableSizeLimitReached;
          final AtomicBoolean injectFlushException;
          final AtomicLong injectFlushExceptionForLedger;
          final AtomicInteger numOfTimesFlushSnapshotCalled = new AtomicInteger(0);
          static final long FORALLLEDGERS = -1;
          ServerConfiguration conf;
          StatsLogger statsLogger;
  
          public FlushTestSortedLedgerStorage() {
              super();
              injectMemTableSizeLimitReached = new AtomicBoolean();
              injectFlushException = new AtomicBoolean();
              injectFlushExceptionForLedger = new AtomicLong(FORALLLEDGERS);
          }
  
          public void setInjectMemTableSizeLimitReached(boolean setValue) {
              injectMemTableSizeLimitReached.set(setValue);
          }
  
          public void setInjectFlushException(boolean setValue, long ledgerId) {
              injectFlushException.set(setValue);
              injectFlushExceptionForLedger.set(ledgerId);
          }
  
          public void incrementNumOfTimesFlushSnapshotCalled() {
              numOfTimesFlushSnapshotCalled.incrementAndGet();
          }
  
          public int getNumOfTimesFlushSnapshotCalled() {
              return numOfTimesFlushSnapshotCalled.get();
          }
  
          @Override
          public void initialize(ServerConfiguration conf,
                                 LedgerManager ledgerManager,
                                 LedgerDirsManager ledgerDirsManager,
                                 LedgerDirsManager indexDirsManager,
                                 StatsLogger statsLogger,
                                 ByteBufAllocator allocator) throws IOException {
              super.initialize(
                  conf,
                  ledgerManager,
                  ledgerDirsManager,
                  indexDirsManager,
                  statsLogger,
                  allocator);
              this.conf = conf;
              this.statsLogger = statsLogger;
          }
  
          @Override
          public void setCheckpointSource(CheckpointSource checkpointSource) {
              super.setCheckpointSource(checkpointSource);
              if (this.memTable instanceof EntryMemTableWithParallelFlusher) {
                  this.memTable = new EntryMemTableWithParallelFlusher(conf, checkpointSource, statsLogger) {
                      @Override
                      boolean isSizeLimitReached() {
                          return (injectMemTableSizeLimitReached.get() || super.isSizeLimitReached());
                      }
  
                      @Override
                      long flushSnapshot(final SkipListFlusher flusher, Checkpoint checkpoint) throws IOException {
                          incrementNumOfTimesFlushSnapshotCalled();
                          return super.flushSnapshot(flusher, checkpoint);
                      }
                  };
              } else {
                  this.memTable = new EntryMemTable(conf, checkpointSource, statsLogger) {
                      @Override
                      boolean isSizeLimitReached() {
                          return (injectMemTableSizeLimitReached.get() || super.isSizeLimitReached());
                      }
  
                      @Override
                      long flushSnapshot(final SkipListFlusher flusher, Checkpoint checkpoint) throws IOException {
                          incrementNumOfTimesFlushSnapshotCalled();
                          return super.flushSnapshot(flusher, checkpoint);
                      }
                  };
              }
          }
  
          @Override
          public void process(long ledgerId, long entryId, ByteBuf buffer) throws IOException {
              if (injectFlushException.get() && ((injectFlushExceptionForLedger.get() == FORALLLEDGERS)
                      || (injectFlushExceptionForLedger.get() == ledgerId))) {
                  throw new IOException("Injected Exception");
              }
              super.process(ledgerId, entryId, buffer);
          }
  
          // simplified memTable full callback.
          @Override
          public void onSizeLimitReached(final CheckpointSource.Checkpoint cp) throws IOException {
              LOG.info("Reached size {}", cp);
              // use synchronous way
              try {
                  LOG.info("Started flushing mem table.");
                  memTable.flush(FlushTestSortedLedgerStorage.this);
              } catch (IOException e) {
                  getStateManager().doTransitionToReadOnlyMode(false);
                  LOG.error("Exception thrown while flushing skip list cache.", e);
           }
           }
  
      }
  
      @Test
      public void testEntryMemTableFlushFailure() throws Exception {
          File tmpDir = createTempDir("bkTest", ".dir");
          File curDir = BookieImpl.getCurrentDirectory(tmpDir);
          BookieImpl.checkDirectoryStructure(curDir);
  
          int gcWaitTime = 1000;
          ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
          conf.setGcWaitTime(gcWaitTime);
          conf.setLedgerDirNames(new String[] { tmpDir.toString() });
          conf.setLedgerStorageClass(FlushTestSortedLedgerStorage.class.getName());
  
          Bookie bookie = new TestBookieImpl(conf);
          FlushTestSortedLedgerStorage flushTestSortedLedgerStorage =
              (FlushTestSortedLedgerStorage) bookie.getLedgerStorage();
          EntryMemTable memTable = flushTestSortedLedgerStorage.memTable;
  
          // this bookie.addEntry call is required. FileInfo for Ledger 1 would be created with this call.
          // without the fileinfo, 'flushTestSortedLedgerStorage.addEntry' calls will fail
          // because of BOOKKEEPER-965 change.
          bookie.addEntry(generateEntry(1, 1), false, new BookieImpl.NopWriteCallback(), null, "passwd".getBytes());
  
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 2));
          assertFalse("Bookie is expected to be in ReadWrite mode", bookie.isReadOnly());
          assertTrue("EntryMemTable SnapShot is expected to be empty", memTable.snapshot.isEmpty());
  
          // set flags, so that FlushTestSortedLedgerStorage simulates FlushFailure scenario
          flushTestSortedLedgerStorage.setInjectMemTableSizeLimitReached(true);
          flushTestSortedLedgerStorage.setInjectFlushException(true, FlushTestSortedLedgerStorage.FORALLLEDGERS);
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 2));
  
          // since we simulated sizeLimitReached, snapshot shouldn't be empty
          assertFalse("EntryMemTable SnapShot is not expected to be empty", memTable.snapshot.isEmpty());
          assertEquals("Flusher called", 1, flushTestSortedLedgerStorage.getNumOfTimesFlushSnapshotCalled());
  
          // set the flags to false, so flush will succeed this time
          flushTestSortedLedgerStorage.setInjectMemTableSizeLimitReached(false);
          flushTestSortedLedgerStorage.setInjectFlushException(false, FlushTestSortedLedgerStorage.FORALLLEDGERS);
  
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 3));
          // since we expect memtable flush to succeed, memtable snapshot should be empty
          assertTrue("EntryMemTable SnapShot is expected to be empty, because of successful flush",
                  memTable.snapshot.isEmpty());
      }
  
      @Test
      public void testSortedLedgerFlushFailure() throws Exception {
          // most of the code is same to the testEntryMemTableFlushFailure
          File tmpDir = createTempDir("bkTest", ".dir");
          File curDir = BookieImpl.getCurrentDirectory(tmpDir);
          BookieImpl.checkDirectoryStructure(curDir);
  
          int gcWaitTime = 1000;
          ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
          conf.setGcWaitTime(gcWaitTime)
              .setLedgerDirNames(new String[] { tmpDir.toString() })
              .setJournalDirName(tmpDir.toString())
              .setLedgerStorageClass(FlushTestSortedLedgerStorage.class.getName());
  
          Bookie bookie = new TestBookieImpl(conf);
          bookie.start();
          FlushTestSortedLedgerStorage flushTestSortedLedgerStorage = (FlushTestSortedLedgerStorage) bookie.
                  getLedgerStorage();
          EntryMemTable memTable = flushTestSortedLedgerStorage.memTable;
  
          bookie.addEntry(generateEntry(1, 1), false, new BookieImpl.NopWriteCallback(), null, "passwd".getBytes());
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 2));
          assertFalse("Bookie is expected to be in ReadWrite mode", bookie.isReadOnly());
          assertTrue("EntryMemTable SnapShot is expected to be empty", memTable.snapshot.isEmpty());
  
          // set flags, so that FlushTestSortedLedgerStorage simulates FlushFailure scenario
          flushTestSortedLedgerStorage.setInjectMemTableSizeLimitReached(true);
          flushTestSortedLedgerStorage.setInjectFlushException(true, FlushTestSortedLedgerStorage.FORALLLEDGERS);
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 2));
  
          // since we simulated sizeLimitReached, snapshot shouldn't be empty
          assertFalse("EntryMemTable SnapShot is not expected to be empty", memTable.snapshot.isEmpty());
          // after flush failure, the bookie is set to readOnly
          assertTrue("Bookie is expected to be in Read mode", bookie.isReadOnly());
          // write fail
          CountDownLatch latch = new CountDownLatch(1);
          bookie.addEntry(generateEntry(1, 3), false, new BookkeeperInternalCallbacks.WriteCallback(){
              public void writeComplete(int rc, long ledgerId, long entryId, BookieId addr, Object ctx){
                  LOG.info("Write to bk succeed due to the bookie readOnly mode check is in the request parse step. "
                      + "In the addEntry step, we won't check bookie's mode");
                  assertEquals(OK, rc);
                  latch.countDown();
              }
  
          }, null, "passwd".getBytes());
          latch.await();
          bookie.shutdown();
  
      }
  
      private ByteBuf generateEntry(long ledger, long entry) {
          byte[] data = ("ledger-" + ledger + "-" + entry).getBytes();
          ByteBuf bb = Unpooled.buffer(8 + 8 + data.length);
          bb.writeLong(ledger);
          bb.writeLong(entry);
          bb.writeBytes(data);
          return bb;
      }
  
      @Test
      public void testEntryMemTableParallelFlush() throws Exception {
          int gcWaitTime = 1000;
          ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
          conf.setGcWaitTime(gcWaitTime);
          conf.setLedgerDirNames(createAndGetLedgerDirs(1));
          conf.setLedgerStorageClass(FlushTestSortedLedgerStorage.class.getName());
          // enable entrylog per ledger
          conf.setEntryLogPerLedgerEnabled(true);
  
          Bookie bookie = new TestBookieImpl(conf);
          FlushTestSortedLedgerStorage flushTestSortedLedgerStorage =
              (FlushTestSortedLedgerStorage) bookie.getLedgerStorage();
          EntryMemTable memTable = flushTestSortedLedgerStorage.memTable;
  
          /*
           * this bookie.addEntry call is required. FileInfo for Ledger 1, 2, 3
           * would be created with this call. without the fileinfo,
           * 'flushTestSortedLedgerStorage.addEntry' calls will fail because of
           * BOOKKEEPER-965 change.
           */
          bookie.addEntry(generateEntry(1, 1), false, new BookieImpl.NopWriteCallback(), null, "passwd".getBytes());
          bookie.addEntry(generateEntry(2, 1), false, new BookieImpl.NopWriteCallback(), null, "passwd".getBytes());
          bookie.addEntry(generateEntry(3, 1), false, new BookieImpl.NopWriteCallback(), null, "passwd".getBytes());
  
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 2));
          flushTestSortedLedgerStorage.addEntry(generateEntry(2, 2));
          flushTestSortedLedgerStorage.addEntry(generateEntry(3, 2));
  
          assertTrue("EntryMemTable SnapShot is expected to be empty", memTable.snapshot.isEmpty());
          assertFalse("EntryMemTable is not expected to be empty", memTable.isEmpty());
  
          // inject MemTableSizeLimitReached, so entrymemtable will be flushed
          flushTestSortedLedgerStorage.setInjectMemTableSizeLimitReached(true);
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 3));
  
          // since we simulated sizeLimitReached, snapshot should have been created and flushed
          assertTrue("EntryMemTable SnapShot is expected to be empty", memTable.snapshot.isEmpty());
          assertEquals("Flusher called", 1, flushTestSortedLedgerStorage.getNumOfTimesFlushSnapshotCalled());
      }
  
      @Test
      public void testEntryMemTableParallelFlushWithFlushException() throws Exception {
          int gcWaitTime = 1000;
          ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
          conf.setGcWaitTime(gcWaitTime);
          conf.setLedgerDirNames(createAndGetLedgerDirs(1));
          conf.setLedgerStorageClass(FlushTestSortedLedgerStorage.class.getName());
          // enable entrylog per ledger
          conf.setEntryLogPerLedgerEnabled(true);
  
          Bookie bookie = new TestBookieImpl(conf);
          FlushTestSortedLedgerStorage flushTestSortedLedgerStorage =
              (FlushTestSortedLedgerStorage) bookie.getLedgerStorage();
          EntryMemTable memTable = flushTestSortedLedgerStorage.memTable;
  
          /*
           * this bookie.addEntry call is required. FileInfo for Ledger 1, 2, 3
           * would be created with this call. without the fileinfo,
           * 'flushTestSortedLedgerStorage.addEntry' calls will fail because of
           * BOOKKEEPER-965 change.
           */
          bookie.addEntry(generateEntry(1, 1), false, new BookieImpl.NopWriteCallback(), null, "passwd".getBytes());
          bookie.addEntry(generateEntry(2, 1), false, new BookieImpl.NopWriteCallback(), null, "passwd".getBytes());
          bookie.addEntry(generateEntry(3, 1), false, new BookieImpl.NopWriteCallback(), null, "passwd".getBytes());
  
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 4));
          flushTestSortedLedgerStorage.addEntry(generateEntry(2, 4));
          flushTestSortedLedgerStorage.addEntry(generateEntry(3, 4));
  
          // inject MemTableSizeLimitReached and FlushException, so entrymemtable flush will fail
          flushTestSortedLedgerStorage.setInjectMemTableSizeLimitReached(true);
          flushTestSortedLedgerStorage.setInjectFlushException(true, 1L);
  
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 5));
          // since we simulate FlushException, memtable snapshot should not be empty
          assertFalse("EntryMemTable SnapShot is not expected to be empty", memTable.snapshot.isEmpty());
          assertEquals("Flusher called", 1, flushTestSortedLedgerStorage.getNumOfTimesFlushSnapshotCalled());
  
          flushTestSortedLedgerStorage.setInjectFlushException(false, FlushTestSortedLedgerStorage.FORALLLEDGERS);
          flushTestSortedLedgerStorage.addEntry(generateEntry(1, 5));
          /*
           * since MemTableSizeLimitReached is already set to true, and flush
           * exception is disabled, this time memtable snapshot should be flushed
           */
          assertTrue("EntryMemTable SnapShot is expected to be empty", memTable.snapshot.isEmpty());
          assertEquals("Flusher called", 2, flushTestSortedLedgerStorage.getNumOfTimesFlushSnapshotCalled());
      }
  
      String[] createAndGetLedgerDirs(int numOfLedgerDirs) throws IOException {
          File ledgerDir;
          File curDir;
          String[] ledgerDirsPath = new String[numOfLedgerDirs];
          for (int i = 0; i < numOfLedgerDirs; i++) {
              ledgerDir = createTempDir("bkTest", ".dir");
              curDir = BookieImpl.getCurrentDirectory(ledgerDir);
              BookieImpl.checkDirectoryStructure(curDir);
              ledgerDirsPath[i] = ledgerDir.getAbsolutePath();
          }
          return ledgerDirsPath;
      }
  }
  