| /* |
| * |
| * 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; |
| |
| import static org.apache.bookkeeper.bookie.BookKeeperServerStats.ACTIVE_ENTRY_LOG_COUNT; |
| import static org.apache.bookkeeper.bookie.BookKeeperServerStats.ACTIVE_ENTRY_LOG_SPACE_BYTES; |
| import static org.apache.bookkeeper.bookie.BookKeeperServerStats.MAJOR_COMPACTION_COUNT; |
| import static org.apache.bookkeeper.bookie.BookKeeperServerStats.MINOR_COMPACTION_COUNT; |
| import static org.apache.bookkeeper.bookie.BookKeeperServerStats.RECLAIMED_COMPACTION_SPACE_BYTES; |
| import static org.apache.bookkeeper.bookie.BookKeeperServerStats.RECLAIMED_DELETION_SPACE_BYTES; |
| import static org.apache.bookkeeper.bookie.BookKeeperServerStats.THREAD_RUNTIME; |
| import static org.apache.bookkeeper.bookie.TransactionalEntryLogCompactor.COMPACTED_SUFFIX; |
| import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithLedgerManagerFactory; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| import com.google.common.util.concurrent.UncheckedExecutionException; |
| import io.netty.buffer.ByteBuf; |
| import io.netty.buffer.Unpooled; |
| import io.netty.buffer.UnpooledByteBufAllocator; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.Enumeration; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException; |
| import org.apache.bookkeeper.client.BookKeeper; |
| import org.apache.bookkeeper.client.BookKeeper.DigestType; |
| import org.apache.bookkeeper.client.LedgerEntry; |
| import org.apache.bookkeeper.client.LedgerHandle; |
| import org.apache.bookkeeper.client.api.LedgerMetadata; |
| import org.apache.bookkeeper.conf.ServerConfiguration; |
| import org.apache.bookkeeper.conf.TestBKConfiguration; |
| import org.apache.bookkeeper.meta.LedgerManager; |
| import org.apache.bookkeeper.proto.BookieServer; |
| import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener; |
| import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.Processor; |
| import org.apache.bookkeeper.proto.checksum.DigestManager; |
| import org.apache.bookkeeper.stats.NullStatsLogger; |
| import org.apache.bookkeeper.test.BookKeeperClusterTestCase; |
| import org.apache.bookkeeper.test.TestStatsProvider; |
| import org.apache.bookkeeper.util.DiskChecker; |
| import org.apache.bookkeeper.util.HardLink; |
| import org.apache.bookkeeper.util.TestUtils; |
| import org.apache.bookkeeper.versioning.Version; |
| import org.apache.bookkeeper.versioning.Versioned; |
| import org.apache.zookeeper.AsyncCallback; |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * This class tests the entry log compaction functionality. |
| */ |
| public abstract class CompactionTest extends BookKeeperClusterTestCase { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(CompactionTest.class); |
| |
| private static final int ENTRY_SIZE = 1024; |
| private static final int NUM_BOOKIES = 1; |
| |
| private final boolean isThrottleByBytes; |
| private final DigestType digestType; |
| private final byte[] passwdBytes; |
| private final int numEntries; |
| private final int gcWaitTime; |
| private final double minorCompactionThreshold; |
| private final double majorCompactionThreshold; |
| private final long minorCompactionInterval; |
| private final long majorCompactionInterval; |
| private final String msg; |
| |
| public CompactionTest(boolean isByBytes) { |
| super(NUM_BOOKIES); |
| |
| this.isThrottleByBytes = isByBytes; |
| this.digestType = DigestType.CRC32; |
| this.passwdBytes = "".getBytes(); |
| numEntries = 100; |
| gcWaitTime = 1000; |
| minorCompactionThreshold = 0.1f; |
| majorCompactionThreshold = 0.5f; |
| minorCompactionInterval = 2 * gcWaitTime / 1000; |
| majorCompactionInterval = 4 * gcWaitTime / 1000; |
| |
| // a dummy message |
| StringBuilder msgSB = new StringBuilder(); |
| for (int i = 0; i < ENTRY_SIZE; i++) { |
| msgSB.append("a"); |
| } |
| msg = msgSB.toString(); |
| } |
| |
| @Before |
| @Override |
| public void setUp() throws Exception { |
| // Set up the configuration properties needed. |
| baseConf.setEntryLogSizeLimit(numEntries * ENTRY_SIZE); |
| // Disable skip list for compaction |
| baseConf.setGcWaitTime(gcWaitTime); |
| baseConf.setFlushInterval(100); |
| baseConf.setMinorCompactionThreshold(minorCompactionThreshold); |
| baseConf.setMajorCompactionThreshold(majorCompactionThreshold); |
| baseConf.setMinorCompactionInterval(minorCompactionInterval); |
| baseConf.setMajorCompactionInterval(majorCompactionInterval); |
| baseConf.setEntryLogFilePreAllocationEnabled(false); |
| baseConf.setLedgerStorageClass(InterleavedLedgerStorage.class.getName()); |
| baseConf.setIsThrottleByBytes(this.isThrottleByBytes); |
| baseConf.setIsForceGCAllowWhenNoSpace(false); |
| |
| super.setUp(); |
| } |
| |
| private GarbageCollectorThread getGCThread() { |
| assertEquals(1, bs.size()); |
| BookieServer server = bs.get(0); |
| return ((InterleavedLedgerStorage) server.getBookie().getLedgerStorage()).gcThread; |
| } |
| |
| LedgerHandle[] prepareData(int numEntryLogs, boolean changeNum) |
| throws Exception { |
| // since an entry log file can hold at most 100 entries |
| // first ledger write 2 entries, which is less than low water mark |
| int num1 = 2; |
| // third ledger write more than high water mark entries |
| int num3 = (int) (numEntries * 0.7f); |
| // second ledger write remaining entries, which is higher than low water mark |
| // and less than high water mark |
| int num2 = numEntries - num3 - num1; |
| |
| LedgerHandle[] lhs = new LedgerHandle[3]; |
| for (int i = 0; i < 3; ++i) { |
| lhs[i] = bkc.createLedger(NUM_BOOKIES, NUM_BOOKIES, digestType, passwdBytes); |
| } |
| |
| for (int n = 0; n < numEntryLogs; n++) { |
| for (int k = 0; k < num1; k++) { |
| lhs[0].addEntry(msg.getBytes()); |
| } |
| for (int k = 0; k < num2; k++) { |
| lhs[1].addEntry(msg.getBytes()); |
| } |
| for (int k = 0; k < num3; k++) { |
| lhs[2].addEntry(msg.getBytes()); |
| } |
| if (changeNum) { |
| --num2; |
| ++num3; |
| } |
| } |
| |
| return lhs; |
| } |
| |
| private void verifyLedger(long lid, long startEntryId, long endEntryId) throws Exception { |
| LedgerHandle lh = bkc.openLedger(lid, digestType, passwdBytes); |
| Enumeration<LedgerEntry> entries = lh.readEntries(startEntryId, endEntryId); |
| while (entries.hasMoreElements()) { |
| LedgerEntry entry = entries.nextElement(); |
| assertEquals(msg, new String(entry.getEntry())); |
| } |
| } |
| |
| @Test |
| public void testDisableCompaction() throws Exception { |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, false); |
| |
| // disable compaction |
| baseConf.setMinorCompactionThreshold(0.0f); |
| baseConf.setMajorCompactionThreshold(0.0f); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| |
| // remove ledger2 and ledger3 |
| // so entry log 1 and 2 would have ledger1 entries left |
| bkc.deleteLedger(lhs[1].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| |
| assertFalse(getGCThread().enableMajorCompaction); |
| assertFalse(getGCThread().enableMinorCompaction); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, compaction should not be executed |
| assertEquals(lastMinorCompactionTime, getGCThread().lastMinorCompactionTime); |
| assertEquals(lastMajorCompactionTime, getGCThread().lastMajorCompactionTime); |
| |
| // entry logs ([0,1].log) should not be compacted. |
| for (File ledgerDirectory : tmpDirs) { |
| assertTrue("Not Found entry log file ([0,1].log that should have been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, false, 0, 1)); |
| } |
| } |
| |
| @Test |
| public void testForceGarbageCollectionWhenDisableCompactionConfigurationSettings() throws Exception { |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, false); |
| |
| baseConf.setForceAllowCompaction(true); |
| baseConf.setMajorCompactionThreshold(0.5f); |
| baseConf.setMinorCompactionThreshold(0.2f); |
| baseConf.setMajorCompactionInterval(0); |
| baseConf.setMinorCompactionInterval(0); |
| restartBookies(baseConf); |
| |
| assertFalse(getGCThread().enableMajorCompaction); |
| assertFalse(getGCThread().enableMinorCompaction); |
| assertTrue(getGCThread().isForceMajorCompactionAllow); |
| assertTrue(getGCThread().isForceMinorCompactionAllow); |
| |
| assertEquals(0.5f, getGCThread().majorCompactionThreshold, 0f); |
| assertEquals(0.2f, getGCThread().minorCompactionThreshold, 0f); |
| } |
| |
| @Test |
| public void testForceGarbageCollection() throws Exception { |
| testForceGarbageCollection(true); |
| testForceGarbageCollection(false); |
| } |
| |
| public void testForceGarbageCollection(boolean isForceCompactionAllowWhenDisableCompaction) throws Exception { |
| ServerConfiguration conf = newServerConfiguration(); |
| conf.setGcWaitTime(60000); |
| if (isForceCompactionAllowWhenDisableCompaction) { |
| conf.setMinorCompactionInterval(0); |
| conf.setMajorCompactionInterval(0); |
| conf.setForceAllowCompaction(true); |
| conf.setMajorCompactionThreshold(0.5f); |
| conf.setMinorCompactionThreshold(0.2f); |
| } else { |
| conf.setMinorCompactionInterval(120000); |
| conf.setMajorCompactionInterval(240000); |
| } |
| LedgerDirsManager dirManager = new LedgerDirsManager(conf, conf.getLedgerDirs(), |
| new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); |
| CheckpointSource cp = new CheckpointSource() { |
| @Override |
| public Checkpoint newCheckpoint() { |
| // Do nothing. |
| return null; |
| } |
| |
| @Override |
| public void checkpointComplete(Checkpoint checkPoint, boolean compact) |
| throws IOException { |
| // Do nothing. |
| } |
| }; |
| for (File journalDir : conf.getJournalDirs()) { |
| BookieImpl.checkDirectoryStructure(journalDir); |
| } |
| for (File dir : dirManager.getAllLedgerDirs()) { |
| BookieImpl.checkDirectoryStructure(dir); |
| } |
| runFunctionWithLedgerManagerFactory(conf, lmf -> { |
| try (LedgerManager lm = lmf.newLedgerManager()) { |
| InterleavedLedgerStorage storage = new InterleavedLedgerStorage(); |
| storage.initialize( |
| conf, |
| lm, |
| dirManager, |
| dirManager, |
| null, |
| cp, |
| Checkpointer.NULL, |
| NullStatsLogger.INSTANCE, |
| UnpooledByteBufAllocator.DEFAULT); |
| storage.start(); |
| long startTime = System.currentTimeMillis(); |
| storage.gcThread.enableForceGC(); |
| storage.gcThread.triggerGC().get(); //major |
| storage.gcThread.triggerGC().get(); //minor |
| // Minor and Major compaction times should be larger than when we started |
| // this test. |
| assertTrue("Minor or major compaction did not trigger even on forcing.", |
| storage.gcThread.lastMajorCompactionTime > startTime |
| && storage.gcThread.lastMinorCompactionTime > startTime); |
| storage.shutdown(); |
| } catch (Exception e) { |
| throw new UncheckedExecutionException(e.getMessage(), e); |
| } |
| return null; |
| }); |
| } |
| |
| @Test |
| public void testMinorCompaction() throws Exception { |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, false); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // disable major compaction |
| baseConf.setMajorCompactionThreshold(0.0f); |
| baseConf.setGcWaitTime(60000); |
| baseConf.setMinorCompactionInterval(120000); |
| baseConf.setMajorCompactionInterval(240000); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| assertTrue( |
| "ACTIVE_ENTRY_LOG_COUNT should have been updated", |
| getStatsProvider(0) |
| .getGauge("bookie.gc." + ACTIVE_ENTRY_LOG_COUNT) |
| .getSample().intValue() > 0); |
| assertTrue( |
| "ACTIVE_ENTRY_LOG_SPACE_BYTES should have been updated", |
| getStatsProvider(0) |
| .getGauge("bookie.gc." + ACTIVE_ENTRY_LOG_SPACE_BYTES) |
| .getSample().intValue() > 0); |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertFalse(getGCThread().enableMajorCompaction); |
| assertTrue(getGCThread().enableMinorCompaction); |
| |
| // remove ledger2 and ledger3 |
| bkc.deleteLedger(lhs[1].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, major compaction should not be executed |
| assertEquals(lastMajorCompactionTime, getGCThread().lastMajorCompactionTime); |
| assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); |
| |
| // entry logs ([0,1,2].log) should be compacted. |
| for (File ledgerDirectory : tmpDirs) { |
| assertFalse("Found entry log file ([0,1,2].log that should have not been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2)); |
| } |
| |
| // even though entry log files are removed, we still can access entries for ledger1 |
| // since those entries have been compacted to a new entry log |
| verifyLedger(lhs[0].getId(), 0, lhs[0].getLastAddConfirmed()); |
| |
| assertTrue( |
| "RECLAIMED_COMPACTION_SPACE_BYTES should have been updated", |
| getStatsProvider(0) |
| .getCounter("bookie.gc." + RECLAIMED_COMPACTION_SPACE_BYTES) |
| .get().intValue() > 0); |
| assertTrue( |
| "RECLAIMED_DELETION_SPACE_BYTES should have been updated", |
| getStatsProvider(0) |
| .getCounter("bookie.gc." + RECLAIMED_DELETION_SPACE_BYTES) |
| .get().intValue() > 0); |
| } |
| |
| @Test |
| public void testMinorCompactionWithMaxTimeMillis() throws Exception { |
| // prepare data |
| LedgerHandle[] lhs = prepareData(6, false); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // disable major compaction |
| baseConf.setMajorCompactionThreshold(0.0f); |
| baseConf.setGcWaitTime(60000); |
| baseConf.setMinorCompactionInterval(120000); |
| baseConf.setMajorCompactionInterval(240000); |
| |
| // Setup limit on compaction duration. |
| baseConf.setMinorCompactionMaxTimeMillis(15); |
| baseConf.setMajorCompactionMaxTimeMillis(15); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| assertTrue( |
| "ACTIVE_ENTRY_LOG_COUNT should have been updated", |
| getStatsProvider(0) |
| .getGauge("bookie.gc." + ACTIVE_ENTRY_LOG_COUNT) |
| .getSample().intValue() > 0); |
| assertTrue( |
| "ACTIVE_ENTRY_LOG_SPACE_BYTES should have been updated", |
| getStatsProvider(0) |
| .getGauge("bookie.gc." + ACTIVE_ENTRY_LOG_SPACE_BYTES) |
| .getSample().intValue() > 0); |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertFalse(getGCThread().enableMajorCompaction); |
| assertTrue(getGCThread().enableMinorCompaction); |
| |
| // remove ledger2 and ledger3 |
| bkc.deleteLedger(lhs[1].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, major compaction should not be executed |
| assertEquals(lastMajorCompactionTime, getGCThread().lastMajorCompactionTime); |
| assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); |
| |
| // entry logs ([0,1,2].log) should be compacted. |
| for (File ledgerDirectory : tmpDirs) { |
| assertFalse("Found entry log file ([0,1,2].log that should have not been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2)); |
| } |
| |
| // even though entry log files are removed, we still can access entries for ledger1 |
| // since those entries have been compacted to a new entry log |
| verifyLedger(lhs[0].getId(), 0, lhs[0].getLastAddConfirmed()); |
| |
| assertTrue( |
| "RECLAIMED_COMPACTION_SPACE_BYTES should have been updated", |
| getStatsProvider(0) |
| .getCounter("bookie.gc." + RECLAIMED_COMPACTION_SPACE_BYTES) |
| .get().intValue() > 0); |
| assertTrue( |
| "RECLAIMED_DELETION_SPACE_BYTES should have been updated", |
| getStatsProvider(0) |
| .getCounter("bookie.gc." + RECLAIMED_DELETION_SPACE_BYTES) |
| .get().intValue() > 0); |
| } |
| |
| |
| |
| |
| @Test |
| public void testForceMinorCompaction() throws Exception { |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, false); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // disable major compaction |
| baseConf.setMajorCompactionThreshold(0.0f); |
| baseConf.setGcWaitTime(60000); |
| baseConf.setMinorCompactionInterval(-1); |
| baseConf.setMajorCompactionInterval(-1); |
| baseConf.setForceAllowCompaction(true); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| assertTrue( |
| "ACTIVE_ENTRY_LOG_COUNT should have been updated", |
| getStatsProvider(0) |
| .getGauge("bookie.gc." + ACTIVE_ENTRY_LOG_COUNT) |
| .getSample().intValue() > 0); |
| assertTrue( |
| "ACTIVE_ENTRY_LOG_SPACE_BYTES should have been updated", |
| getStatsProvider(0) |
| .getGauge("bookie.gc." + ACTIVE_ENTRY_LOG_SPACE_BYTES) |
| .getSample().intValue() > 0); |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertFalse(getGCThread().enableMajorCompaction); |
| assertFalse(getGCThread().enableMinorCompaction); |
| |
| // remove ledger2 and ledger3 |
| bkc.deleteLedger(lhs[1].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, major compaction should not be executed |
| assertEquals(lastMajorCompactionTime, getGCThread().lastMajorCompactionTime); |
| assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); |
| |
| // entry logs ([0,1,2].log) should be compacted. |
| for (File ledgerDirectory : tmpDirs) { |
| assertFalse("Found entry log file ([0,1,2].log that should have not been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2)); |
| } |
| |
| // even though entry log files are removed, we still can access entries for ledger1 |
| // since those entries have been compacted to a new entry log |
| verifyLedger(lhs[0].getId(), 0, lhs[0].getLastAddConfirmed()); |
| |
| assertTrue( |
| "RECLAIMED_COMPACTION_SPACE_BYTES should have been updated", |
| getStatsProvider(0) |
| .getCounter("bookie.gc." + RECLAIMED_COMPACTION_SPACE_BYTES) |
| .get().intValue() > 0); |
| assertTrue( |
| "RECLAIMED_DELETION_SPACE_BYTES should have been updated", |
| getStatsProvider(0) |
| .getCounter("bookie.gc." + RECLAIMED_DELETION_SPACE_BYTES) |
| .get().intValue() > 0); |
| } |
| |
| @Test |
| public void testMinorCompactionWithNoWritableLedgerDirs() throws Exception { |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, false); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // disable major compaction |
| baseConf.setMajorCompactionThreshold(0.0f); |
| baseConf.setGcWaitTime(60000); |
| baseConf.setMinorCompactionInterval(120000); |
| baseConf.setMajorCompactionInterval(240000); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertFalse(getGCThread().enableMajorCompaction); |
| assertTrue(getGCThread().enableMinorCompaction); |
| |
| for (BookieServer bookieServer : bs) { |
| BookieImpl bookie = ((BookieImpl) bookieServer.getBookie()); |
| LedgerDirsManager ledgerDirsManager = bookie.getLedgerDirsManager(); |
| List<File> ledgerDirs = ledgerDirsManager.getAllLedgerDirs(); |
| // if all the discs are full then Major and Minor compaction would be disabled since |
| // 'isForceGCAllowWhenNoSpace' is not enabled. Check LedgerDirsListener of interleavedLedgerStorage. |
| for (File ledgerDir : ledgerDirs) { |
| ledgerDirsManager.addToFilledDirs(ledgerDir); |
| } |
| } |
| |
| // remove ledger2 and ledger3 |
| bkc.deleteLedger(lhs[1].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, major compaction should not be executed |
| assertEquals(lastMajorCompactionTime, getGCThread().lastMajorCompactionTime); |
| assertEquals(lastMinorCompactionTime, getGCThread().lastMinorCompactionTime); |
| |
| // entry logs ([0,1,2].log) should still remain, because both major and Minor compaction are disabled. |
| for (File ledgerDirectory : tmpDirs) { |
| assertTrue( |
| "All the entry log files ([0,1,2].log are not available, which is not expected" + ledgerDirectory, |
| TestUtils.hasLogFiles(ledgerDirectory, false, 0, 1, 2)); |
| } |
| } |
| |
| @Test |
| public void testMinorCompactionWithNoWritableLedgerDirsButIsForceGCAllowWhenNoSpaceIsSet() throws Exception { |
| stopAllBookies(); |
| ServerConfiguration conf = newServerConfiguration(); |
| // disable major compaction |
| conf.setMajorCompactionThreshold(0.0f); |
| // here we are setting isForceGCAllowWhenNoSpace to true, so Major and Minor compaction wont be disabled in case |
| // when discs are full |
| conf.setIsForceGCAllowWhenNoSpace(true); |
| conf.setGcWaitTime(600000); |
| conf.setMinorCompactionInterval(120000); |
| conf.setMajorCompactionInterval(240000); |
| // We need at least 2 ledger dirs because compaction will flush ledger cache, and will |
| // trigger relocateIndexFileAndFlushHeader. If we only have one ledger dir, compaction will always fail |
| // when there's no writeable ledger dir. |
| File ledgerDir1 = createTempDir("ledger", "test1"); |
| File ledgerDir2 = createTempDir("ledger", "test2"); |
| File journalDir = createTempDir("journal", "test"); |
| String[] ledgerDirNames = new String[]{ |
| ledgerDir1.getPath(), |
| ledgerDir2.getPath() |
| }; |
| conf.setLedgerDirNames(ledgerDirNames); |
| conf.setJournalDirName(journalDir.getPath()); |
| BookieServer server = startBookie(conf); |
| bs.add(server); |
| bsConfs.add(conf); |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, false); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertFalse(getGCThread().enableMajorCompaction); |
| assertTrue(getGCThread().enableMinorCompaction); |
| |
| for (BookieServer bookieServer : bs) { |
| BookieImpl bookie = ((BookieImpl) bookieServer.getBookie()); |
| bookie.getLedgerStorage().flush(); |
| bookie.dirsMonitor.shutdown(); |
| LedgerDirsManager ledgerDirsManager = bookie.getLedgerDirsManager(); |
| List<File> ledgerDirs = ledgerDirsManager.getAllLedgerDirs(); |
| // Major and Minor compaction are not disabled even though discs are full. Check LedgerDirsListener of |
| // interleavedLedgerStorage. |
| for (File ledgerDir : ledgerDirs) { |
| ledgerDirsManager.addToFilledDirs(ledgerDir); |
| } |
| } |
| |
| // remove ledger2 and ledger3 |
| bkc.deleteLedger(lhs[1].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| getGCThread().triggerGC(true, false, false).get(); |
| |
| // after garbage collection, major compaction should not be executed |
| assertEquals(lastMajorCompactionTime, getGCThread().lastMajorCompactionTime); |
| assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); |
| |
| // though all discs are added to filled dirs list, compaction would succeed, because in EntryLogger for |
| // allocating newlog |
| // we get getWritableLedgerDirsForNewLog() of ledgerDirsManager instead of getWritableLedgerDirs() |
| // entry logs ([0,1,2].log) should be compacted. |
| for (File ledgerDirectory : ((BookieImpl) server.getBookie()).getLedgerDirsManager().getAllLedgerDirs()) { |
| assertFalse("Found entry log file ([0,1,2].log that should have not been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory.getParentFile(), true, 0, 1, 2)); |
| } |
| |
| // even entry log files are removed, we still can access entries for ledger1 |
| // since those entries has been compacted to new entry log |
| verifyLedger(lhs[0].getId(), 0, lhs[0].getLastAddConfirmed()); |
| |
| // for the sake of validity of test lets make sure that there is no writableLedgerDir in the bookies |
| for (BookieServer bookieServer : bs) { |
| BookieImpl bookie = (BookieImpl) bookieServer.getBookie(); |
| LedgerDirsManager ledgerDirsManager = bookie.getLedgerDirsManager(); |
| try { |
| List<File> ledgerDirs = ledgerDirsManager.getWritableLedgerDirs(); |
| // it is expected not to have any writableLedgerDirs since we added all of them to FilledDirs |
| fail("It is expected not to have any writableLedgerDirs"); |
| } catch (NoWritableLedgerDirException nwe) { |
| |
| } |
| } |
| } |
| |
| @Test |
| public void testMajorCompaction() throws Exception { |
| |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, true); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // disable minor compaction |
| baseConf.setMinorCompactionThreshold(0.0f); |
| baseConf.setGcWaitTime(60000); |
| baseConf.setMinorCompactionInterval(120000); |
| baseConf.setMajorCompactionInterval(240000); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertTrue(getGCThread().enableMajorCompaction); |
| assertFalse(getGCThread().enableMinorCompaction); |
| |
| // remove ledger1 and ledger3 |
| bkc.deleteLedger(lhs[0].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, minor compaction should not be executed |
| assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); |
| assertTrue(getGCThread().lastMajorCompactionTime > lastMajorCompactionTime); |
| |
| // entry logs ([0,1,2].log) should be compacted |
| for (File ledgerDirectory : tmpDirs) { |
| assertFalse("Found entry log file ([0,1,2].log that should have not been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2)); |
| } |
| |
| // even entry log files are removed, we still can access entries for ledger2 |
| // since those entries has been compacted to new entry log |
| verifyLedger(lhs[1].getId(), 0, lhs[1].getLastAddConfirmed()); |
| } |
| |
| @Test |
| public void testForceMajorCompaction() throws Exception { |
| |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, true); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // disable minor compaction |
| baseConf.setMinorCompactionThreshold(0.0f); |
| baseConf.setGcWaitTime(60000); |
| baseConf.setMinorCompactionInterval(-1); |
| baseConf.setMajorCompactionInterval(-1); |
| baseConf.setForceAllowCompaction(true); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertFalse(getGCThread().enableMajorCompaction); |
| assertFalse(getGCThread().enableMinorCompaction); |
| assertTrue(getGCThread().isForceMajorCompactionAllow); |
| assertFalse(getGCThread().isForceMinorCompactionAllow); |
| |
| // remove ledger1 and ledger3 |
| bkc.deleteLedger(lhs[0].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, minor compaction should not be executed |
| assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); |
| assertTrue(getGCThread().lastMajorCompactionTime > lastMajorCompactionTime); |
| |
| // entry logs ([0,1,2].log) should be compacted |
| for (File ledgerDirectory : tmpDirs) { |
| assertFalse("Found entry log file ([0,1,2].log that should have not been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2)); |
| } |
| |
| // even entry log files are removed, we still can access entries for ledger2 |
| // since those entries has been compacted to new entry log |
| verifyLedger(lhs[1].getId(), 0, lhs[1].getLastAddConfirmed()); |
| } |
| |
| @Test |
| public void testCompactionPersistence() throws Exception { |
| /* |
| * for this test scenario we are assuming that there will be only one |
| * bookie in the cluster |
| */ |
| assertEquals("Numbers of Bookies in this cluster", 1, numBookies); |
| /* |
| * this test is for validating EntryLogCompactor, so make sure |
| * TransactionalCompaction is not enabled. |
| */ |
| assertFalse("Bookies must be using EntryLogCompactor", baseConf.getUseTransactionalCompaction()); |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, true); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // disable minor compaction |
| baseConf.setMinorCompactionThreshold(0.0f); |
| baseConf.setGcWaitTime(60000); |
| baseConf.setMinorCompactionInterval(120000); |
| baseConf.setMajorCompactionInterval(240000); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertTrue(getGCThread().enableMajorCompaction); |
| assertFalse(getGCThread().enableMinorCompaction); |
| |
| // remove ledger1 and ledger3 |
| bkc.deleteLedger(lhs[0].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, minor compaction should not be executed |
| assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); |
| assertTrue(getGCThread().lastMajorCompactionTime > lastMajorCompactionTime); |
| |
| // entry logs ([0,1,2].log) should be compacted |
| for (File ledgerDirectory : tmpDirs) { |
| assertFalse("Found entry log file ([0,1,2].log that should have not been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2)); |
| } |
| |
| // even entry log files are removed, we still can access entries for |
| // ledger2 |
| // since those entries has been compacted to new entry log |
| long ledgerId = lhs[1].getId(); |
| long lastAddConfirmed = lhs[1].getLastAddConfirmed(); |
| verifyLedger(ledgerId, 0, lastAddConfirmed); |
| |
| /* |
| * there is only one bookie in the cluster so we should be able to read |
| * entries from this bookie. |
| */ |
| ServerConfiguration bookieServerConfig = ((BookieImpl) bs.get(0).getBookie()).conf; |
| ServerConfiguration newBookieConf = new ServerConfiguration(bookieServerConfig); |
| /* |
| * by reusing bookieServerConfig and setting metadataServiceUri to null |
| * we can create/start new Bookie instance using the same data |
| * (journal/ledger/index) of the existing BookeieServer for our testing |
| * purpose. |
| */ |
| newBookieConf.setMetadataServiceUri(null); |
| Bookie newbookie = new BookieImpl(newBookieConf); |
| |
| DigestManager digestManager = DigestManager.instantiate(ledgerId, passwdBytes, |
| BookKeeper.DigestType.toProtoDigestType(digestType), UnpooledByteBufAllocator.DEFAULT, |
| baseClientConf.getUseV2WireProtocol()); |
| |
| for (long entryId = 0; entryId <= lastAddConfirmed; entryId++) { |
| ByteBuf readEntryBufWithChecksum = newbookie.readEntry(ledgerId, entryId); |
| ByteBuf readEntryBuf = digestManager.verifyDigestAndReturnData(entryId, readEntryBufWithChecksum); |
| byte[] readEntryBytes = new byte[readEntryBuf.readableBytes()]; |
| readEntryBuf.readBytes(readEntryBytes); |
| assertEquals(msg, new String(readEntryBytes)); |
| } |
| } |
| |
| @Test |
| public void testCompactionWhenLedgerDirsAreFull() throws Exception { |
| /* |
| * for this test scenario we are assuming that there will be only one |
| * bookie in the cluster |
| */ |
| assertEquals("Numbers of Bookies in this cluster", 1, bsConfs.size()); |
| ServerConfiguration serverConfig = bsConfs.get(0); |
| File ledgerDir = serverConfig.getLedgerDirs()[0]; |
| assertEquals("Number of Ledgerdirs for this bookie", 1, serverConfig.getLedgerDirs().length); |
| assertTrue("indexdirs should be configured to null", null == serverConfig.getIndexDirs()); |
| /* |
| * this test is for validating EntryLogCompactor, so make sure |
| * TransactionalCompaction is not enabled. |
| */ |
| assertFalse("Bookies must be using EntryLogCompactor", baseConf.getUseTransactionalCompaction()); |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, true); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| bs.get(0).getBookie().getLedgerStorage().flush(); |
| assertTrue( |
| "entry log file ([0,1,2].log should be available in ledgerDirectory: " |
| + serverConfig.getLedgerDirs()[0], |
| TestUtils.hasLogFiles(serverConfig.getLedgerDirs()[0], false, 0, 1, 2)); |
| |
| long usableSpace = ledgerDir.getUsableSpace(); |
| long totalSpace = ledgerDir.getTotalSpace(); |
| |
| baseConf.setForceReadOnlyBookie(true); |
| baseConf.setIsForceGCAllowWhenNoSpace(true); |
| // disable minor compaction |
| baseConf.setMinorCompactionThreshold(0.0f); |
| baseConf.setGcWaitTime(60000); |
| baseConf.setMinorCompactionInterval(120000); |
| baseConf.setMajorCompactionInterval(240000); |
| baseConf.setMinUsableSizeForEntryLogCreation(1); |
| baseConf.setMinUsableSizeForIndexFileCreation(1); |
| baseConf.setDiskUsageThreshold((1.0f - ((float) usableSpace / (float) totalSpace)) * 0.9f); |
| baseConf.setDiskUsageWarnThreshold(0.0f); |
| |
| /* |
| * because of the value set for diskUsageThreshold, when bookie is |
| * restarted it wouldn't find any writableledgerdir. But we have set |
| * very low values for minUsableSizeForEntryLogCreation and |
| * minUsableSizeForIndexFileCreation, so it should be able to create |
| * EntryLog file and Index file for doing compaction. |
| */ |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| assertFalse("There shouldn't be any writable ledgerDir", |
| ((BookieImpl) bs.get(0).getBookie()).getLedgerDirsManager().hasWritableLedgerDirs()); |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertTrue(getGCThread().enableMajorCompaction); |
| assertFalse(getGCThread().enableMinorCompaction); |
| |
| // remove ledger1 and ledger3 |
| bkc.deleteLedger(lhs[0].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, minor compaction should not be executed |
| assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); |
| assertTrue(getGCThread().lastMajorCompactionTime > lastMajorCompactionTime); |
| |
| /* |
| * GarbageCollection should have succeeded, so no previous entrylog |
| * should be available. |
| */ |
| |
| // entry logs ([0,1,2].log) should be compacted |
| for (File ledgerDirectory : tmpDirs) { |
| assertFalse("Found entry log file ([0,1,2].log that should have not been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2)); |
| } |
| |
| // even entry log files are removed, we still can access entries for |
| // ledger2 |
| // since those entries has been compacted to new entry log |
| long ledgerId = lhs[1].getId(); |
| long lastAddConfirmed = lhs[1].getLastAddConfirmed(); |
| verifyLedger(ledgerId, 0, lastAddConfirmed); |
| } |
| |
| @Test |
| public void testMajorCompactionAboveThreshold() throws Exception { |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, false); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| long lastMinorCompactionTime = getGCThread().lastMinorCompactionTime; |
| long lastMajorCompactionTime = getGCThread().lastMajorCompactionTime; |
| assertTrue(getGCThread().enableMajorCompaction); |
| assertTrue(getGCThread().enableMinorCompaction); |
| |
| // remove ledger1 and ledger2 |
| bkc.deleteLedger(lhs[0].getId()); |
| bkc.deleteLedger(lhs[1].getId()); |
| LOG.info("Finished deleting the ledgers contains less entries."); |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| |
| // after garbage collection, minor compaction should not be executed |
| assertTrue(getGCThread().lastMinorCompactionTime > lastMinorCompactionTime); |
| assertTrue(getGCThread().lastMajorCompactionTime > lastMajorCompactionTime); |
| |
| // entry logs ([0,1,2].log) should not be compacted |
| for (File ledgerDirectory : tmpDirs) { |
| assertTrue("Not Found entry log file ([1,2].log that should have been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, false, 0, 1, 2)); |
| } |
| } |
| |
| @Test |
| public void testCompactionSmallEntryLogs() throws Exception { |
| |
| // create a ledger to write a few entries |
| LedgerHandle alh = bkc.createLedger(NUM_BOOKIES, NUM_BOOKIES, digestType, "".getBytes()); |
| for (int i = 0; i < 3; i++) { |
| alh.addEntry(msg.getBytes()); |
| } |
| alh.close(); |
| |
| // restart bookie to roll entry log files |
| restartBookies(); |
| |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, false); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // remove ledger2 and ledger3 |
| bkc.deleteLedger(lhs[1].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| // restart bookies again to roll entry log files. |
| restartBookies(); |
| |
| getGCThread().enableForceGC(); |
| getGCThread().triggerGC().get(); |
| |
| // entry logs (0.log) should not be compacted |
| // entry logs ([1,2,3].log) should be compacted. |
| for (File ledgerDirectory : tmpDirs) { |
| assertTrue("Not Found entry log file ([0].log that should have been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0)); |
| assertFalse("Found entry log file ([1,2,3].log that should have not been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 1, 2, 3)); |
| } |
| |
| // even entry log files are removed, we still can access entries for ledger1 |
| // since those entries has been compacted to new entry log |
| verifyLedger(lhs[0].getId(), 0, lhs[0].getLastAddConfirmed()); |
| } |
| |
| /** |
| * Test that compaction doesnt add to index without having persisted |
| * entrylog first. This is needed because compaction doesn't go through the journal. |
| * {@see https://issues.apache.org/jira/browse/BOOKKEEPER-530} |
| * {@see https://issues.apache.org/jira/browse/BOOKKEEPER-664} |
| */ |
| @Test |
| public void testCompactionSafety() throws Exception { |
| tearDown(); // I dont want the test infrastructure |
| ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); |
| final Set<Long> ledgers = Collections.newSetFromMap(new ConcurrentHashMap<Long, Boolean>()); |
| LedgerManager manager = getLedgerManager(ledgers); |
| |
| File tmpDir = createTempDir("bkTest", ".dir"); |
| File curDir = BookieImpl.getCurrentDirectory(tmpDir); |
| BookieImpl.checkDirectoryStructure(curDir); |
| conf.setLedgerDirNames(new String[] {tmpDir.toString()}); |
| |
| conf.setEntryLogSizeLimit(EntryLogger.LOGFILE_HEADER_SIZE + 3 * (4 + ENTRY_SIZE)); |
| conf.setGcWaitTime(100); |
| conf.setMinorCompactionThreshold(0.7f); |
| conf.setMajorCompactionThreshold(0.0f); |
| conf.setMinorCompactionInterval(1); |
| conf.setMajorCompactionInterval(10); |
| conf.setPageLimit(1); |
| |
| CheckpointSource checkpointSource = new CheckpointSource() { |
| AtomicInteger idGen = new AtomicInteger(0); |
| class MyCheckpoint implements CheckpointSource.Checkpoint { |
| int id = idGen.incrementAndGet(); |
| @Override |
| public int compareTo(CheckpointSource.Checkpoint o) { |
| if (o == CheckpointSource.Checkpoint.MAX) { |
| return -1; |
| } else if (o == CheckpointSource.Checkpoint.MIN) { |
| return 1; |
| } |
| return id - ((MyCheckpoint) o).id; |
| } |
| } |
| |
| @Override |
| public CheckpointSource.Checkpoint newCheckpoint() { |
| return new MyCheckpoint(); |
| } |
| |
| public void checkpointComplete(CheckpointSource.Checkpoint checkpoint, boolean compact) |
| throws IOException { |
| } |
| }; |
| final byte[] key = "foobar".getBytes(); |
| File log0 = new File(curDir, "0.log"); |
| LedgerDirsManager dirs = new LedgerDirsManager(conf, conf.getLedgerDirs(), |
| new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); |
| assertFalse("Log shouldnt exist", log0.exists()); |
| InterleavedLedgerStorage storage = new InterleavedLedgerStorage(); |
| storage.initialize( |
| conf, |
| manager, |
| dirs, |
| dirs, |
| null, |
| checkpointSource, |
| Checkpointer.NULL, |
| NullStatsLogger.INSTANCE, |
| UnpooledByteBufAllocator.DEFAULT); |
| ledgers.add(1L); |
| ledgers.add(2L); |
| ledgers.add(3L); |
| storage.setMasterKey(1, key); |
| storage.setMasterKey(2, key); |
| storage.setMasterKey(3, key); |
| storage.addEntry(genEntry(1, 1, ENTRY_SIZE)); |
| storage.addEntry(genEntry(2, 1, ENTRY_SIZE)); |
| storage.addEntry(genEntry(2, 2, ENTRY_SIZE)); |
| storage.addEntry(genEntry(3, 2, ENTRY_SIZE)); |
| storage.flush(); |
| storage.shutdown(); |
| |
| assertTrue("Log should exist", log0.exists()); |
| ledgers.remove(2L); |
| ledgers.remove(3L); |
| |
| storage = new InterleavedLedgerStorage(); |
| storage.initialize( |
| conf, |
| manager, |
| dirs, dirs, null, |
| checkpointSource, |
| Checkpointer.NULL, |
| NullStatsLogger.INSTANCE, |
| UnpooledByteBufAllocator.DEFAULT); |
| storage.start(); |
| for (int i = 0; i < 10; i++) { |
| if (!log0.exists()) { |
| break; |
| } |
| Thread.sleep(1000); |
| storage.entryLogger.flush(); // simulate sync thread |
| } |
| assertFalse("Log shouldnt exist", log0.exists()); |
| |
| ledgers.add(4L); |
| storage.setMasterKey(4, key); |
| storage.addEntry(genEntry(4, 1, ENTRY_SIZE)); // force ledger 1 page to flush |
| storage.shutdown(); |
| |
| storage = new InterleavedLedgerStorage(); |
| storage.initialize( |
| conf, |
| manager, |
| dirs, |
| dirs, |
| null, |
| checkpointSource, |
| Checkpointer.NULL, |
| NullStatsLogger.INSTANCE, |
| UnpooledByteBufAllocator.DEFAULT); |
| storage.getEntry(1, 1); // entry should exist |
| } |
| |
| private LedgerManager getLedgerManager(final Set<Long> ledgers) { |
| LedgerManager manager = new LedgerManager() { |
| @Override |
| public CompletableFuture<Versioned<LedgerMetadata>> createLedgerMetadata(long lid, |
| LedgerMetadata metadata) { |
| unsupported(); |
| return null; |
| } |
| @Override |
| public CompletableFuture<Void> removeLedgerMetadata(long ledgerId, Version version) { |
| unsupported(); |
| return null; |
| } |
| @Override |
| public CompletableFuture<Versioned<LedgerMetadata>> readLedgerMetadata(long ledgerId) { |
| unsupported(); |
| return null; |
| } |
| @Override |
| public CompletableFuture<Versioned<LedgerMetadata>> writeLedgerMetadata(long ledgerId, |
| LedgerMetadata metadata, |
| Version currentVersion) { |
| unsupported(); |
| return null; |
| } |
| @Override |
| public void asyncProcessLedgers(Processor<Long> processor, |
| AsyncCallback.VoidCallback finalCb, |
| Object context, int successRc, int failureRc) { |
| unsupported(); |
| } |
| @Override |
| public void registerLedgerMetadataListener(long ledgerId, |
| LedgerMetadataListener listener) { |
| unsupported(); |
| } |
| @Override |
| public void unregisterLedgerMetadataListener(long ledgerId, |
| LedgerMetadataListener listener) { |
| unsupported(); |
| } |
| @Override |
| public void close() throws IOException {} |
| |
| void unsupported() { |
| LOG.error("Unsupported operation called", new Exception()); |
| throw new RuntimeException("Unsupported op"); |
| } |
| |
| @Override |
| public LedgerRangeIterator getLedgerRanges(long zkOpTimeoutMs) { |
| final AtomicBoolean hasnext = new AtomicBoolean(true); |
| return new LedgerManager.LedgerRangeIterator() { |
| @Override |
| public boolean hasNext() throws IOException { |
| return hasnext.get(); |
| } |
| @Override |
| public LedgerManager.LedgerRange next() throws IOException { |
| hasnext.set(false); |
| return new LedgerManager.LedgerRange(ledgers); |
| } |
| }; |
| } |
| }; |
| return manager; |
| } |
| |
| /** |
| * Test that compaction should execute silently when there is no entry logs |
| * to compact. {@see https://issues.apache.org/jira/browse/BOOKKEEPER-700} |
| */ |
| @Test |
| public void testWhenNoLogsToCompact() throws Exception { |
| tearDown(); // I dont want the test infrastructure |
| ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); |
| File tmpDir = createTempDir("bkTest", ".dir"); |
| File curDir = BookieImpl.getCurrentDirectory(tmpDir); |
| BookieImpl.checkDirectoryStructure(curDir); |
| conf.setLedgerDirNames(new String[] { tmpDir.toString() }); |
| |
| LedgerDirsManager dirs = new LedgerDirsManager(conf, conf.getLedgerDirs(), |
| new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); |
| 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, |
| null, |
| checkpointSource, |
| Checkpointer.NULL, |
| NullStatsLogger.INSTANCE, |
| UnpooledByteBufAllocator.DEFAULT); |
| |
| double threshold = 0.1; |
| long limit = 0; |
| |
| // shouldn't throw exception |
| storage.gcThread.doCompactEntryLogs(threshold, limit); |
| } |
| |
| /** |
| * Test extractMetaFromEntryLogs optimized method to avoid excess memory usage. |
| */ |
| public void testExtractMetaFromEntryLogs() throws Exception { |
| // Always run this test with Throttle enabled. |
| baseConf.setIsThrottleByBytes(true); |
| // restart bookies |
| restartBookies(baseConf); |
| ServerConfiguration conf = TestBKConfiguration.newServerConfiguration(); |
| File tmpDir = createTempDir("bkTest", ".dir"); |
| File curDir = BookieImpl.getCurrentDirectory(tmpDir); |
| BookieImpl.checkDirectoryStructure(curDir); |
| conf.setLedgerDirNames(new String[] { tmpDir.toString() }); |
| |
| LedgerDirsManager dirs = new LedgerDirsManager(conf, conf.getLedgerDirs(), |
| new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); |
| 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, null, checkpointSource, |
| Checkpointer.NULL, NullStatsLogger.INSTANCE, UnpooledByteBufAllocator.DEFAULT); |
| |
| 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, null, checkpointSource, |
| Checkpointer.NULL, NullStatsLogger.INSTANCE, UnpooledByteBufAllocator.DEFAULT); |
| |
| long startingEntriesCount = storage.gcThread.entryLogger.getLeastUnflushedLogId() |
| - storage.gcThread.scannedLogId; |
| LOG.info("The old Log Entry count is: " + startingEntriesCount); |
| |
| Map<Long, EntryLogMetadata> entryLogMetaData = new HashMap<>(); |
| 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 ByteBuf genEntry(long ledger, long entry, int size) { |
| ByteBuf bb = Unpooled.buffer(size); |
| bb.writeLong(ledger); |
| bb.writeLong(entry); |
| while (bb.isWritable()) { |
| bb.writeByte((byte) 0xFF); |
| } |
| return bb; |
| } |
| |
| /** |
| * Suspend garbage collection when suspendMajor/suspendMinor is set. |
| */ |
| @Test |
| public void testSuspendGarbageCollection() throws Exception { |
| ServerConfiguration conf = newServerConfiguration(); |
| conf.setGcWaitTime(500); |
| conf.setMinorCompactionInterval(1); |
| conf.setMajorCompactionInterval(2); |
| runFunctionWithLedgerManagerFactory(conf, lmf -> { |
| try (LedgerManager lm = lmf.newLedgerManager()) { |
| testSuspendGarbageCollection(conf, lm); |
| } catch (Exception e) { |
| throw new UncheckedExecutionException(e.getMessage(), e); |
| } |
| return null; |
| }); |
| } |
| |
| private void testSuspendGarbageCollection(ServerConfiguration conf, |
| LedgerManager lm) throws Exception { |
| LedgerDirsManager dirManager = new LedgerDirsManager(conf, conf.getLedgerDirs(), |
| new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())); |
| CheckpointSource cp = new CheckpointSource() { |
| @Override |
| public Checkpoint newCheckpoint() { |
| // Do nothing. |
| return null; |
| } |
| |
| @Override |
| public void checkpointComplete(Checkpoint checkPoint, boolean compact) |
| throws IOException { |
| // Do nothing. |
| } |
| }; |
| for (File journalDir : conf.getJournalDirs()) { |
| BookieImpl.checkDirectoryStructure(journalDir); |
| } |
| for (File dir : dirManager.getAllLedgerDirs()) { |
| BookieImpl.checkDirectoryStructure(dir); |
| } |
| InterleavedLedgerStorage storage = new InterleavedLedgerStorage(); |
| TestStatsProvider stats = new TestStatsProvider(); |
| storage.initialize( |
| conf, |
| lm, |
| dirManager, |
| dirManager, |
| null, |
| cp, |
| Checkpointer.NULL, |
| stats.getStatsLogger("storage"), |
| UnpooledByteBufAllocator.DEFAULT); |
| storage.start(); |
| |
| int majorCompactions = stats.getCounter("storage.gc." + MAJOR_COMPACTION_COUNT).get().intValue(); |
| int minorCompactions = stats.getCounter("storage.gc." + MINOR_COMPACTION_COUNT).get().intValue(); |
| Thread.sleep(conf.getMajorCompactionInterval() * 1000 |
| + conf.getGcWaitTime()); |
| assertTrue( |
| "Major compaction should have happened", |
| stats.getCounter("storage.gc." + MAJOR_COMPACTION_COUNT).get() > majorCompactions); |
| |
| // test suspend Major GC. |
| storage.gcThread.suspendMajorGC(); |
| |
| Thread.sleep(1000); |
| long startTime = System.currentTimeMillis(); |
| majorCompactions = stats.getCounter("storage.gc." + MAJOR_COMPACTION_COUNT).get().intValue(); |
| Thread.sleep(conf.getMajorCompactionInterval() * 1000 |
| + conf.getGcWaitTime()); |
| assertTrue("major compaction triggered while suspended", |
| storage.gcThread.lastMajorCompactionTime < startTime); |
| assertTrue("major compaction triggered while suspended", |
| stats.getCounter("storage.gc." + MAJOR_COMPACTION_COUNT).get() == majorCompactions); |
| |
| // test suspend Major GC. |
| Thread.sleep(conf.getMinorCompactionInterval() * 1000 |
| + conf.getGcWaitTime()); |
| assertTrue( |
| "Minor compaction should have happened", |
| stats.getCounter("storage.gc." + MINOR_COMPACTION_COUNT).get() > minorCompactions); |
| |
| // test suspend Minor GC. |
| storage.gcThread.suspendMinorGC(); |
| |
| Thread.sleep(1000); |
| startTime = System.currentTimeMillis(); |
| minorCompactions = stats.getCounter("storage.gc." + MINOR_COMPACTION_COUNT).get().intValue(); |
| Thread.sleep(conf.getMajorCompactionInterval() * 1000 |
| + conf.getGcWaitTime()); |
| assertTrue("minor compaction triggered while suspended", |
| storage.gcThread.lastMinorCompactionTime < startTime); |
| assertTrue("minor compaction triggered while suspended", |
| stats.getCounter("storage.gc." + MINOR_COMPACTION_COUNT).get() == minorCompactions); |
| |
| // test resume |
| storage.gcThread.resumeMinorGC(); |
| storage.gcThread.resumeMajorGC(); |
| |
| Thread.sleep((conf.getMajorCompactionInterval() + conf.getMinorCompactionInterval()) * 1000 |
| + (conf.getGcWaitTime() * 2)); |
| assertTrue( |
| "Major compaction should have happened", |
| stats.getCounter("storage.gc." + MAJOR_COMPACTION_COUNT).get() > majorCompactions); |
| assertTrue( |
| "Minor compaction should have happened", |
| stats.getCounter("storage.gc." + MINOR_COMPACTION_COUNT).get() > minorCompactions); |
| assertTrue( |
| "gcThreadRunttime should be non-zero", |
| stats.getOpStatsLogger("storage.gc." + THREAD_RUNTIME).getSuccessCount() > 0); |
| |
| } |
| |
| @Test |
| public void testRecoverIndexWhenIndexIsPartiallyFlush() throws Exception { |
| // prepare data |
| LedgerHandle[] lhs = prepareData(3, false); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // disable compaction |
| baseConf.setMinorCompactionThreshold(0.0f); |
| baseConf.setMajorCompactionThreshold(0.0f); |
| baseConf.setGcWaitTime(600000); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| BookieImpl bookie = ((BookieImpl) bs.get(0).getBookie()); |
| InterleavedLedgerStorage storage = (InterleavedLedgerStorage) bookie.ledgerStorage; |
| |
| // remove ledger2 and ledger3 |
| bkc.deleteLedger(lhs[1].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| |
| MockTransactionalEntryLogCompactor partialCompactionWorker = new MockTransactionalEntryLogCompactor( |
| ((InterleavedLedgerStorage) bookie.getLedgerStorage()).gcThread); |
| |
| for (long logId = 0; logId < 3; logId++) { |
| EntryLogMetadata meta = storage.entryLogger.getEntryLogMetadata(logId); |
| partialCompactionWorker.compactWithIndexFlushFailure(meta); |
| } |
| |
| // entry logs ([0,1,2].log) should not be compacted because of partial flush throw IOException |
| for (File ledgerDirectory : tmpDirs) { |
| assertTrue("Entry log file ([0,1,2].log should not be compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2)); |
| } |
| |
| // entries should be available |
| verifyLedger(lhs[0].getId(), 0, lhs[0].getLastAddConfirmed()); |
| |
| // But we should see .compacted file with index flush failure |
| assertEquals(findCompactedEntryLogFiles().size(), 3); |
| |
| // Now try to recover those flush failed index files |
| partialCompactionWorker.cleanUpAndRecover(); |
| |
| // There should be no .compacted files after recovery |
| assertEquals(findCompactedEntryLogFiles().size(), 0); |
| |
| // compaction worker should recover partial flushed index and delete [0,1,2].log |
| for (File ledgerDirectory : tmpDirs) { |
| assertFalse("Entry log file ([0,1,2].log should have been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2)); |
| } |
| |
| // even entry log files are removed, we still can access entries for ledger1 |
| // since those entries has been compacted to new entry log |
| verifyLedger(lhs[0].getId(), 0, lhs[0].getLastAddConfirmed()); |
| } |
| |
| @Test |
| public void testCompactionFailureShouldNotResultInDuplicatedData() throws Exception { |
| // prepare data |
| LedgerHandle[] lhs = prepareData(5, false); |
| |
| for (LedgerHandle lh : lhs) { |
| lh.close(); |
| } |
| |
| // disable compaction |
| baseConf.setMinorCompactionThreshold(0.0f); |
| baseConf.setMajorCompactionThreshold(0.0f); |
| baseConf.setUseTransactionalCompaction(true); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| // remove ledger2 and ledger3 |
| bkc.deleteLedger(lhs[1].getId()); |
| bkc.deleteLedger(lhs[2].getId()); |
| |
| LOG.info("Finished deleting the ledgers contains most entries."); |
| Thread.sleep(baseConf.getMajorCompactionInterval() * 1000 |
| + baseConf.getGcWaitTime()); |
| BookieImpl bookie = (BookieImpl) bs.get(0).getBookie(); |
| InterleavedLedgerStorage storage = (InterleavedLedgerStorage) bookie.ledgerStorage; |
| |
| List<File> ledgerDirs = bookie.getLedgerDirsManager().getAllLedgerDirs(); |
| List<Long> usageBeforeCompaction = new ArrayList<>(); |
| ledgerDirs.forEach(file -> usageBeforeCompaction.add(getDirectorySpaceUsage(file))); |
| |
| MockTransactionalEntryLogCompactor partialCompactionWorker = new MockTransactionalEntryLogCompactor( |
| ((InterleavedLedgerStorage) bookie.ledgerStorage).gcThread); |
| |
| for (long logId = 0; logId < 5; logId++) { |
| EntryLogMetadata meta = storage.entryLogger.getEntryLogMetadata(logId); |
| partialCompactionWorker.compactWithLogFlushFailure(meta); |
| } |
| |
| // entry logs ([0-4].log) should not be compacted because of failure in flush compaction log |
| for (File ledgerDirectory : tmpDirs) { |
| assertTrue("Entry log file ([0,1,2].log should not be compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2, 3, 4)); |
| } |
| // even entry log files are removed, we still can access entries for ledger1 |
| // since those entries has been compacted to new entry log |
| verifyLedger(lhs[0].getId(), 0, lhs[0].getLastAddConfirmed()); |
| |
| List<Long> freeSpaceAfterCompactionFailed = new ArrayList<>(); |
| ledgerDirs.forEach(file -> freeSpaceAfterCompactionFailed.add(getDirectorySpaceUsage(file))); |
| |
| // No extra data is generated after compaction fail |
| for (int i = 0; i < usageBeforeCompaction.size(); i++) { |
| assertEquals(usageBeforeCompaction.get(i), freeSpaceAfterCompactionFailed.get(i)); |
| } |
| |
| // now enable normal compaction |
| baseConf.setMajorCompactionThreshold(0.5f); |
| |
| // restart bookies |
| restartBookies(baseConf); |
| |
| Thread.sleep(baseConf.getMajorCompactionInterval() * 1000 |
| + baseConf.getGcWaitTime()); |
| // compaction worker should compact [0-4].log |
| for (File ledgerDirectory : tmpDirs) { |
| assertFalse("Entry log file ([0,1,2].log should have been compacted in ledgerDirectory: " |
| + ledgerDirectory, TestUtils.hasLogFiles(ledgerDirectory, true, 0, 1, 2, 3, 4)); |
| } |
| |
| // even entry log files are removed, we still can access entries for ledger1 |
| // since those entries has been compacted to new entry log |
| verifyLedger(lhs[0].getId(), 0, lhs[0].getLastAddConfirmed()); |
| } |
| |
| private long getDirectorySpaceUsage(File dir) { |
| long size = 0; |
| for (File file : dir.listFiles()) { |
| size += file.length(); |
| } |
| return size; |
| } |
| |
| private Set<File> findCompactedEntryLogFiles() { |
| Set<File> compactedLogFiles = new HashSet<>(); |
| for (File ledgerDirectory : tmpDirs) { |
| File[] files = BookieImpl.getCurrentDirectory(ledgerDirectory).listFiles( |
| file -> file.getName().endsWith(COMPACTED_SUFFIX)); |
| if (files != null) { |
| Collections.addAll(compactedLogFiles, files); |
| } |
| } |
| return compactedLogFiles; |
| } |
| |
| private static class MockTransactionalEntryLogCompactor extends TransactionalEntryLogCompactor { |
| |
| public MockTransactionalEntryLogCompactor(GarbageCollectorThread gcThread) { |
| super(gcThread.conf, |
| gcThread.entryLogger, |
| gcThread.ledgerStorage, |
| (long entry) -> { |
| gcThread.removeEntryLog(entry); |
| }); |
| } |
| |
| synchronized void compactWithIndexFlushFailure(EntryLogMetadata metadata) { |
| LOG.info("Compacting entry log {}.", metadata.getEntryLogId()); |
| CompactionPhase scanEntryLog = new ScanEntryLogPhase(metadata); |
| if (!scanEntryLog.run()) { |
| LOG.info("Compaction for {} end in ScanEntryLogPhase.", metadata.getEntryLogId()); |
| return; |
| } |
| File compactionLogFile = entryLogger.getCurCompactionLogFile(); |
| CompactionPhase flushCompactionLog = new FlushCompactionLogPhase(metadata.getEntryLogId()); |
| if (!flushCompactionLog.run()) { |
| LOG.info("Compaction for {} end in FlushCompactionLogPhase.", metadata.getEntryLogId()); |
| return; |
| } |
| File compactedLogFile = getCompactedLogFile(compactionLogFile, metadata.getEntryLogId()); |
| CompactionPhase partialFlushIndexPhase = new PartialFlushIndexPhase(compactedLogFile); |
| if (!partialFlushIndexPhase.run()) { |
| LOG.info("Compaction for {} end in PartialFlushIndexPhase.", metadata.getEntryLogId()); |
| return; |
| } |
| logRemovalListener.removeEntryLog(metadata.getEntryLogId()); |
| LOG.info("Compacted entry log : {}.", metadata.getEntryLogId()); |
| } |
| |
| synchronized void compactWithLogFlushFailure(EntryLogMetadata metadata) { |
| LOG.info("Compacting entry log {}", metadata.getEntryLogId()); |
| CompactionPhase scanEntryLog = new ScanEntryLogPhase(metadata); |
| if (!scanEntryLog.run()) { |
| LOG.info("Compaction for {} end in ScanEntryLogPhase.", metadata.getEntryLogId()); |
| return; |
| } |
| File compactionLogFile = entryLogger.getCurCompactionLogFile(); |
| CompactionPhase logFlushFailurePhase = new LogFlushFailurePhase(metadata.getEntryLogId()); |
| if (!logFlushFailurePhase.run()) { |
| LOG.info("Compaction for {} end in FlushCompactionLogPhase.", metadata.getEntryLogId()); |
| return; |
| } |
| File compactedLogFile = getCompactedLogFile(compactionLogFile, metadata.getEntryLogId()); |
| CompactionPhase updateIndex = new UpdateIndexPhase(compactedLogFile); |
| if (!updateIndex.run()) { |
| LOG.info("Compaction for entry log {} end in UpdateIndexPhase.", metadata.getEntryLogId()); |
| return; |
| } |
| logRemovalListener.removeEntryLog(metadata.getEntryLogId()); |
| LOG.info("Compacted entry log : {}.", metadata.getEntryLogId()); |
| } |
| |
| private class PartialFlushIndexPhase extends UpdateIndexPhase { |
| |
| public PartialFlushIndexPhase(File compactedLogFile) { |
| super(compactedLogFile); |
| } |
| |
| @Override |
| void start() throws IOException { |
| if (compactedLogFile != null && compactedLogFile.exists()) { |
| File dir = compactedLogFile.getParentFile(); |
| String compactedFilename = compactedLogFile.getName(); |
| // create a hard link "x.log" for file "x.log.y.compacted" |
| this.newEntryLogFile = new File(dir, compactedFilename.substring(0, |
| compactedFilename.indexOf(".log") + 4)); |
| File hardlinkFile = new File(dir, newEntryLogFile.getName()); |
| if (!hardlinkFile.exists()) { |
| HardLink.createHardLink(compactedLogFile, hardlinkFile); |
| } |
| assertTrue(offsets.size() > 1); |
| // only flush index for one entry location |
| EntryLocation el = offsets.get(0); |
| ledgerStorage.updateEntriesLocations(offsets); |
| ledgerStorage.flushEntriesLocationsIndex(); |
| throw new IOException("Flush ledger index encounter exception"); |
| } |
| } |
| } |
| |
| private class LogFlushFailurePhase extends FlushCompactionLogPhase { |
| |
| LogFlushFailurePhase(long compactingLogId) { |
| super(compactingLogId); |
| } |
| |
| @Override |
| void start() throws IOException { |
| // flush the current compaction log |
| entryLogger.flushCompactionLog(); |
| throw new IOException("Encounter IOException when trying to flush compaction log"); |
| } |
| } |
| } |
| |
| } |
