[bookie] provide option to skip adding entry to journal

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java

@@ -592,6 +592,7 @@ static void writePaddingBytes(JournalChannel jc, ByteBuf paddingBuffer, int jour
     final int maxBackupJournals;
 
     final File journalDirectory;
+    final boolean journalSkipEntryEnable;
     final ServerConfiguration conf;
     final ForceWriteThread forceWriteThread;
     // Time after which we will stop grouping and issue the flush
@@ -655,6 +656,7 @@ public Journal(int journalIndex, File journalDirectory, ServerConfiguration conf
         this.ledgerDirsManager = ledgerDirsManager;
         this.conf = conf;
         this.journalDirectory = journalDirectory;
+        this.journalSkipEntryEnable = conf.getJournalSkipEntryEnable();
         this.maxJournalSize = conf.getMaxJournalSizeMB() * MB;
         this.journalPreAllocSize = conf.getJournalPreAllocSizeMB() * MB;
         this.journalWriteBufferSize = conf.getJournalWriteBufferSizeKB() * KB;
@@ -855,14 +857,18 @@ public void logAddEntry(ByteBuf entry, boolean ackBeforeSync, WriteCallback cb,
     public void logAddEntry(long ledgerId, long entryId, ByteBuf entry,
                             boolean ackBeforeSync, WriteCallback cb, Object ctx)
             throws InterruptedException {
+        QueueEntry queueEntry = QueueEntry.create(entry, ackBeforeSync, ledgerId, entryId, cb, ctx,
+                MathUtils.nowInNano(), journalStats.getJournalAddEntryStats(), journalStats.getJournalQueueSize());
+        // skip adding entry to journal if skip-journal is enabled
+        if (this.journalSkipEntryEnable) {
+            cbThreadPool.execute(queueEntry);
+            return;
+        }
         //Retain entry until it gets written to journal
         entry.retain();
 
         journalStats.getJournalQueueSize().inc();
-        queue.put(QueueEntry.create(
-                entry, ackBeforeSync,  ledgerId, entryId, cb, ctx, MathUtils.nowInNano(),
-                journalStats.getJournalAddEntryStats(),
-                journalStats.getJournalQueueSize()));
+        queue.put(queueEntry);
     }
 
     void forceLedger(long ledgerId, WriteCallback cb, Object ctx) {

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java

@@ -135,6 +135,7 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
     protected static final String NUM_JOURNAL_CALLBACK_THREADS = "numJournalCallbackThreads";
     protected static final String JOURNAL_FORMAT_VERSION_TO_WRITE = "journalFormatVersionToWrite";
     protected static final String JOURNAL_QUEUE_SIZE = "journalQueueSize";
+    protected static final String JOURNAL_SKIP_ENTRY_ENABLE = "journalSkipEntryEnable";
     // backpressure control
     protected static final String MAX_ADDS_IN_PROGRESS_LIMIT = "maxAddsInProgressLimit";
     protected static final String MAX_READS_IN_PROGRESS_LIMIT = "maxReadsInProgressLimit";
@@ -787,6 +788,29 @@ public ServerConfiguration setJournalFormatVersionToWrite(int version) {
         return this;
     }
 
+    /**
+     * Is bookie enable to skip adding entry to journal and acknowledge
+     * immediately once written to ledger cache.
+     *
+     * @return journal format version to write.
+     */
+    public boolean getJournalSkipEntryEnable() {
+        return this.getBoolean(JOURNAL_SKIP_ENTRY_ENABLE, false);
+    }
+
+    /**
+     * Enable skip add-entry to journal and acknowledge immediately once written
+     * to ledger cache.
+     *
+     * @param journalSkipEntryEnable
+     *            flag to skip adding entry to journal
+     * @return server configuration.
+     */
+    public ServerConfiguration setJournalSkipEntryEnable(boolean journalSkipEntryEnable) {
+        this.setProperty(JOURNAL_SKIP_ENTRY_ENABLE, journalSkipEntryEnable);
+        return this;
+    }
+
     /**
      * Set the size of the journal queue.
      *

bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieJournalRollingTest.java

@@ -25,17 +25,25 @@
 import static org.junit.Assert.assertTrue;
 
 import java.io.File;
+import java.lang.reflect.Field;
 import java.util.Enumeration;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.bookkeeper.bookie.Bookie;
+import org.apache.bookkeeper.bookie.Journal;
+import org.apache.bookkeeper.bookie.stats.JournalStats;
 import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
 import org.apache.bookkeeper.client.BKException;
 import org.apache.bookkeeper.client.BookKeeper.DigestType;
 import org.apache.bookkeeper.client.LedgerEntry;
 import org.apache.bookkeeper.client.LedgerHandle;
 import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.conf.TestBKConfiguration;
+import org.apache.bookkeeper.stats.Counter;
+import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.stats.StatsLogger;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -280,4 +288,113 @@ public void testReplayDeletedLedgerJournalEntries() throws Exception {
         restartBookies(newConf);
     }
 
+    /**
+     * Skip writing entry to journal and restart bookie with the change.
+     * This test verifies 1. bookie restart with disable/enable change is
+     * gracefully handled by bookie and bookie comes up gracefully. 2. bookie
+     * doesn't add entry into journal when journalSkipEntryEnable config is
+     * enabled.
+     * @throws Exception
+     */
+    @SuppressWarnings("unchecked")
+    @Test
+    public void testSkipJournal() throws Exception {
+        // set flush interval to a large value
+        ServerConfiguration newConf = TestBKConfiguration.newServerConfiguration();
+        newConf.setFlushInterval(999999999);
+        newConf.setAllowEphemeralPorts(false);
+        Field journalsField = Bookie.class.getDeclaredField("journals");
+        journalsField.setAccessible(true);
+        Field journalStatsField = Journal.class.getDeclaredField("journalStats");
+        journalStatsField.setAccessible(true);
+        final int totalAddEntry = 10;
+
+        // restart bookies
+        restartBookies(newConf);
+        // Write enough ledger entries so that we roll over journals
+        JournalStatsTest stats = new JournalStatsTest(new NullStatsLogger());
+        Bookie bookie = bs.get(0).getBookie();
+        Journal journal = ((List<Journal>) journalsField.get(bookie)).get(0);
+        journalStatsField.set(journal, stats);
+        writeLedgerEntries(1, 1024, totalAddEntry);
+        assertTrue(stats.totalAddEntry() > 0);
+
+        newConf.setJournalSkipEntryEnable(true);
+        // restart bookie with skip-journal option enabled.
+        restartBookies(newConf);
+        JournalStatsTest statsWithSkipJournalEnabled = new JournalStatsTest(new NullStatsLogger());
+        bookie = bs.get(0).getBookie();
+        journal = ((List<Journal>) journalsField.get(bookie)).get(0);
+        journalStatsField.set(journal, statsWithSkipJournalEnabled);
+        writeLedgerEntries(1, 1024, totalAddEntry);
+        assertEquals(statsWithSkipJournalEnabled.totalAddEntry(), 0);
+
+        newConf.setJournalSkipEntryEnable(false);
+        // restart bookie with skip-journal option disabled.
+        restartBookies(newConf);
+        JournalStatsTest statsWithSkipJournalDisabled = new JournalStatsTest(new NullStatsLogger());
+        bookie = bs.get(0).getBookie();
+        journal = ((List<Journal>) journalsField.get(bookie)).get(0);
+        journalStatsField.set(journal, statsWithSkipJournalDisabled);
+        writeLedgerEntries(1, 1024, totalAddEntry);
+        assertTrue(statsWithSkipJournalDisabled.totalAddEntry() > 0);
+    }
+
+    /**
+     * Test JournalStats class to capture add-entry counts for journal.
+     */
+    public static class JournalStatsTest extends JournalStats {
+        private TestCounter queueSizeCounter = new TestCounter();
+        public JournalStatsTest(StatsLogger statsLogger) {
+            super(statsLogger);
+        }
+        @Override
+        public Counter getJournalQueueSize() {
+            return queueSizeCounter;
+        }
+
+        public long totalAddEntry() {
+            return queueSizeCounter.totalAddCount;
+        }
+
+        /**
+         * Test count to capture add count.
+         */
+        public static class TestCounter implements Counter {
+            public volatile long totalAddCount = 0;
+            long count = 0;
+
+            @Override
+            public void clear() {
+                count = 0;
+            }
+
+            @Override
+            public void inc() {
+                System.out.println("incrementing");
+                count++;
+                totalAddCount++;
+            }
+
+            @Override
+            public void dec() {
+                count--;
+
+            }
+
+            @Override
+            public void add(long delta) {
+                count += delta;
+                totalAddCount += delta;
+
+            }
+
+            @Override
+            public Long get() {
+                return count;
+            }
+
+        }
+    }
+
 }

conf/bk_server.conf

@@ -294,6 +294,10 @@ journalDirectories=/tmp/bk-txn
 # and onward versions.
 # journalFormatVersionToWrite=6
 
+# Enable skip add-entry to journal and acknowledge immediately once written
+# to ledger cache.
+# journalSkipEntryEnable=false
+
 # Max file size of journal file, in mega bytes
 # A new journal file will be created when the old one reaches the file size limitation
 # journalMaxSizeMB=2048