From 46f64ff392f14ae6ab971f40334cd1efe78bdf44 Mon Sep 17 00:00:00 2001 From: chenhang Date: Thu, 10 Nov 2022 23:19:28 +0800 Subject: [PATCH 1/4] add gc small files check --- .../bookie/GarbageCollectorThread.java | 29 +++- .../bookkeeper/conf/ServerConfiguration.java | 34 +++++ .../bookie/GarbageCollectorThreadTest.java | 140 ++++++++++++++++++ conf/bk_server.conf | 7 + 4 files changed, 205 insertions(+), 5 deletions(-) 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 index d17450426e7..3b632df451e 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java @@ -546,14 +546,24 @@ private boolean removeIfLedgerNotExists(EntryLogMetadata meta) throws EntryLogMe @VisibleForTesting void doCompactEntryLogs(double threshold, long maxTimeMillis) throws EntryLogMetadataMapException { LOG.info("Do compaction to compact those files lower than {}", threshold); + double tmpGcEntryLogSizeRatio = conf.getGcEntryLogSizeRatio() <= 0 ? 0 : conf.getGcEntryLogSizeRatio(); + final double gcEntryLogSizeRatio = tmpGcEntryLogSizeRatio > 1 ? 1.0 : tmpGcEntryLogSizeRatio; + if (gcEntryLogSizeRatio > 0.5) { + LOG.warn("Configured gcEntryLogSizeRatio: {}, updated gcEntryLogSizeRatio: {} gratter than 0.5, " + + "which means any entryLog file size less than {} will be compacted and it will bring heavy pressure " + + "on garbage collection.", conf.getGcEntryLogSizeRatio(), gcEntryLogSizeRatio, + gcEntryLogSizeRatio * conf.getEntryLogSizeLimit()); + } final int numBuckets = 10; int[] entryLogUsageBuckets = new int[numBuckets]; int[] compactedBuckets = new int[numBuckets]; ArrayList> compactableBuckets = new ArrayList<>(numBuckets); + ArrayList> smallFilesCompactableBuckets = new ArrayList<>(numBuckets); for (int i = 0; i < numBuckets; i++) { compactableBuckets.add(new LinkedList<>()); + smallFilesCompactableBuckets.add(new LinkedList<>()); } long start = System.currentTimeMillis(); @@ -568,14 +578,20 @@ void doCompactEntryLogs(double threshold, long maxTimeMillis) throws EntryLogMet end.setValue(System.currentTimeMillis()); timeDiff.setValue(end.getValue() - start); } - if (meta.getUsage() >= threshold || (maxTimeMillis > 0 && timeDiff.getValue() >= maxTimeMillis) - || !running) { + if ((meta.getUsage() >= threshold + && meta.getTotalSize() >= conf.getEntryLogSizeLimit() * gcEntryLogSizeRatio) + || (maxTimeMillis > 0 && timeDiff.getValue() >= maxTimeMillis) + || !running) { // We allow the usage limit calculation to continue so that we get an accurate // report of where the usage was prior to running compaction. return; } - compactableBuckets.get(bucketIndex).add(meta.getEntryLogId()); + if (meta.getTotalSize() < conf.getEntryLogSizeLimit() * gcEntryLogSizeRatio) { + smallFilesCompactableBuckets.get(bucketIndex).add(meta.getEntryLogId()); + } else { + compactableBuckets.get(bucketIndex).add(meta.getEntryLogId()); + } }); LOG.info( @@ -584,8 +600,11 @@ void doCompactEntryLogs(double threshold, long maxTimeMillis) throws EntryLogMet final int maxBucket = calculateUsageIndex(numBuckets, threshold); stopCompaction: - for (int currBucket = 0; currBucket <= maxBucket; currBucket++) { - LinkedList entryLogIds = compactableBuckets.get(currBucket); + for (int currBucket = 0; currBucket < numBuckets; currBucket++) { + LinkedList entryLogIds = smallFilesCompactableBuckets.get(currBucket); + if (currBucket <= maxBucket) { + entryLogIds.addAll(compactableBuckets.get(currBucket)); + } while (!entryLogIds.isEmpty()) { if (timeDiff.getValue() < maxTimeMillis) { end.setValue(System.currentTimeMillis()); 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 3fb8d18ef9c..8c0af95cba4 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 @@ -116,6 +116,7 @@ public class ServerConfiguration extends AbstractConfiguration 0.0 * 1GB, we will skip the entry log file garbage collection + * case 2: GC_ENTRYLOG_SIZE_RATIO = 0.5, 1MB < 1.0 * 1GB, we will compact the entry log file + * case 3: GC_ENTRYLOG_SIZE_RATIO = 1.0, 1MB < 1.0 * 1GB, we will compact the entry log file + * + * If GC_ENTRYLOG_SIZE_RATIO <= 0.0, it means disable the entry log file size check. + * By default, the GC_ENTRYLOG_SIZE_RATIO = 0.0. + * + * @return the gcEntryLogSizeRatio, default is 1.0. + */ + public double getGcEntryLogSizeRatio() { + return getDouble(GC_ENTRYLOG_SIZE_RATIO, 0.0); + } + + public ServerConfiguration setGcEntryLogSizeRatio(double gcEntryLogSizeRatio) { + this.setProperty(GC_ENTRYLOG_SIZE_RATIO, gcEntryLogSizeRatio); + return this; + } + /** * Get whether local scrub is enabled. * diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java index 31828423a43..3d66e271451 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java @@ -30,6 +30,7 @@ import static org.hamcrest.Matchers.empty; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.greaterThan; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.any; @@ -238,4 +239,143 @@ private void testExtractMetaFromEntryLogs(EntryLogger entryLogger, File ledgerDi assertTrue(entryLogMetaMap.isEmpty()); assertFalse(entryLogger.logExists(logId3)); } + + @Test + public void testCompactionWithFileSizeCheck() throws Exception { + File ledgerDir = tmpDirs.createNew("testFileSize", "ledgers"); + EntryLogger entryLogger = newLegacyEntryLogger(20000, ledgerDir); + + MockLedgerStorage storage = new MockLedgerStorage(); + MockLedgerManager lm = new MockLedgerManager(); + + GarbageCollectorThread gcThread = new GarbageCollectorThread( + TestBKConfiguration.newServerConfiguration().setGcEntryLogSizeRatio(0.5), lm, + newDirsManager(ledgerDir), + storage, entryLogger, NullStatsLogger.INSTANCE); + + // Add entries. + // Ledger 1 is on first entry log + // Ledger 2 spans first, second and third entry log + // Ledger 3 is on the third entry log (which is still active when extract meta) + long loc1 = entryLogger.addEntry(1L, makeEntry(1L, 1L, 5000)); + long loc2 = entryLogger.addEntry(2L, makeEntry(2L, 1L, 5000)); + assertThat(logIdFromLocation(loc2), equalTo(logIdFromLocation(loc1))); + long loc3 = entryLogger.addEntry(2L, makeEntry(2L, 2L, 15000)); + assertThat(logIdFromLocation(loc3), greaterThan(logIdFromLocation(loc2))); + long loc4 = entryLogger.addEntry(2L, makeEntry(2L, 3L, 15000)); + assertThat(logIdFromLocation(loc4), greaterThan(logIdFromLocation(loc3))); + long loc5 = entryLogger.addEntry(3L, makeEntry(3L, 1L, 1000)); + assertThat(logIdFromLocation(loc5), equalTo(logIdFromLocation(loc4))); + + long logId1 = logIdFromLocation(loc2); + long logId2 = logIdFromLocation(loc3); + long logId3 = logIdFromLocation(loc5); + entryLogger.flush(); + + storage.setMasterKey(1L, new byte[0]); + storage.setMasterKey(2L, new byte[0]); + storage.setMasterKey(3L, new byte[0]); + + assertThat(entryLogger.getFlushedLogIds(), containsInAnyOrder(logId1, logId2)); + assertTrue(entryLogger.logExists(logId1)); + assertTrue(entryLogger.logExists(logId2)); + assertTrue(entryLogger.logExists(logId3)); + + // all ledgers exist, nothing should disappear + final EntryLogMetadataMap entryLogMetaMap = gcThread.getEntryLogMetaMap(); + gcThread.extractMetaFromEntryLogs(); + + assertThat(entryLogger.getFlushedLogIds(), containsInAnyOrder(logId1, logId2)); + assertTrue(entryLogMetaMap.containsKey(logId1)); + assertTrue(entryLogMetaMap.containsKey(logId2)); + assertTrue(entryLogger.logExists(logId3)); + + gcThread.runWithFlags(true, true, false); + + // logId1 and logId2 should be compacted + assertFalse(entryLogger.logExists(logId1)); + assertFalse(entryLogger.logExists(logId2)); + assertTrue(entryLogger.logExists(logId3)); + assertFalse(entryLogMetaMap.containsKey(logId1)); + assertFalse(entryLogMetaMap.containsKey(logId2)); + + assertEquals(3, storage.getUpdatedLocations().size()); + + EntryLocation location1 = storage.getUpdatedLocations().get(0); + assertEquals(1, location1.getLedger()); + assertEquals(1, location1.getEntry()); + assertThat(logIdFromLocation(location1.getLocation()), greaterThan((int) logId3)); + + EntryLocation location2 = storage.getUpdatedLocations().get(1); + assertEquals(2, location2.getLedger()); + assertEquals(1, location2.getEntry()); + assertEquals(logIdFromLocation(location2.getLocation()), logIdFromLocation(location1.getLocation())); + + EntryLocation location3 = storage.getUpdatedLocations().get(2); + assertEquals(2, location3.getLedger()); + assertEquals(2, location3.getEntry()); + assertThat(logIdFromLocation(loc4), greaterThan(logIdFromLocation(loc3))); + } + + @Test + public void testCompactionWithoutFileSizeCheck() throws Exception { + File ledgerDir = tmpDirs.createNew("testFileSize", "ledgers"); + EntryLogger entryLogger = newLegacyEntryLogger(20000, ledgerDir); + + MockLedgerStorage storage = new MockLedgerStorage(); + MockLedgerManager lm = new MockLedgerManager(); + + GarbageCollectorThread gcThread = new GarbageCollectorThread( + TestBKConfiguration.newServerConfiguration(), lm, + newDirsManager(ledgerDir), + storage, entryLogger, NullStatsLogger.INSTANCE); + + // Add entries. + // Ledger 1 is on first entry log + // Ledger 2 spans first, second and third entry log + // Ledger 3 is on the third entry log (which is still active when extract meta) + long loc1 = entryLogger.addEntry(1L, makeEntry(1L, 1L, 5000)); + long loc2 = entryLogger.addEntry(2L, makeEntry(2L, 1L, 5000)); + assertThat(logIdFromLocation(loc2), equalTo(logIdFromLocation(loc1))); + long loc3 = entryLogger.addEntry(2L, makeEntry(2L, 2L, 15000)); + assertThat(logIdFromLocation(loc3), greaterThan(logIdFromLocation(loc2))); + long loc4 = entryLogger.addEntry(2L, makeEntry(2L, 3L, 15000)); + assertThat(logIdFromLocation(loc4), greaterThan(logIdFromLocation(loc3))); + long loc5 = entryLogger.addEntry(3L, makeEntry(3L, 1L, 1000)); + assertThat(logIdFromLocation(loc5), equalTo(logIdFromLocation(loc4))); + + long logId1 = logIdFromLocation(loc2); + long logId2 = logIdFromLocation(loc3); + long logId3 = logIdFromLocation(loc5); + entryLogger.flush(); + + storage.setMasterKey(1L, new byte[0]); + storage.setMasterKey(2L, new byte[0]); + storage.setMasterKey(3L, new byte[0]); + + assertThat(entryLogger.getFlushedLogIds(), containsInAnyOrder(logId1, logId2)); + assertTrue(entryLogger.logExists(logId1)); + assertTrue(entryLogger.logExists(logId2)); + assertTrue(entryLogger.logExists(logId3)); + + // all ledgers exist, nothing should disappear + final EntryLogMetadataMap entryLogMetaMap = gcThread.getEntryLogMetaMap(); + gcThread.extractMetaFromEntryLogs(); + + assertThat(entryLogger.getFlushedLogIds(), containsInAnyOrder(logId1, logId2)); + assertTrue(entryLogMetaMap.containsKey(logId1)); + assertTrue(entryLogMetaMap.containsKey(logId2)); + assertTrue(entryLogger.logExists(logId3)); + + gcThread.runWithFlags(true, true, false); + + assertTrue(entryLogger.logExists(logId1)); + assertTrue(entryLogger.logExists(logId2)); + assertTrue(entryLogger.logExists(logId3)); + assertTrue(entryLogMetaMap.containsKey(logId1)); + assertTrue(entryLogMetaMap.containsKey(logId2)); + + assertEquals(0, storage.getUpdatedLocations().size()); + } + } diff --git a/conf/bk_server.conf b/conf/bk_server.conf index cef24eaad15..946b502d094 100755 --- a/conf/bk_server.conf +++ b/conf/bk_server.conf @@ -621,6 +621,13 @@ gcEntryLogMetadataCacheEnabled=false # name "entrylogIndexCache"] # gcEntryLogMetadataCachePath= +# When the current entry log total size less than gcEntryLogSizeRatio * logSizeLimit, +# and the remaining usage less than the threshold, this entry log file will be compacted. +# This configuration is aiming to reduce small entry log files when using TransactionalEntryLogCompactor. +# If gcEntryLogSizeRatio <= 0.0, it means disable the entry log file size check. +# By default, the entry log file size check is disabled. +# gcEntryLogSizeRatio = 0.0 + ############################################################################# ## Disk utilization ############################################################################# From e8f21c14c865839b1f01352b07459d6a58b45bd5 Mon Sep 17 00:00:00 2001 From: chenhang Date: Thu, 10 Nov 2022 23:30:34 +0800 Subject: [PATCH 2/4] format code --- .../java/org/apache/bookkeeper/conf/ServerConfiguration.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 8c0af95cba4..c826f9156a5 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 @@ -558,8 +558,8 @@ public ServerConfiguration setGcEntryLogMetadataCachePath(String gcEntrylogMetad /** * Get the gc entry log size ratio. * When the current entry log total size less than GC_ENTRYLOG_SIZE_RATIO * ENTRY_LOG_SIZE_LIMIT, - * and the remaining usage less than the threshold, this entry log file will be compacted. - * This configuration is aiming to reduce small entry log files when using TransactionalEntryLogCompactor. + * this entry log file will be compacted. This configuration is aiming to reduce small entry log files + * when using TransactionalEntryLogCompactor. * * For example, * - the current entry log file total size is 1MB From d6a2ccc0a64ecc6d4e10c83d551953ad929abaa3 Mon Sep 17 00:00:00 2001 From: chenhang Date: Mon, 13 Mar 2023 15:56:45 +0800 Subject: [PATCH 3/4] address comments --- .../bookie/GarbageCollectorThread.java | 34 ++++++------------- .../bookkeeper/conf/ServerConfiguration.java | 34 +++---------------- .../bookkeeper/proto/BookieNettyServer.java | 1 + .../bookie/GarbageCollectorThreadTest.java | 24 +++++-------- 4 files changed, 24 insertions(+), 69 deletions(-) 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 index 3b632df451e..890f9585ce3 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java @@ -546,24 +546,14 @@ private boolean removeIfLedgerNotExists(EntryLogMetadata meta) throws EntryLogMe @VisibleForTesting void doCompactEntryLogs(double threshold, long maxTimeMillis) throws EntryLogMetadataMapException { LOG.info("Do compaction to compact those files lower than {}", threshold); - double tmpGcEntryLogSizeRatio = conf.getGcEntryLogSizeRatio() <= 0 ? 0 : conf.getGcEntryLogSizeRatio(); - final double gcEntryLogSizeRatio = tmpGcEntryLogSizeRatio > 1 ? 1.0 : tmpGcEntryLogSizeRatio; - if (gcEntryLogSizeRatio > 0.5) { - LOG.warn("Configured gcEntryLogSizeRatio: {}, updated gcEntryLogSizeRatio: {} gratter than 0.5, " - + "which means any entryLog file size less than {} will be compacted and it will bring heavy pressure " - + "on garbage collection.", conf.getGcEntryLogSizeRatio(), gcEntryLogSizeRatio, - gcEntryLogSizeRatio * conf.getEntryLogSizeLimit()); - } final int numBuckets = 10; int[] entryLogUsageBuckets = new int[numBuckets]; int[] compactedBuckets = new int[numBuckets]; ArrayList> compactableBuckets = new ArrayList<>(numBuckets); - ArrayList> smallFilesCompactableBuckets = new ArrayList<>(numBuckets); for (int i = 0; i < numBuckets; i++) { compactableBuckets.add(new LinkedList<>()); - smallFilesCompactableBuckets.add(new LinkedList<>()); } long start = System.currentTimeMillis(); @@ -571,27 +561,26 @@ void doCompactEntryLogs(double threshold, long maxTimeMillis) throws EntryLogMet MutableLong timeDiff = new MutableLong(0); entryLogMetaMap.forEach((entryLogId, meta) -> { - int bucketIndex = calculateUsageIndex(numBuckets, meta.getUsage()); + double usage = meta.getUsage(); + if (conf.isUseTargetEntryLogSizeForGc() && usage < 1.0d) { + usage = (double) meta.getRemainingSize() / Math.max(meta.getTotalSize(), conf.getEntryLogSizeLimit()); + } + int bucketIndex = calculateUsageIndex(numBuckets, usage); entryLogUsageBuckets[bucketIndex]++; if (timeDiff.getValue() < maxTimeMillis) { end.setValue(System.currentTimeMillis()); timeDiff.setValue(end.getValue() - start); } - if ((meta.getUsage() >= threshold - && meta.getTotalSize() >= conf.getEntryLogSizeLimit() * gcEntryLogSizeRatio) + if ((usage >= threshold || (maxTimeMillis > 0 && timeDiff.getValue() >= maxTimeMillis) - || !running) { + || !running)) { // We allow the usage limit calculation to continue so that we get an accurate // report of where the usage was prior to running compaction. return; } - if (meta.getTotalSize() < conf.getEntryLogSizeLimit() * gcEntryLogSizeRatio) { - smallFilesCompactableBuckets.get(bucketIndex).add(meta.getEntryLogId()); - } else { - compactableBuckets.get(bucketIndex).add(meta.getEntryLogId()); - } + compactableBuckets.get(bucketIndex).add(meta.getEntryLogId()); }); LOG.info( @@ -600,11 +589,8 @@ void doCompactEntryLogs(double threshold, long maxTimeMillis) throws EntryLogMet final int maxBucket = calculateUsageIndex(numBuckets, threshold); stopCompaction: - for (int currBucket = 0; currBucket < numBuckets; currBucket++) { - LinkedList entryLogIds = smallFilesCompactableBuckets.get(currBucket); - if (currBucket <= maxBucket) { - entryLogIds.addAll(compactableBuckets.get(currBucket)); - } + for (int currBucket = 0; currBucket < maxBucket; currBucket++) { + LinkedList entryLogIds = compactableBuckets.get(currBucket); while (!entryLogIds.isEmpty()) { if (timeDiff.getValue() < maxTimeMillis) { end.setValue(System.currentTimeMillis()); 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 c826f9156a5..4db536ab495 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 @@ -116,7 +116,7 @@ public class ServerConfiguration extends AbstractConfiguration 0.0 * 1GB, we will skip the entry log file garbage collection - * case 2: GC_ENTRYLOG_SIZE_RATIO = 0.5, 1MB < 1.0 * 1GB, we will compact the entry log file - * case 3: GC_ENTRYLOG_SIZE_RATIO = 1.0, 1MB < 1.0 * 1GB, we will compact the entry log file - * - * If GC_ENTRYLOG_SIZE_RATIO <= 0.0, it means disable the entry log file size check. - * By default, the GC_ENTRYLOG_SIZE_RATIO = 0.0. - * - * @return the gcEntryLogSizeRatio, default is 1.0. - */ - public double getGcEntryLogSizeRatio() { - return getDouble(GC_ENTRYLOG_SIZE_RATIO, 0.0); + public boolean isUseTargetEntryLogSizeForGc() { + return getBoolean(USE_TARGET_ENTRYLOG_SIZE_FOR_GC, false); } - public ServerConfiguration setGcEntryLogSizeRatio(double gcEntryLogSizeRatio) { - this.setProperty(GC_ENTRYLOG_SIZE_RATIO, gcEntryLogSizeRatio); + public ServerConfiguration setUseTargetEntryLogSizeForGc(boolean useTargetEntryLogSizeForGc) { + this.setProperty(USE_TARGET_ENTRYLOG_SIZE_FOR_GC, useTargetEntryLogSizeForGc); return this; } diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java index eeffcac0d97..bb63dade05a 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java @@ -308,6 +308,7 @@ private void listenOn(InetSocketAddress address, BookieSocketAddress bookieAddre ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.option(ChannelOption.ALLOCATOR, allocator); bootstrap.childOption(ChannelOption.ALLOCATOR, allocator); + bootstrap.childOption(ChannelOption.MAX_MESSAGES_PER_READ, 100); bootstrap.group(acceptorGroup, eventLoopGroup); bootstrap.childOption(ChannelOption.TCP_NODELAY, conf.getServerTcpNoDelay()); bootstrap.childOption(ChannelOption.SO_LINGER, conf.getServerSockLinger()); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java index 3d66e271451..fe7f650e389 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java @@ -249,7 +249,7 @@ public void testCompactionWithFileSizeCheck() throws Exception { MockLedgerManager lm = new MockLedgerManager(); GarbageCollectorThread gcThread = new GarbageCollectorThread( - TestBKConfiguration.newServerConfiguration().setGcEntryLogSizeRatio(0.5), lm, + TestBKConfiguration.newServerConfiguration().setUseTargetEntryLogSizeForGc(true), lm, newDirsManager(ledgerDir), storage, entryLogger, NullStatsLogger.INSTANCE); @@ -290,31 +290,23 @@ public void testCompactionWithFileSizeCheck() throws Exception { assertTrue(entryLogMetaMap.containsKey(logId2)); assertTrue(entryLogger.logExists(logId3)); + storage.deleteLedger(1); + // only logId 1 will be compacted. gcThread.runWithFlags(true, true, false); // logId1 and logId2 should be compacted assertFalse(entryLogger.logExists(logId1)); - assertFalse(entryLogger.logExists(logId2)); + assertTrue(entryLogger.logExists(logId2)); assertTrue(entryLogger.logExists(logId3)); assertFalse(entryLogMetaMap.containsKey(logId1)); - assertFalse(entryLogMetaMap.containsKey(logId2)); - - assertEquals(3, storage.getUpdatedLocations().size()); + assertTrue(entryLogMetaMap.containsKey(logId2)); - EntryLocation location1 = storage.getUpdatedLocations().get(0); - assertEquals(1, location1.getLedger()); - assertEquals(1, location1.getEntry()); - assertThat(logIdFromLocation(location1.getLocation()), greaterThan((int) logId3)); + assertEquals(1, storage.getUpdatedLocations().size()); - EntryLocation location2 = storage.getUpdatedLocations().get(1); + EntryLocation location2 = storage.getUpdatedLocations().get(0); assertEquals(2, location2.getLedger()); assertEquals(1, location2.getEntry()); - assertEquals(logIdFromLocation(location2.getLocation()), logIdFromLocation(location1.getLocation())); - - EntryLocation location3 = storage.getUpdatedLocations().get(2); - assertEquals(2, location3.getLedger()); - assertEquals(2, location3.getEntry()); - assertThat(logIdFromLocation(loc4), greaterThan(logIdFromLocation(loc3))); + assertEquals(logIdFromLocation(location2.getLocation()), logId3 + 1); } @Test From c48bf69463ddfd7ce33e0f2a65e03410e0e1d110 Mon Sep 17 00:00:00 2001 From: chenhang Date: Mon, 13 Mar 2023 16:19:29 +0800 Subject: [PATCH 4/4] format code --- .../bookkeeper/bookie/GarbageCollectorThread.java | 2 +- .../apache/bookkeeper/proto/BookieNettyServer.java | 1 - .../bookie/GarbageCollectorThreadTest.java | 9 ++++++--- conf/bk_server.conf | 14 ++++++++------ 4 files changed, 15 insertions(+), 11 deletions(-) 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 index 890f9585ce3..1420c5ca6f1 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java @@ -589,7 +589,7 @@ void doCompactEntryLogs(double threshold, long maxTimeMillis) throws EntryLogMet final int maxBucket = calculateUsageIndex(numBuckets, threshold); stopCompaction: - for (int currBucket = 0; currBucket < maxBucket; currBucket++) { + for (int currBucket = 0; currBucket <= maxBucket; currBucket++) { LinkedList entryLogIds = compactableBuckets.get(currBucket); while (!entryLogIds.isEmpty()) { if (timeDiff.getValue() < maxTimeMillis) { diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java index bb63dade05a..eeffcac0d97 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java @@ -308,7 +308,6 @@ private void listenOn(InetSocketAddress address, BookieSocketAddress bookieAddre ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.option(ChannelOption.ALLOCATOR, allocator); bootstrap.childOption(ChannelOption.ALLOCATOR, allocator); - bootstrap.childOption(ChannelOption.MAX_MESSAGES_PER_READ, 100); bootstrap.group(acceptorGroup, eventLoopGroup); bootstrap.childOption(ChannelOption.TCP_NODELAY, conf.getServerTcpNoDelay()); bootstrap.childOption(ChannelOption.SO_LINGER, conf.getServerSockLinger()); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java index fe7f650e389..ebe07ce2303 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/GarbageCollectorThreadTest.java @@ -266,26 +266,29 @@ public void testCompactionWithFileSizeCheck() throws Exception { assertThat(logIdFromLocation(loc4), greaterThan(logIdFromLocation(loc3))); long loc5 = entryLogger.addEntry(3L, makeEntry(3L, 1L, 1000)); assertThat(logIdFromLocation(loc5), equalTo(logIdFromLocation(loc4))); + long loc6 = entryLogger.addEntry(3L, makeEntry(3L, 2L, 5000)); long logId1 = logIdFromLocation(loc2); long logId2 = logIdFromLocation(loc3); long logId3 = logIdFromLocation(loc5); + long logId4 = logIdFromLocation(loc6); entryLogger.flush(); storage.setMasterKey(1L, new byte[0]); storage.setMasterKey(2L, new byte[0]); storage.setMasterKey(3L, new byte[0]); - assertThat(entryLogger.getFlushedLogIds(), containsInAnyOrder(logId1, logId2)); + assertThat(entryLogger.getFlushedLogIds(), containsInAnyOrder(logId1, logId2, logId3)); assertTrue(entryLogger.logExists(logId1)); assertTrue(entryLogger.logExists(logId2)); assertTrue(entryLogger.logExists(logId3)); + assertTrue(entryLogger.logExists(logId4)); // all ledgers exist, nothing should disappear final EntryLogMetadataMap entryLogMetaMap = gcThread.getEntryLogMetaMap(); gcThread.extractMetaFromEntryLogs(); - assertThat(entryLogger.getFlushedLogIds(), containsInAnyOrder(logId1, logId2)); + assertThat(entryLogger.getFlushedLogIds(), containsInAnyOrder(logId1, logId2, logId3)); assertTrue(entryLogMetaMap.containsKey(logId1)); assertTrue(entryLogMetaMap.containsKey(logId2)); assertTrue(entryLogger.logExists(logId3)); @@ -306,7 +309,7 @@ public void testCompactionWithFileSizeCheck() throws Exception { EntryLocation location2 = storage.getUpdatedLocations().get(0); assertEquals(2, location2.getLedger()); assertEquals(1, location2.getEntry()); - assertEquals(logIdFromLocation(location2.getLocation()), logId3 + 1); + assertEquals(logIdFromLocation(location2.getLocation()), logId4); } @Test diff --git a/conf/bk_server.conf b/conf/bk_server.conf index 946b502d094..812c5b072d0 100755 --- a/conf/bk_server.conf +++ b/conf/bk_server.conf @@ -621,12 +621,14 @@ gcEntryLogMetadataCacheEnabled=false # name "entrylogIndexCache"] # gcEntryLogMetadataCachePath= -# When the current entry log total size less than gcEntryLogSizeRatio * logSizeLimit, -# and the remaining usage less than the threshold, this entry log file will be compacted. -# This configuration is aiming to reduce small entry log files when using TransactionalEntryLogCompactor. -# If gcEntryLogSizeRatio <= 0.0, it means disable the entry log file size check. -# By default, the entry log file size check is disabled. -# gcEntryLogSizeRatio = 0.0 +# When judging whether an entry log file need to be compacted, we calculate the usage rate of the entry log file based +# on the actual size of the entry log file. However, if an entry log file is 1MB in size and 0.9MB of data is +# being used, this entry log file won't be compacted by garbage collector due to the high usage ratio, +# which will result in many small entry log files. +# We introduced the parameter `useTargetEntryLogSizeForGc` to determine whether to calculate entry log file usage +# based on the configured target entry log file size, which is configured by `logSizeLimit`. +# Default: useTargetEntryLogSizeForGc is false. +# useTargetEntryLogSizeForGc=false ############################################################################# ## Disk utilization