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 123ab14f17d..ba0286429f2 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 @@ -400,35 +400,43 @@ public void runWithFlags(boolean force, boolean suspendMajor, boolean suspendMin // enter major compaction LOG.info("Enter major compaction, suspendMajor {}", suspendMajor); majorCompacting.set(true); - doCompactEntryLogs(majorCompactionThreshold, majorCompactionMaxTimeMillis); - lastMajorCompactionTime = System.currentTimeMillis(); - // and also move minor compaction time - lastMinorCompactionTime = lastMajorCompactionTime; - gcStats.getMajorCompactionCounter().inc(); - majorCompacting.set(false); + try { + doCompactEntryLogs(majorCompactionThreshold, majorCompactionMaxTimeMillis); + } finally { + lastMajorCompactionTime = System.currentTimeMillis(); + // and also move minor compaction time + lastMinorCompactionTime = lastMajorCompactionTime; + gcStats.getMajorCompactionCounter().inc(); + majorCompacting.set(false); + } } else if (((isForceMinorCompactionAllow && force) || (enableMinorCompaction && (force || curTime - lastMinorCompactionTime > minorCompactionInterval))) && (!suspendMinor)) { // enter minor compaction LOG.info("Enter minor compaction, suspendMinor {}", suspendMinor); minorCompacting.set(true); - doCompactEntryLogs(minorCompactionThreshold, minorCompactionMaxTimeMillis); - lastMinorCompactionTime = System.currentTimeMillis(); - gcStats.getMinorCompactionCounter().inc(); - minorCompacting.set(false); + try { + doCompactEntryLogs(minorCompactionThreshold, minorCompactionMaxTimeMillis); + } finally { + lastMinorCompactionTime = System.currentTimeMillis(); + gcStats.getMinorCompactionCounter().inc(); + minorCompacting.set(false); + } } - + gcStats.getGcThreadRuntime().registerSuccessfulEvent( + MathUtils.nowInNano() - threadStart, TimeUnit.NANOSECONDS); + } catch (EntryLogMetadataMapException e) { + LOG.error("Error in entryLog-metadatamap, Failed to complete GC/Compaction due to entry-log {}", + e.getMessage(), e); + gcStats.getGcThreadRuntime().registerFailedEvent( + MathUtils.nowInNano() - threadStart, TimeUnit.NANOSECONDS); + } finally { if (force && forceGarbageCollection.compareAndSet(true, false)) { LOG.info("{} Set forceGarbageCollection to false after force GC to make it forceGC-able again.", Thread.currentThread().getName()); } - } catch (EntryLogMetadataMapException e) { - LOG.error("Error in entryLog-metadatamap, Failed to complete GC/Compaction due to entry-log {}", - e.getMessage(), e); } - gcStats.getGcThreadRuntime().registerSuccessfulEvent( - MathUtils.nowInNano() - threadStart, TimeUnit.NANOSECONDS); } /** diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/PersistentEntryLogMetadataMap.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/PersistentEntryLogMetadataMap.java index 95b365970f9..99f60691f55 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/PersistentEntryLogMetadataMap.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/PersistentEntryLogMetadataMap.java @@ -32,6 +32,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiConsumer; +import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.bookie.BookieException.EntryLogMetadataMapException; import org.apache.bookkeeper.bookie.EntryLogMetadata; import org.apache.bookkeeper.bookie.EntryLogMetadata.EntryLogMetadataRecyclable; @@ -39,15 +40,13 @@ import org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorage.CloseableIterator; import org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorageFactory.DbConfigType; import org.apache.bookkeeper.conf.ServerConfiguration; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Persistent entryLogMetadata-map that stores entry-loggers metadata into * rocksDB. */ +@Slf4j public class PersistentEntryLogMetadataMap implements EntryLogMetadataMap { - private static final Logger LOG = LoggerFactory.getLogger(PersistentEntryLogMetadataMap.class); // persistent Rocksdb to store metadata-map private final KeyValueStorage metadataMapDB; private AtomicBoolean isClosed = new AtomicBoolean(false); @@ -78,11 +77,11 @@ protected DataInputStream initialValue() { }; public PersistentEntryLogMetadataMap(String metadataPath, ServerConfiguration conf) throws IOException { - LOG.info("Loading persistent entrylog metadata-map from {}", metadataPath + "/" + METADATA_CACHE); + log.info("Loading persistent entrylog metadata-map from {}/{}", metadataPath, METADATA_CACHE); File dir = new File(metadataPath); if (!dir.mkdirs() && !dir.exists()) { String err = "Unable to create directory " + dir; - LOG.error(err); + log.error(err); throw new IOException(err); } metadataMapDB = KeyValueStorageRocksDB.factory.newKeyValueStorage(metadataPath, METADATA_CACHE, @@ -91,6 +90,7 @@ public PersistentEntryLogMetadataMap(String metadataPath, ServerConfiguration co @Override public boolean containsKey(long entryLogId) throws EntryLogMetadataMapException { + throwIfClosed(); LongWrapper key = LongWrapper.get(entryLogId); try { boolean isExist; @@ -107,6 +107,7 @@ public boolean containsKey(long entryLogId) throws EntryLogMetadataMapException @Override public void put(long entryLogId, EntryLogMetadata entryLogMeta) throws EntryLogMetadataMapException { + throwIfClosed(); LongWrapper key = LongWrapper.get(entryLogId); try { baos.get().reset(); @@ -114,7 +115,7 @@ public void put(long entryLogId, EntryLogMetadata entryLogMeta) throws EntryLogM entryLogMeta.serialize(dataos.get()); metadataMapDB.put(key.array, baos.get().toByteArray()); } catch (IllegalStateException | IOException e) { - LOG.error("Failed to serialize entrylog-metadata, entryLogId {}", entryLogId); + log.error("Failed to serialize entrylog-metadata, entryLogId {}", entryLogId); throw new EntryLogMetadataMapException(e); } } finally { @@ -129,6 +130,7 @@ public void put(long entryLogId, EntryLogMetadata entryLogMeta) throws EntryLogM */ @Override public void forEach(BiConsumer action) throws EntryLogMetadataMapException { + throwIfClosed(); CloseableIterator> iterator = metadataMapDB.iterator(); try { while (iterator.hasNext()) { @@ -158,19 +160,20 @@ public void forEach(BiConsumer action) throws EntryLogMe } } } catch (IOException e) { - LOG.error("Failed to iterate over entry-log metadata map {}", e.getMessage(), e); + log.error("Failed to iterate over entry-log metadata map {}", e.getMessage(), e); throw new EntryLogMetadataMapException(e); } finally { try { iterator.close(); } catch (IOException e) { - LOG.error("Failed to close entry-log metadata-map rocksDB iterator {}", e.getMessage(), e); + log.error("Failed to close entry-log metadata-map rocksDB iterator {}", e.getMessage(), e); } } } @Override public void remove(long entryLogId) throws EntryLogMetadataMapException { + throwIfClosed(); LongWrapper key = LongWrapper.get(entryLogId); try { try { @@ -185,6 +188,7 @@ public void remove(long entryLogId) throws EntryLogMetadataMapException { @Override public int size() throws EntryLogMetadataMapException { + throwIfClosed(); try { return (int) metadataMapDB.count(); } catch (IOException e) { @@ -194,8 +198,18 @@ public int size() throws EntryLogMetadataMapException { @Override public void close() throws IOException { - isClosed.set(true); - metadataMapDB.close(); + if (isClosed.compareAndSet(false, true)) { + metadataMapDB.close(); + } else { + log.warn("Attempted to close already closed PersistentEntryLogMetadataMap"); + } } + public void throwIfClosed() throws EntryLogMetadataMapException { + if (isClosed.get()) { + final String msg = "Attempted to use PersistentEntryLogMetadataMap after it was closed"; + log.error(msg); + throw new EntryLogMetadataMapException(new IOException(msg)); + } + } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByBytesTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByBytesWithMetadataCacheTest.java similarity index 85% rename from bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByBytesTest.java rename to bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByBytesWithMetadataCacheTest.java index 29303d742d5..8f22f625175 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByBytesTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByBytesWithMetadataCacheTest.java @@ -23,8 +23,8 @@ /** * Test compaction by bytes. */ -public class CompactionByBytesTest extends CompactionTest { - public CompactionByBytesTest() { - super(true); +public class CompactionByBytesWithMetadataCacheTest extends CompactionTest { + public CompactionByBytesWithMetadataCacheTest() { + super(true, true); } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByBytesWithoutMetadataCacheTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByBytesWithoutMetadataCacheTest.java new file mode 100644 index 00000000000..7b1418dab94 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByBytesWithoutMetadataCacheTest.java @@ -0,0 +1,30 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.bookkeeper.bookie; + +/** + * Test compaction by bytes. + */ +public class CompactionByBytesWithoutMetadataCacheTest extends CompactionTest { + public CompactionByBytesWithoutMetadataCacheTest() { + super(true, false); + } +} diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByEntriesTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByEntriesWithMetadataCacheTest.java similarity index 84% rename from bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByEntriesTest.java rename to bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByEntriesWithMetadataCacheTest.java index df871a3d497..86a1f3bd980 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByEntriesTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByEntriesWithMetadataCacheTest.java @@ -23,8 +23,8 @@ /** * Test compactions by entries. */ -public class CompactionByEntriesTest extends CompactionTest { - public CompactionByEntriesTest() { - super(false); +public class CompactionByEntriesWithMetadataCacheTest extends CompactionTest { + public CompactionByEntriesWithMetadataCacheTest() { + super(false, true); } } diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByEntriesWithoutMetadataCacheTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByEntriesWithoutMetadataCacheTest.java new file mode 100644 index 00000000000..2a02eb26b05 --- /dev/null +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionByEntriesWithoutMetadataCacheTest.java @@ -0,0 +1,30 @@ +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ +package org.apache.bookkeeper.bookie; + +/** + * Test compactions by entries. + */ +public class CompactionByEntriesWithoutMetadataCacheTest extends CompactionTest { + public CompactionByEntriesWithoutMetadataCacheTest() { + super(false, false); + } +} 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 index aef181f2482..2b35d7f0b53 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/CompactionTest.java @@ -104,11 +104,13 @@ public abstract class CompactionTest extends BookKeeperClusterTestCase { private final long minorCompactionInterval; private final long majorCompactionInterval; private final String msg; + private final boolean useMetadataCache; - public CompactionTest(boolean isByBytes) { + public CompactionTest(boolean isByBytes, boolean useMetadataCache) { super(NUM_BOOKIES); this.isThrottleByBytes = isByBytes; + this.useMetadataCache = useMetadataCache; this.digestType = DigestType.CRC32; this.passwdBytes = "".getBytes(); numEntries = 100; @@ -142,7 +144,7 @@ public void setUp() throws Exception { baseConf.setLedgerStorageClass(InterleavedLedgerStorage.class.getName()); baseConf.setIsThrottleByBytes(this.isThrottleByBytes); baseConf.setIsForceGCAllowWhenNoSpace(false); - baseConf.setGcEntryLogMetadataCacheEnabled(true); + baseConf.setGcEntryLogMetadataCacheEnabled(useMetadataCache); super.setUp(); } @@ -221,7 +223,9 @@ public void testDisableCompaction() throws Exception { assertFalse(getGCThread().enableMajorCompaction); assertFalse(getGCThread().enableMinorCompaction); getGCThread().triggerGC().get(); - assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + if (useMetadataCache) { + assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + } // after garbage collection, compaction should not be executed assertEquals(lastMinorCompactionTime, getGCThread().lastMinorCompactionTime); @@ -350,7 +354,9 @@ public void testMinorCompaction() throws Exception { getGCThread().enableForceGC(); getGCThread().triggerGC().get(); - assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + if (useMetadataCache) { + assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + } assertTrue( "ACTIVE_ENTRY_LOG_COUNT should have been updated", getStatsProvider(0) @@ -725,8 +731,9 @@ public void testMinorCompactionWithNoWritableLedgerDirs() throws Exception { LOG.info("Finished deleting the ledgers contains most entries."); getGCThread().triggerGC().get(); - assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); - + if (useMetadataCache) { + assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + } // after garbage collection, major compaction should not be executed assertEquals(lastMajorCompactionTime, getGCThread().lastMajorCompactionTime); assertEquals(lastMinorCompactionTime, getGCThread().lastMinorCompactionTime); @@ -775,8 +782,9 @@ public void testMinorCompactionWithNoWritableLedgerDirsButIsForceGCAllowWhenNoSp long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; assertFalse(getGCThread().enableMajorCompaction); assertTrue(getGCThread().enableMinorCompaction); - assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); - + if (useMetadataCache) { + assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + } for (int i = 0; i < bookieCount(); i++) { BookieImpl bookie = ((BookieImpl) serverByIndex(i).getBookie()); bookie.getLedgerStorage().flush(); @@ -859,8 +867,9 @@ public void testMajorCompaction() throws Exception { LOG.info("Finished deleting the ledgers contains most entries."); getGCThread().enableForceGC(); getGCThread().triggerGC().get(); - assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); - + if (useMetadataCache) { + assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + } // after garbage collection, minor compaction should not be executed assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); assertTrue(getGCThread().lastMajorCompactionTime > lastMajorCompactionTime); @@ -910,8 +919,9 @@ public void testForceMajorCompaction() throws Exception { LOG.info("Finished deleting the ledgers contains most entries."); getGCThread().enableForceGC(); getGCThread().triggerGC().get(); - assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); - + if (useMetadataCache) { + assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + } // after garbage collection, minor compaction should not be executed assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); assertTrue(getGCThread().lastMajorCompactionTime > lastMajorCompactionTime); @@ -1085,8 +1095,9 @@ public void testCompactionWhenLedgerDirsAreFull() throws Exception { LOG.info("Finished deleting the ledgers contains most entries."); getGCThread().enableForceGC(); getGCThread().triggerGC().get(); - assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); - + if (useMetadataCache) { + assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + } // after garbage collection, minor compaction should not be executed assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); assertTrue(getGCThread().lastMajorCompactionTime > lastMajorCompactionTime); @@ -1171,8 +1182,9 @@ public void testCompactionSmallEntryLogs() throws Exception { getGCThread().enableForceGC(); getGCThread().triggerGC().get(); - assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); - + if (useMetadataCache) { + assertTrue(getGCThread().getEntryLogMetaMap() instanceof PersistentEntryLogMetadataMap); + } // entry logs (0.log) should not be compacted // entry logs ([1,2,3].log) should be compacted. for (File ledgerDirectory : bookieLedgerDirs()) {