ISSUE-2806: OOM as 1 million ledgers per entry log

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

@@ -35,10 +35,13 @@ public class EntryLogMetadata {
     private final ConcurrentLongLongHashMap ledgersMap;
 
     public EntryLogMetadata(long logId) {
+        this(logId, 1);
+    }
+    public EntryLogMetadata(long logId, int concurrencyLevel) {
         this.entryLogId = logId;
 
         totalSize = remainingSize = 0;
-        ledgersMap = new ConcurrentLongLongHashMap(256, 1);
+        ledgersMap = new ConcurrentLongLongHashMap(256, concurrencyLevel);
     }
 
     public void addLedgerSize(long ledgerId, long size) {

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

@@ -89,9 +89,13 @@ static class BufferedLogChannel extends BufferedChannel {
 
         public BufferedLogChannel(ByteBufAllocator allocator, FileChannel fc, int writeCapacity, int readCapacity,
                 long logId, File logFile, long unpersistedBytesBound) throws IOException {
+            this(allocator, fc, writeCapacity, readCapacity, logId, logFile, unpersistedBytesBound, 1);
+        }
+        public BufferedLogChannel(ByteBufAllocator allocator, FileChannel fc, int writeCapacity, int readCapacity,
+                                  long logId, File logFile, long unpersistedBytesBound, int logMetaMapConcurrency) throws IOException {
             super(allocator, fc, writeCapacity, readCapacity, unpersistedBytesBound);
             this.logId = logId;
-            this.entryLogMetadata = new EntryLogMetadata(logId);
+            this.entryLogMetadata = new EntryLogMetadata(logId, logMetaMapConcurrency);
             this.logFile = logFile;
         }
         public long getLogId() {
@@ -229,11 +233,13 @@ private static class Header {
         final int version;
         final long ledgersMapOffset;
         final int ledgersCount;
+        final long createTimestamp;
 
-        Header(int version, long ledgersMapOffset, int ledgersCount) {
+        Header(int version, long ledgersMapOffset, int ledgersCount, long createTimestamp) {
             this.version = version;
             this.ledgersMapOffset = ledgersMapOffset;
             this.ledgersCount = ledgersCount;
+            this.createTimestamp = createTimestamp;
         }
     }
 
@@ -254,6 +260,7 @@ private static class Header {
 
     static final int HEADER_VERSION_POSITION = 4;
     static final int LEDGERS_MAP_OFFSET_POSITION = HEADER_VERSION_POSITION + 4;
+    static final int CREATE_TIMESTAMP_POSITION = LEDGERS_MAP_OFFSET_POSITION + 8 + 4;
 
     /**
      * Ledgers map is composed of multiple parts that can be split into separated entries. Each of them is composed of:
@@ -357,6 +364,8 @@ public EntryLogger(ServerConfiguration conf,
         // this header buffer is cleared before writing it into the new logChannel.
         logfileHeader.writeBytes("BKLO".getBytes(UTF_8));
         logfileHeader.writeInt(HEADER_CURRENT_VERSION);
+        logfileHeader.writerIndex(CREATE_TIMESTAMP_POSITION);
+        logfileHeader.writeLong(System.currentTimeMillis());
         logfileHeader.writerIndex(LOGFILE_HEADER_SIZE);
 
         // Find the largest logId
@@ -902,11 +911,16 @@ private Header getHeaderForLogId(long entryLogId) throws IOException {
 
             long ledgersMapOffset = headers.readLong();
             int ledgersCount = headers.readInt();
-            return new Header(headerVersion, ledgersMapOffset, ledgersCount);
+            long createTimestamp = headers.readLong();
+            return new Header(headerVersion, ledgersMapOffset, ledgersCount, createTimestamp);
         } finally {
             headers.release();
         }
     }
+    public long getCreateTimestampForLogId(long entryLogId) throws IOException {
+        Header header = getHeaderForLogId(entryLogId);
+        return header.createTimestamp;
+    }
 
     private BufferedReadChannel getChannelForLogId(long entryLogId) throws IOException {
         BufferedReadChannel fc = getFromChannels(entryLogId);
@@ -1087,7 +1101,7 @@ EntryLogMetadata extractEntryLogMetadataFromIndex(long entryLogId) throws IOExce
 
         // There can be multiple entries containing the various components of the serialized ledgers map
         long offset = header.ledgersMapOffset;
-        EntryLogMetadata meta = new EntryLogMetadata(entryLogId);
+        EntryLogMetadata meta = new EntryLogMetadata(entryLogId, conf.getEntryLogLedgerMapConcurrency());
 
         final int maxMapSize = LEDGERS_MAP_HEADER_SIZE + LEDGERS_MAP_ENTRY_SIZE * LEDGERS_MAP_MAX_BATCH_SIZE;
         ByteBuf ledgersMap = allocator.directBuffer(maxMapSize);
@@ -1151,7 +1165,7 @@ EntryLogMetadata extractEntryLogMetadataFromIndex(long entryLogId) throws IOExce
     }
 
     private EntryLogMetadata extractEntryLogMetadataByScanning(long entryLogId) throws IOException {
-        final EntryLogMetadata meta = new EntryLogMetadata(entryLogId);
+        final EntryLogMetadata meta = new EntryLogMetadata(entryLogId, conf.getEntryLogLedgerMapConcurrency());
 
         // Read through the entry log file and extract the entry log meta
         scanEntryLog(entryLogId, new EntryLogScanner() {

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

@@ -84,6 +84,8 @@ class EntryLoggerAllocator {
         // this header buffer is cleared before writing it into the new logChannel.
         logfileHeader.writeBytes("BKLO".getBytes(UTF_8));
         logfileHeader.writeInt(EntryLogger.HEADER_CURRENT_VERSION);
+        logfileHeader.writerIndex(EntryLogger.CREATE_TIMESTAMP_POSITION);
+        logfileHeader.writeLong(System.currentTimeMillis());
         logfileHeader.writerIndex(EntryLogger.LOGFILE_HEADER_SIZE);
 
     }
@@ -167,7 +169,12 @@ private synchronized BufferedLogChannel allocateNewLog(File dirForNextEntryLog,
         FileChannel channel = new RandomAccessFile(newLogFile, "rw").getChannel();
 
         BufferedLogChannel logChannel = new BufferedLogChannel(byteBufAllocator, channel, conf.getWriteBufferBytes(),
-                conf.getReadBufferBytes(), preallocatedLogId, newLogFile, conf.getFlushIntervalInBytes());
+                conf.getReadBufferBytes(), preallocatedLogId, newLogFile, conf.getFlushIntervalInBytes(),
+                conf.getEntryLogLedgerMapConcurrency());
+        //update create timestamp
+        logfileHeader.writerIndex(EntryLogger.CREATE_TIMESTAMP_POSITION);
+        logfileHeader.writeLong(System.currentTimeMillis());
+        logfileHeader.writerIndex(EntryLogger.LOGFILE_HEADER_SIZE);
         logfileHeader.readerIndex(0);
         logChannel.write(logfileHeader);
 

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

@@ -613,6 +613,10 @@ protected Map<Long, EntryLogMetadata> extractMetaFromEntryLogs(Map<Long, EntryLo
                 continue;
             }
 
+            if (!isNeedExtractMeta(entryLogId)) {
+                LOG.debug("entry log:{} do not need to extract meta");
+                continue;
+            }
             LOG.info("Extracting entry log meta from entryLogId: {}", entryLogId);
 
             try {
@@ -641,6 +645,21 @@ protected Map<Long, EntryLogMetadata> extractMetaFromEntryLogs(Map<Long, EntryLo
         return entryLogMetaMap;
     }
 
+    private boolean isNeedExtractMeta(long entryLogId) {
+        try {
+            if (conf.getExtractMetaDelayTimeSeconds() > 0) {
+                long createTimestamp = entryLogger.getCreateTimestampForLogId(entryLogId);
+                if (createTimestamp > 0 && System.currentTimeMillis() - createTimestamp
+                        < conf.getExtractMetaDelayTimeSeconds() * 1000) {
+                    return false;
+                }
+            }
+        } catch (Exception ex) {
+            LOG.warn("Failed to get create timestamp from: {}.log : {}", entryLogId, ex.getMessage());
+        }
+        return true;
+    }
+
     CompactableLedgerStorage getLedgerStorage() {
         return ledgerStorage;
     }

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java

@@ -149,8 +149,10 @@ public SingleDirectoryDbLedgerStorage(ServerConfiguration conf, LedgerManager le
         log.info("Creating single directory db ledger storage on {}", baseDir);
 
         this.writeCacheMaxSize = writeCacheSize;
-        this.writeCache = new WriteCache(allocator, writeCacheMaxSize / 2);
-        this.writeCacheBeingFlushed = new WriteCache(allocator, writeCacheMaxSize / 2);
+        this.writeCache = new WriteCache(allocator, writeCacheMaxSize / 2,
+                conf.getFlushEntrySortBufferInitSize());
+        this.writeCacheBeingFlushed = new WriteCache(allocator, writeCacheMaxSize / 2,
+                conf.getFlushEntrySortBufferInitSize());
 
         this.checkpointSource = checkpointSource;
 

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/WriteCache.java

@@ -84,10 +84,15 @@ public interface EntryConsumer {
 
     public WriteCache(ByteBufAllocator allocator, long maxCacheSize) {
         // Default maxSegmentSize set to 1Gb
-        this(allocator, maxCacheSize, 1 * 1024 * 1024 * 1024);
+        this(allocator, maxCacheSize, 1 * 1024 * 1024 * 1024, 0);
     }
 
-    public WriteCache(ByteBufAllocator allocator, long maxCacheSize, int maxSegmentSize) {
+    public WriteCache(ByteBufAllocator allocator, long maxCacheSize, int entrySortBufferInitSize) {
+        // Default maxSegmentSize set to 1Gb
+        this(allocator, maxCacheSize, 1 * 1024 * 1024 * 1024, entrySortBufferInitSize);
+    }
+
+    public WriteCache(ByteBufAllocator allocator, long maxCacheSize, int maxSegmentSize, int entrySortBufferInitSize) {
         checkArgument(maxSegmentSize > 0);
 
         long alignedMaxSegmentSize = alignToPowerOfTwo(maxSegmentSize);
@@ -110,6 +115,11 @@ public WriteCache(ByteBufAllocator allocator, long maxCacheSize, int maxSegmentS
 
         int lastSegmentSize = (int) (maxCacheSize % maxSegmentSize);
         cacheSegments[segmentsCount - 1] = Unpooled.directBuffer(lastSegmentSize, lastSegmentSize);
+
+        if (entrySortBufferInitSize > 8) {
+            int bufferSize = entrySortBufferInitSize / 8;
+            sortedEntries = new long[bufferSize];
+        }
     }
 
     public void clear() {

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

@@ -115,12 +115,14 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
             "gcOverreplicatedLedgerMaxConcurrentRequests";
     protected static final String USE_TRANSACTIONAL_COMPACTION = "useTransactionalCompaction";
     protected static final String VERIFY_METADATA_ON_GC = "verifyMetadataOnGC";
+    protected static final String EXTRACT_META_DELAY_TIME_SECONDS = "extractMetaDelayTimeSeconds";
     // Scrub Parameters
     protected static final String LOCAL_SCRUB_PERIOD = "localScrubInterval";
     protected static final String LOCAL_SCRUB_RATE_LIMIT = "localScrubRateLimit";
     // Sync Parameters
     protected static final String FLUSH_INTERVAL = "flushInterval";
     protected static final String FLUSH_ENTRYLOG_INTERVAL_BYTES = "flushEntrylogBytes";
+    protected static final String FLUSH_ENTRY_SORT_BUFFER_INIT_SIZE = "flushEntrySortBufferInitSize";
     // Bookie death watch interval
     protected static final String DEATH_WATCH_INTERVAL = "bookieDeathWatchInterval";
     // Ledger Cache Parameters
@@ -279,7 +281,8 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
     protected static final String ENTRY_LOG_PER_LEDGER_ENABLED = "entryLogPerLedgerEnabled";
     // In the case of multipleentrylogs, multiple threads can be used to flush the memtable parallelly.
     protected static final String NUMBER_OF_MEMTABLE_FLUSH_THREADS = "numOfMemtableFlushThreads";
-
+    //ledgers map concurrency of entry log meta data
+    protected static final String ENTRY_LOG_LEDGER_MAP_CONCURRENCY = "entryLogLedgerMapConcurrency";
 
     /*
      * config specifying if the entrylog per ledger is enabled, then the amount
@@ -480,6 +483,27 @@ public ServerConfiguration setVerifyMetadataOnGc(boolean verifyMetadataOnGC) {
         return this;
     }
 
+    /**
+     * Get extract meta delay time. In delay time, meta data will not load, to reduce memory usage.
+     * If the retention time of all the entry is the same and ledger count is very large, the delay time can
+     * be set to the retention time.
+     * @return meta extracting delay time in second
+     */
+    public long getExtractMetaDelayTimeSeconds(){
+        return this.getLong(EXTRACT_META_DELAY_TIME_SECONDS, 0);
+    }
+
+    /**
+     * Set extract meta delay time.
+     *
+     * @param delayTimeSeconds
+     * @return server configuration
+     */
+    public ServerConfiguration setExtractMetaDelayTimeSeconds(long delayTimeSeconds){
+        this.setProperty(EXTRACT_META_DELAY_TIME_SECONDS, delayTimeSeconds);
+        return this;
+    }
+
     /**
      * Get whether local scrub is enabled.
      *
@@ -573,6 +597,28 @@ public ServerConfiguration setFlushIntervalInBytes(long flushInterval) {
         return this;
     }
 
+    /**
+     * Get entry sorted buffer init size.
+     *
+     * <p>Default is 0. Greater than 1, will pre-allocated buffer to reduce the probability of allocation failure.
+     *
+     * @return entry sorted buffer init size
+     */
+
+    public int getFlushEntrySortBufferInitSize() {
+        return this.getInt(FLUSH_ENTRY_SORT_BUFFER_INIT_SIZE, 0);
+    }
+
+    /**
+     * Set flush entry sorted buffer init size
+     *
+     * @param initSize
+     * @return server configuration
+     */
+    public ServerConfiguration SetFlushEntrySortBufferInitSize(int initSize) {
+        this.setProperty(FLUSH_ENTRY_SORT_BUFFER_INIT_SIZE, Integer.toString(initSize));
+        return this;
+    }
 
     /**
      * Get bookie death watch interval.
@@ -3561,6 +3607,28 @@ public ServerConfiguration setNumOfMemtableFlushThreads(int numOfMemtableFlushTh
         return this;
     }
 
+    /**
+     * Get ledger map concurrency of entry log. It should be set greater than 1 to avoid allocating a large contiguous
+     * memory, if there are many ledgers in a entry log.
+     *
+     * @return ledger map concurrency.
+     */
+    public int getEntryLogLedgerMapConcurrency() {
+        return this.getInt(ENTRY_LOG_LEDGER_MAP_CONCURRENCY, 1);
+    }
+
+    /**
+     * Set ledger map concurrency of entry log.
+     *
+     * @param concurrency
+     *          map concurrency level
+     * @return client configuration.
+     */
+    public ServerConfiguration setEntryLogLedgerMapConcurrency(int concurrency) {
+        this.setProperty(ENTRY_LOG_LEDGER_MAP_CONCURRENCY, Integer.toString(concurrency));
+        return this;
+    }
+
     /*
      * in entryLogPerLedger feature, this specifies the time, once this duration
      * has elapsed after the entry's last access, that entry should be

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/EntryLogTest.java

@@ -29,6 +29,7 @@
 import com.google.common.collect.Sets;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
 import io.netty.buffer.UnpooledByteBufAllocator;
 
@@ -1824,4 +1825,50 @@ public void testSwappingEntryLogManager(boolean initialEntryLogPerLedgerEnabled,
         }
     }
 
+
+    /**
+     * testcase for createtimestamp of header
+     */
+    @Test
+    public void testCreatTimestamp() throws Exception {
+        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+        conf.setLedgerDirNames(createAndGetLedgerDirs(1));
+        LedgerDirsManager ledgerDirsManager = new LedgerDirsManager(conf, conf.getLedgerDirs(),
+                new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold()));
+
+        long timestamp = System.currentTimeMillis();
+        EntryLogger entryLogger = new EntryLogger(conf, ledgerDirsManager);
+        EntryLogManagerForSingleEntryLog entryLogManager =
+                (EntryLogManagerForSingleEntryLog) entryLogger.getEntryLogManager();
+
+        long entry0Position = entryLogger.addEntry(0L, generateEntry(0, 1));
+        long entry1Position = entryLogger.addEntry(1L, generateEntry(1, 1));
+        long entry2Position = entryLogger.addEntry(2L, generateEntry(2, 1));
+
+        BufferedLogChannel bufferedLogChannel = entryLogManager.getCurrentLogForLedger(0);
+
+        entryLogger.flush();
+        bufferedLogChannel.appendLedgersMap();
+
+        // Allocate buffer to read (version, ledgersMapOffset, ledgerCount)
+        ByteBuf headers = ByteBufAllocator.DEFAULT.buffer(1024);
+        bufferedLogChannel.read(headers, 0);
+
+        // Skip marker string "BKLO"
+        headers.readInt();
+        headers.readInt();
+
+        headers.readLong();
+        int ledgersCount = headers.readInt();
+        long createTimestampFile = headers.readLong();
+
+        //verify headers
+        assertEquals(3, ledgersCount);
+        assertTrue(timestamp - createTimestampFile < 4);
+
+        //verify entry
+        assertEquals(0, generateEntry(0, 1).compareTo(entryLogger.readEntry(0, 1, entry0Position)));
+        assertEquals(0, generateEntry(1, 1).compareTo(entryLogger.readEntry(1, 1, entry1Position)));
+        assertEquals(0, generateEntry(2, 1).compareTo(entryLogger.readEntry(2, 1, entry2Position)));
+    }
 }