diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java index 19d5a7961f7..b16b229d606 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java +++ b/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) { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java index 9a2c42d317b..879151a3608 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java @@ -135,6 +135,7 @@ public class ServerConfiguration extends AbstractConfiguration) 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) 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) 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; + } + + } + } + } diff --git a/conf/bk_server.conf b/conf/bk_server.conf index 4da030d712f..e57b5ec10f6 100755 --- a/conf/bk_server.conf +++ b/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