Merge pull request #21 from kishorvpatil/yahooGcThreadFix17

revans2 · web-flow · commit 5058f08287c5 · 2018-02-20T10:43:54.000-06:00
YBK-156 - Lower memory usage in GarbageCollectionThread while extracting all ledger meta data
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java
@@ -378,16 +378,7 @@ private void doGcEntryLogs() {
         // Loop through all of the entry logs and remove the non-active ledgers.
         for (Long entryLogId : entryLogMetaMap.keySet()) {
             EntryLogMetadata meta = entryLogMetaMap.get(entryLogId);
-            for (Long entryLogLedger : meta.getLedgersMap().keySet()) {
-                // Remove the entry log ledger from the set if it isn't active.
-                try {
-                    if (!ledgerStorage.ledgerExists(entryLogLedger)) {
-                        meta.removeLedger(entryLogLedger);
-                    }
-                } catch (IOException e) {
-                    LOG.error("Error reading from ledger storage", e);
-                }
-            }
+            removeIfLedgerNotExists(meta);
             if (meta.isEmpty()) {
                 // This means the entry log is not associated with any active ledgers anymore.
                 // We can remove this entry log file now.
@@ -397,6 +388,19 @@ private void doGcEntryLogs() {
         }
     }
 
+    private void removeIfLedgerNotExists(EntryLogMetadata meta) {
+        for (Long entryLogLedger : meta.getLedgersMap().keySet()) {
+            // Remove the entry log ledger from the set if it isn't active.
+            try {
+                if (!ledgerStorage.ledgerExists(entryLogLedger)) {
+                    meta.removeLedger(entryLogLedger);
+                }
+            } catch (IOException e) {
+                LOG.error("Error reading from ledger storage", e);
+            }
+        }
+    }
+
     /**
      * Compact entry logs if necessary.
      *
@@ -549,7 +553,12 @@ protected Map<Long, EntryLogMetadata> extractMetaFromEntryLogs(Map<Long, EntryLo
             try {
                 // Read through the entry log file and extract the entry log meta
                 EntryLogMetadata entryLogMeta = entryLogger.getEntryLogMetadata(entryLogId);
-                entryLogMetaMap.put(entryLogId, entryLogMeta);
+                removeIfLedgerNotExists(entryLogMeta);
+                if (entryLogMeta.isEmpty()) {
+                    entryLogger.removeEntryLog(entryLogId);
+                } else {
+                    entryLogMetaMap.put(entryLogId, entryLogMeta);
+                }
             } catch (IOException e) {
                 hasExceptionWhenScan = true;
                 LOG.warn("Premature exception when processing " + entryLogId +
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java
@@ -23,6 +23,8 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -568,6 +570,68 @@ public void checkpointComplete(Checkpoint checkpoint,
         storage.gcThread.doCompactEntryLogs(threshold);
     }
 
+    /**
+     * Test extractMetaFromEntryLogs optimized method to avoid excess memory usage.
+     */
+    @Test(timeout = 60000)
+    public void testExtractMetaFromEntryLogs() throws Exception {
+        ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+        File tmpDir = createTempDir("bkTest", ".dir");
+        File curDir = Bookie.getCurrentDirectory(tmpDir);
+        Bookie.checkDirectoryStructure(curDir);
+        conf.setLedgerDirNames(new String[] { tmpDir.toString() });
+
+        LedgerDirsManager dirs = new LedgerDirsManager(conf, conf.getLedgerDirs());
+        final Set<Long> ledgers = Collections
+            .newSetFromMap(new ConcurrentHashMap<Long, Boolean>());
+
+        LedgerManager manager = getLedgerManager(ledgers);
+
+        CheckpointSource checkpointSource = new CheckpointSource() {
+
+            @Override
+            public Checkpoint newCheckpoint() {
+                return null;
+            }
+
+            @Override
+            public void checkpointComplete(Checkpoint checkpoint,
+                                           boolean compact) throws IOException {
+            }
+        };
+        InterleavedLedgerStorage storage = new InterleavedLedgerStorage();
+        storage.initialize(conf, manager, dirs, dirs, checkpointSource, NullStatsLogger.INSTANCE);
+        final byte[] KEY = "foobar".getBytes();
+
+        for (long ledger = 0; ledger <= 10; ledger++) {
+            ledgers.add(ledger);
+            for(int entry = 1; entry <= 50; entry++) {
+                try {
+                    storage.addEntry(genEntry(ledger, entry, ENTRY_SIZE));
+                } catch (IOException e) {
+                    //ignore exception on failure to add entry.
+                }
+            };
+        };
+
+        storage.flush();
+        storage.shutdown();
+
+        storage = new InterleavedLedgerStorage();
+        storage.initialize(conf, manager, dirs, dirs, checkpointSource, NullStatsLogger.INSTANCE);
+
+        long startingEntriesCount = storage.gcThread.entryLogger.getLeastUnflushedLogId() - storage.gcThread.scannedLogId;
+        LOG.info("The old Log Entry count is: " + startingEntriesCount);
+
+        Map<Long, EntryLogMetadata> entryLogMetaData = new HashMap<>();
+        Map<Long, EntryLogMetadata> finalEntryLogMetadataMap = storage.gcThread.extractMetaFromEntryLogs(entryLogMetaData);
+        long finalEntriesCount = storage.gcThread.entryLogger.getLeastUnflushedLogId() - storage.gcThread.scannedLogId;
+        LOG.info("The latest Log Entry count is: " + finalEntriesCount);
+
+        assertTrue("The GC did not clean up entries...", startingEntriesCount != finalEntriesCount);
+        assertTrue("Entries Count is zero", finalEntriesCount == 0);
+    }
+
     private ByteBuffer genEntry(long ledger, long entry, int size) {
         ByteBuffer bb = ByteBuffer.wrap(new byte[size]);
         bb.putLong(ledger);
