diff --git a/.github/workflows/pd-store-ci.yml b/.github/workflows/pd-store-ci.yml index 1a6825e7e4..1ec0c66ae5 100644 --- a/.github/workflows/pd-store-ci.yml +++ b/.github/workflows/pd-store-ci.yml @@ -295,6 +295,11 @@ jobs: mvn test -pl hugegraph-store/hg-store-test -am \ -P store-raftcore-test -Djacoco.sessionId=store-raftcore-test + - name: Run core test + run: | + mvn test -pl hugegraph-store/hg-store-test -am \ + -P store-core-test -Djacoco.sessionId=store-core-test + - name: Generate aggregate coverage report run: | mvn verify -pl hugegraph-store/hg-store-test -am -P jacoco \ @@ -311,15 +316,19 @@ jobs: "$TEST_REPORT_DIR/TEST-org.apache.hugegraph.store.rocksdb.RocksDbSuiteTest.xml" \ --require-test-report \ "$TEST_REPORT_DIR/TEST-org.apache.hugegraph.store.raftcore.RaftSuiteTest.xml" \ + --require-test-report \ + "$TEST_REPORT_DIR/TEST-org.apache.hugegraph.store.core.CoreSuiteTest.xml" \ --require-covered-group hg-store-common \ --require-covered-group hg-store-client \ --require-covered-group hg-store-rocksdb \ + --require-covered-group hg-store-core \ --require-session store-common-test \ --require-session store-client-test \ --require-session store-rocksdb-test \ --require-session store-raftcore-test \ + --require-session store-core-test \ "$REPORT_FILE" \ - hg-store-grpc hg-store-common hg-store-client hg-store-rocksdb + hg-store-grpc hg-store-common hg-store-client hg-store-rocksdb hg-store-core - name: Upload coverage to Codecov uses: codecov/codecov-action@v5 diff --git a/hugegraph-server/hugegraph-dist/src/assembly/travis/test-check-jacoco-report.sh b/hugegraph-server/hugegraph-dist/src/assembly/travis/test-check-jacoco-report.sh index bdb09ba162..133026e594 100755 --- a/hugegraph-server/hugegraph-dist/src/assembly/travis/test-check-jacoco-report.sh +++ b/hugegraph-server/hugegraph-dist/src/assembly/travis/test-check-jacoco-report.sh @@ -499,21 +499,22 @@ assert "mvn verify -pl hugegraph-store/hg-store-test -am -P jacoco \\ " \ "-DskipTests -Deditorconfig.skip=true -ntp" in " ".join(store_job.split()) assert selected_profiles(store_job, "store") == { "store-common-test", "store-client-test", "store-rocksdb-test", - "store-raftcore-test", + "store-raftcore-test", "store-core-test", } assert reports_for_option(store_job, "--require-test-report") == { "TEST-org.apache.hugegraph.store.common.CommonSuiteTest.xml", "TEST-org.apache.hugegraph.store.client.ClientSuiteTest.xml", "TEST-org.apache.hugegraph.store.rocksdb.RocksDbSuiteTest.xml", "TEST-org.apache.hugegraph.store.raftcore.RaftSuiteTest.xml", + "TEST-org.apache.hugegraph.store.core.CoreSuiteTest.xml", } assert not reports_for_option(store_job, "--require-suite-report") assert values_for_option(store_job, "--require-covered-group") == { - "hg-store-common", "hg-store-client", "hg-store-rocksdb", + "hg-store-common", "hg-store-client", "hg-store-rocksdb", "hg-store-core", } assert required_modules(store_job) == { "hg-store-grpc", "hg-store-common", "hg-store-client", - "hg-store-rocksdb", + "hg-store-rocksdb", "hg-store-core", } print("PASS: JaCoCo aggregation configuration contract") diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java index d69b36bd7c..e227808079 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandler.java @@ -230,6 +230,14 @@ void lock(String path) throws InterruptedException, void unlock(String path); + /** + * Non-blocking attempt to reserve the compactRange() window for partition {@code id}. + * Returns false if a compaction is actively running for that partition right now. + */ + boolean tryLockCompactionRange(int id); + + void unlockCompactionRange(int id); + void awaitAndSetLock(int id, int expectedValue, int value) throws InterruptedException, TimeoutException; diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java index f9ee79252d..a5aa8388cf 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/business/BusinessHandlerImpl.java @@ -40,6 +40,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.BiFunction; import java.util.function.Consumer; @@ -138,6 +139,12 @@ public class BusinessHandlerImpl implements BusinessHandler { private static final ConcurrentMap pathLock = new ConcurrentHashMap<>(); private static final ConcurrentMap compactionState = new ConcurrentHashMap<>(); + // Guards the compactRange() window specifically, so a snapshot save can atomically + // check-and-reserve against a compaction that is actually running right now. This is + // narrower than pathLock, which stays held through the post-compaction blank-task + // snapshot and must not be reused here to avoid deadlocking that flow. + private static final ConcurrentMap compactionRangeLock = + new ConcurrentHashMap<>(); // Default core thread count private static final int compactionThreadCount = 64; private static final int compactionMaxThreadCount = 256; @@ -1415,10 +1422,27 @@ public boolean dbCompaction(String graphName, int id, String tableName) { log.info("Partition {} dbCompaction started", id); if (tableName.isEmpty()) { lock(path); - setState(id, doing); - log.info("Partition {}-{} got lock, dbCompaction start", id, path); - op.compactRange(); - setState(id, compactionDone); + ReentrantLock rangeLock = + compactionRangeLock.computeIfAbsent(id, + k -> new ReentrantLock()); + if (!rangeLock.tryLock()) { + // A snapshot save is currently reserving this partition's + // range lock. Skip this compaction pass rather than block + // the compactionPool thread on it - the next scheduled/ + // triggered compaction will retry. + log.info("Partition {} skip dbCompaction, snapshot save in " + + "progress", id); + unlock(path); + return; + } + try { + setState(id, doing); + log.info("Partition {}-{} got lock, dbCompaction start", id, path); + op.compactRange(); + setState(id, compactionDone); + } finally { + rangeLock.unlock(); + } log.info("Partition {} dbCompaction end and start to do snapshot", id); PartitionEngine pe = HgStoreEngine.getInstance().getPartitionEngine(id); // find leader and send blankTask, after execution @@ -1484,6 +1508,20 @@ private boolean compareAndSetLock(String path) { return l.compareAndSet(compactionCanStart, doing); } + @Override + public boolean tryLockCompactionRange(int id) { + ReentrantLock rangeLock = compactionRangeLock.computeIfAbsent(id, k -> new ReentrantLock()); + return rangeLock.tryLock(); + } + + @Override + public void unlockCompactionRange(int id) { + ReentrantLock rangeLock = compactionRangeLock.get(id); + if (rangeLock != null) { + rangeLock.unlock(); + } + } + @Override public void awaitAndSetLock(int id, int expectedValue, int value) throws InterruptedException, TimeoutException { diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/options/RaftRocksdbOptions.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/options/RaftRocksdbOptions.java index cb88814936..2386a1f227 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/options/RaftRocksdbOptions.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/options/RaftRocksdbOptions.java @@ -44,6 +44,7 @@ public class RaftRocksdbOptions { private static RocksdbConfig rocksdbConfig = null; + private static boolean raftRocksdbConfigRegistered = false; private static RocksdbConfig getRocksdbConfig(HugeConfig options) { if (rocksdbConfig == null) { @@ -55,42 +56,55 @@ private static RocksdbConfig getRocksdbConfig(HugeConfig options) { } private static void registerRaftRocksdbConfig(HugeConfig options) { - Cache blockCache = new LRUCache(SizeUnit.GB); - BlockBasedTableConfig tableConfig = new BlockBasedTableConfig() - .setIndexType(IndexType.kTwoLevelIndexSearch) - .setPartitionFilters(true) // - .setMetadataBlockSize(8 * SizeUnit.KB) // - .setCacheIndexAndFilterBlocks( - options.get(RocksDBOptions.PUT_FILTER_AND_INDEX_IN_CACHE)) - .setCacheIndexAndFilterBlocksWithHighPriority(true) - .setPinL0FilterAndIndexBlocksInCache( - options.get(RocksDBOptions.PIN_L0_FILTER_AND_INDEX_IN_CACHE)) - .setBlockSize(4 * SizeUnit.KB) - .setBlockCache(blockCache); - - StorageOptionsFactory.registerRocksDBTableFormatConfig(RocksDBLogStorage.class, - tableConfig); - - DBOptions dbOptions = StorageOptionsFactory.getDefaultRocksDBOptions(); - dbOptions.setEnv(rocksdbConfig.getEnv()); - - // raft rocksdb number is fixed, can be controlled by max_write_buffer_number - //dbOptions.setWriteBufferManager(rocksdbConfig.getBufferManager()); - dbOptions.setUnorderedWrite(true); - StorageOptionsFactory.registerRocksDBOptions(RocksDBLogStorage.class, - dbOptions); - - ColumnFamilyOptions cfOptions = - StorageOptionsFactory.getDefaultRocksDBColumnFamilyOptions(); - cfOptions.setTargetFileSizeBase(256 * SizeUnit.MB); - cfOptions.setWriteBufferSize(8 * SizeUnit.MB); - cfOptions.setNumLevels(3); - cfOptions.setMaxWriteBufferNumber(3); - cfOptions.setCompressionType(CompressionType.NO_COMPRESSION); - cfOptions.setMaxBytesForLevelBase(2048 * SizeUnit.GB); - - StorageOptionsFactory.registerRocksDBColumnFamilyOptions(RocksDBLogStorage.class, - cfOptions); + // StorageOptionsFactory.releaseAllOptions() (called by test setup between runs) + // does not clear its table-format-config table, so registering RocksDBLogStorage's + // config more than once per JVM throws IllegalStateException. Register only once. + // The guard flag is held across the whole registration so a failure partway through + // doesn't leave the flag set to true while some options were never registered. + synchronized (RaftRocksdbOptions.class) { + if (raftRocksdbConfigRegistered) { + return; + } + + Cache blockCache = new LRUCache(SizeUnit.GB); + BlockBasedTableConfig tableConfig = new BlockBasedTableConfig() + .setIndexType(IndexType.kTwoLevelIndexSearch) + .setPartitionFilters(true) // + .setMetadataBlockSize(8 * SizeUnit.KB) // + .setCacheIndexAndFilterBlocks( + options.get(RocksDBOptions.PUT_FILTER_AND_INDEX_IN_CACHE)) + .setCacheIndexAndFilterBlocksWithHighPriority(true) + .setPinL0FilterAndIndexBlocksInCache( + options.get(RocksDBOptions.PIN_L0_FILTER_AND_INDEX_IN_CACHE)) + .setBlockSize(4 * SizeUnit.KB) + .setBlockCache(blockCache); + + StorageOptionsFactory.registerRocksDBTableFormatConfig(RocksDBLogStorage.class, + tableConfig); + + DBOptions dbOptions = StorageOptionsFactory.getDefaultRocksDBOptions(); + dbOptions.setEnv(rocksdbConfig.getEnv()); + + // raft rocksdb number is fixed, can be controlled by max_write_buffer_number + //dbOptions.setWriteBufferManager(rocksdbConfig.getBufferManager()); + dbOptions.setUnorderedWrite(true); + StorageOptionsFactory.registerRocksDBOptions(RocksDBLogStorage.class, + dbOptions); + + ColumnFamilyOptions cfOptions = + StorageOptionsFactory.getDefaultRocksDBColumnFamilyOptions(); + cfOptions.setTargetFileSizeBase(256 * SizeUnit.MB); + cfOptions.setWriteBufferSize(8 * SizeUnit.MB); + cfOptions.setNumLevels(3); + cfOptions.setMaxWriteBufferNumber(3); + cfOptions.setCompressionType(CompressionType.NO_COMPRESSION); + cfOptions.setMaxBytesForLevelBase(2048 * SizeUnit.GB); + + StorageOptionsFactory.registerRocksDBColumnFamilyOptions(RocksDBLogStorage.class, + cfOptions); + + raftRocksdbConfigRegistered = true; + } } public static void initRocksdbGlobalConfig(Map config) { diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java index 73821e4971..922b23b5a0 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/raft/PartitionStateMachine.java @@ -198,7 +198,13 @@ public void onSnapshotSave(final SnapshotWriter writer, final Closure done) { done.run(Status.OK()); } catch (HgStoreException e) { log.error(String.format("Raft %s onSnapshotSave failed. {}", groupId), e); - done.run(new Status(RaftError.EIO, e.toString())); + // A busy compaction-range lock is transient: jRaft's snapshot scheduler + // retries independently, so report EBUSY rather than EIO to avoid + // escalating to reportError()/restartRaftNode() (see SnapshotExecutorImpl + // #onSnapshotSaveDone, which only escalates on EIO). + RaftError raftError = e.getCode() == HgStoreException.EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL ? + RaftError.EBUSY : RaftError.EIO; + done.run(new Status(raftError, e.toString())); } finally { lock.unlock(); } diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/snapshot/SnapshotHandler.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/snapshot/SnapshotHandler.java index 3f26b8eedd..495a675393 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/snapshot/SnapshotHandler.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/snapshot/SnapshotHandler.java @@ -24,7 +24,6 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.Checksum; import org.apache.commons.io.FileUtils; @@ -94,35 +93,41 @@ public void onSnapshotSave(final SnapshotWriter writer) throws HgStoreException final String snapshotDir = writer.getPath(); if (partitionEngine != null) { Integer groupId = partitionEngine.getGroupId(); - AtomicInteger state = businessHandler.getState(groupId); - if (state != null && state.get() == BusinessHandler.doing) { - return; + if (!businessHandler.tryLockCompactionRange(groupId)) { + throw new HgStoreException(HgStoreException.EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL, + String.format( + "Partition %d snapshot save failed: " + + "compaction in progress", groupId)); } - // rocks db snapshot - final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; - businessHandler.saveSnapshot(graphSnapshotDir, "", groupId); - - List files = new ArrayList<>(); - File dir = new File(graphSnapshotDir); - File rootDirFile = new File(writer.getPath()); - // add all files in data dir - findFileList(dir, rootDirFile, files); - - // load snapshot by learner ?? - for (String file : files) { - String checksum = calculateChecksum(writer.getPath() + File.separator + file); - if (checksum.length() != 0) { - LocalFileMetaOutter.LocalFileMeta meta = - LocalFileMetaOutter.LocalFileMeta.newBuilder() - .setChecksum(checksum) - .build(); - writer.addFile(file, meta); - } else { - writer.addFile(file); + try { + // rocks db snapshot + final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; + businessHandler.saveSnapshot(graphSnapshotDir, "", groupId); + + List files = new ArrayList<>(); + File dir = new File(graphSnapshotDir); + File rootDirFile = new File(writer.getPath()); + // add all files in data dir + findFileList(dir, rootDirFile, files); + + // load snapshot by learner ?? + for (String file : files) { + String checksum = calculateChecksum(writer.getPath() + File.separator + file); + if (checksum.length() != 0) { + LocalFileMetaOutter.LocalFileMeta meta = + LocalFileMetaOutter.LocalFileMeta.newBuilder() + .setChecksum(checksum) + .build(); + writer.addFile(file, meta); + } else { + writer.addFile(file); + } } + // should_not_load wound not sync to learner + markShouldNotLoad(writer, true); + } finally { + businessHandler.unlockCompactionRange(groupId); } - // should_not_load wound not sync to learner - markShouldNotLoad(writer, true); } } @@ -169,6 +174,7 @@ private String calculateChecksum(String path) { public void onSnapshotLoad(final SnapshotReader reader, long committedIndex) throws HgStoreException { final String snapshotDir = reader.getPath(); + final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; // No need to load locally saved snapshots if (shouldNotLoad(reader)) { @@ -176,8 +182,15 @@ public void onSnapshotLoad(final SnapshotReader reader, long committedIndex) thr return; } + if (!new File(graphSnapshotDir).isDirectory()) { + throw new HgStoreException(HgStoreException.EC_RKDB_IMPORT_SNAPSHOT_FAIL, + String.format( + "Raft %d snapshot is corrupt, data dir %s is " + + "missing", partitionEngine.getGroupId(), + graphSnapshotDir)); + } + // Use snapshot directly - final String graphSnapshotDir = snapshotDir + File.separator + SNAPSHOT_DATA_PATH; log.info("Raft {} begin loadSnapshot, {}", partitionEngine.getGroupId(), graphSnapshotDir); businessHandler.loadSnapshot(graphSnapshotDir, "", partitionEngine.getGroupId(), committedIndex); diff --git a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/util/HgStoreException.java b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/util/HgStoreException.java index 9284361395..1311fa127e 100644 --- a/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/util/HgStoreException.java +++ b/hugegraph-store/hg-store-core/src/main/java/org/apache/hugegraph/store/util/HgStoreException.java @@ -37,6 +37,7 @@ public class HgStoreException extends RuntimeException { public static final int EC_RKDB_EXPORT_SNAPSHOT_FAIL = 1214; public static final int EC_RKDB_IMPORT_SNAPSHOT_FAIL = 1215; public static final int EC_RKDB_TRANSFER_SNAPSHOT_FAIL = 1216; + public static final int EC_RKDB_SNAPSHOT_SAVE_BUSY_FAIL = 1217; public static final int EC_METRIC_FAIL = 1401; private static final long serialVersionUID = 5193624480997934335L; private final int code; diff --git a/hugegraph-store/hg-store-test/pom.xml b/hugegraph-store/hg-store-test/pom.xml index ed91e011e3..cb9433e903 100644 --- a/hugegraph-store/hg-store-test/pom.xml +++ b/hugegraph-store/hg-store-test/pom.xml @@ -242,6 +242,12 @@ ${basedir}/target/classes/ + + false **/CoreSuiteTest.java **/BatchGraphIsolationTest.java diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java index c222557962..3d78821a63 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/BatchGraphIsolationTest.java @@ -57,7 +57,6 @@ import org.junit.Test; import org.mockito.Mockito; -import com.alipay.sofa.jraft.util.StorageOptionsFactory; import com.google.protobuf.ByteString; public class BatchGraphIsolationTest { @@ -78,7 +77,6 @@ public static void setup() throws IOException { Map rocksdbConfig = new HashMap<>(); rocksdbConfig.put("rocksdb.write_buffer_size", "1048576"); - StorageOptionsFactory.releaseAllOptions(); RaftRocksdbOptions.initRocksdbGlobalConfig(rocksdbConfig); BusinessHandlerImpl.initRocksdb(rocksdbConfig, null); diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java index 68530367a0..6afd046e18 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java @@ -17,13 +17,14 @@ package org.apache.hugegraph.store.core; +import org.apache.hugegraph.store.core.snapshot.HgSnapshotHandlerTest; +import org.junit.runner.RunWith; +import org.junit.runners.Suite; + import lombok.extern.slf4j.Slf4j; -// TODO: uncomment it until all test can run free. -//@RunWith(Suite.class) -//@Suite.SuiteClasses({ +// TODO: uncomment the rest of these classes once they can run free of each other. // HgCmdClientTest.class, -// HgSnapshotHandlerTest.class, // RaftUtilsTest.class, // RaftOperationTest.class, // UnsafeUtilTest.class, @@ -41,8 +42,10 @@ // PartitionInstructionProcessorTest.class, // // Try to put it last // HgBusinessImplTest.class -//}) - +@RunWith(Suite.class) +@Suite.SuiteClasses({ + HgSnapshotHandlerTest.class +}) @Slf4j public class CoreSuiteTest { diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/StoreEngineTestBase.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/StoreEngineTestBase.java index bce07dea5b..ee5cec32a0 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/StoreEngineTestBase.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/StoreEngineTestBase.java @@ -28,13 +28,12 @@ import org.apache.hugegraph.store.meta.Partition; import org.apache.hugegraph.store.meta.ShardGroup; import org.apache.hugegraph.store.options.HgStoreEngineOptions; +import org.apache.hugegraph.store.options.JobOptions; import org.apache.hugegraph.store.options.RaftRocksdbOptions; import org.apache.hugegraph.store.pd.FakePdServiceProvider; import org.junit.AfterClass; import org.junit.BeforeClass; -import com.alipay.sofa.jraft.util.StorageOptionsFactory; - import lombok.extern.slf4j.Slf4j; /** @@ -61,6 +60,11 @@ public static void initEngine() { options.setGrpcAddress("127.0.0.1:6511"); options.setRaftAddress("127.0.0.1:6510"); options.setDataTransfer(new DataManagerImpl()); + JobOptions jobOptions = new JobOptions(); + jobOptions.setUninterruptibleCore(2); + jobOptions.setUninterruptibleMax(8); + jobOptions.setUninterruptibleQueueSize(1024); + options.setJobConfig(jobOptions); options.setFakePdOptions(new HgStoreEngineOptions.FakePdOptions() {{ setStoreList("127.0.0.1"); @@ -70,7 +74,6 @@ public static void initEngine() { }}); if (initCount == 0) { - StorageOptionsFactory.releaseAllOptions(); RaftRocksdbOptions.initRocksdbGlobalConfig(options.getRocksdbConfig()); initCount++; } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java index ff5ef24acf..b60cdfd303 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/HgSnapshotHandlerTest.java @@ -21,27 +21,36 @@ import java.io.File; import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.io.FileUtils; +import org.apache.hugegraph.store.business.BusinessHandler; import org.apache.hugegraph.store.core.StoreEngineTestBase; import org.apache.hugegraph.store.meta.Partition; import org.apache.hugegraph.store.snapshot.HgSnapshotHandler; +import org.apache.hugegraph.store.snapshot.SnapshotHandler; import org.junit.Before; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; import com.alipay.sofa.jraft.entity.RaftOutter; import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader; import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter; import com.google.protobuf.Message; - public class HgSnapshotHandlerTest extends StoreEngineTestBase { private static HgSnapshotHandler hgSnapshotHandlerUnderTest; + @Rule + public TemporaryFolder tmpDir = new TemporaryFolder(); + @Before public void setUp() throws IOException { hgSnapshotHandlerUnderTest = new HgSnapshotHandler(createPartitionEngine(0)); @@ -184,4 +193,88 @@ public void testFindFileList() { // Verify the results } + + /** + * Test that onSnapshotLoad skips loading (rather than throwing) when should_not_load is + * present but data/ is missing: a locally-saved snapshot deliberately has no data/ dir + * since nothing was meant to load, and should_not_load is checked before the data/ dir. + */ + @Test + public void testOnSnapshotLoadSkipsWhenShouldNotLoadPresentButDataMissing() + throws Exception { + // Arrange: snapshot dir has should_not_load but NO data/ subdirectory. + File snapDir = tmpDir.newFolder("snapshot-corrupt"); + File shouldNotLoad = new File(snapDir, "should_not_load"); + Files.write(shouldNotLoad.toPath(), "saved snapshot".getBytes(StandardCharsets.UTF_8)); + // data/ deliberately not created + + SnapshotHandler handler = new SnapshotHandler(createPartitionEngine(1)); + SnapshotReader stubReader = stubReader(snapDir.getAbsolutePath()); + + // Must not throw; should return early at the should_not_load check. + handler.onSnapshotLoad(stubReader, 0L); + } + + /** + * Test that onSnapshotLoad skips loading when snapshot is locally saved (both flags present). + */ + @Test + public void testOnSnapshotLoadSkipsWhenShouldNotLoadPresentAndDataExists() throws Exception { + // Arrange: a healthy local snapshot, both should_not_load and data/ present. + File snapDir = tmpDir.newFolder("snapshot-healthy"); + File shouldNotLoad = new File(snapDir, "should_not_load"); + Files.write(shouldNotLoad.toPath(), "saved snapshot".getBytes(StandardCharsets.UTF_8)); + FileUtils.forceMkdir(new File(snapDir, "data")); + + SnapshotHandler handler = new SnapshotHandler(createPartitionEngine(2)); + SnapshotReader stubReader = stubReader(snapDir.getAbsolutePath()); + + // Must not throw; should return early at the should_not_load + data-exists check. + handler.onSnapshotLoad(stubReader, 0L); + } + + /** + * Test that the compaction-range lock used by onSnapshotSave to guard against a concurrent + * compactRange() call is mutually exclusive and releasable, using the real BusinessHandlerImpl + * rather than a mock, so the actual lock instance backing the check is exercised. The + * concurrent attempt runs on a separate thread because the lock is a ReentrantLock: the + * owning thread can always re-acquire it, so checking from the same thread would not + * exercise exclusion. In production dbCompaction() and onSnapshotSave() run on different + * executor threads, which is what this mirrors. + */ + @Test + public void testCompactionRangeLockIsMutuallyExclusiveAndReleasable() throws InterruptedException { + BusinessHandler businessHandler = getStoreEngine().getBusinessHandler(); + int partitionId = 3; + + assertEquals("first reservation must succeed", true, + businessHandler.tryLockCompactionRange(partitionId)); + + AtomicBoolean concurrentResult = new AtomicBoolean(); + Thread other = new Thread( + () -> concurrentResult.set(businessHandler.tryLockCompactionRange(partitionId))); + other.start(); + other.join(); + assertEquals("a concurrent reservation from another thread must fail while the first " + + "is held", false, concurrentResult.get()); + + businessHandler.unlockCompactionRange(partitionId); + + assertEquals("reservation must succeed again once released", true, + businessHandler.tryLockCompactionRange(partitionId)); + businessHandler.unlockCompactionRange(partitionId); + } + + private static SnapshotReader stubReader(String path) { + return new SnapshotReader() { + @Override public RaftOutter.SnapshotMeta load() { return null; } + @Override public String generateURIForCopy() { return null; } + @Override public boolean init(Void opts) { return false; } + @Override public void shutdown() {} + @Override public String getPath() { return path; } + @Override public Set listFiles() { return null; } + @Override public Message getFileMeta(String fileName) { return null; } + @Override public void close() {} + }; + } } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/SnapshotHandlerTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/SnapshotHandlerTest.java new file mode 100644 index 0000000000..ca888f7035 --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/snapshot/SnapshotHandlerTest.java @@ -0,0 +1,175 @@ +/* + * 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.hugegraph.store.core.snapshot; + +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.contains; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.util.Set; + +import org.apache.hugegraph.store.HgStoreEngine; +import org.apache.hugegraph.store.PartitionEngine; +import org.apache.hugegraph.store.business.BusinessHandler; +import org.apache.hugegraph.store.snapshot.SnapshotHandler; +import org.apache.hugegraph.store.util.HgStoreException; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import com.alipay.sofa.jraft.entity.RaftOutter; +import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader; +import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter; +import com.google.protobuf.Message; + +public class SnapshotHandlerTest { + + @Rule + public TemporaryFolder tmpDir = new TemporaryFolder(); + + /** + * When the compaction-range lock cannot be reserved (a compaction is actively running), + * onSnapshotSave must throw immediately without touching saveSnapshot. The exception signals + * jRaft, which will retry the snapshot later. jRaft's snapshot scheduler runs independently + * and frequently (default 300s, user config 1800s), so the next attempt will succeed once + * compaction releases the lock. + */ + @Test + public void testOnSnapshotSaveThrowsWhenCompactionInProgress() { + PartitionEngine mockEngine = mock(PartitionEngine.class); + HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); + BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); + + when(mockEngine.getGroupId()).thenReturn(0); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + when(mockBusinessHandler.tryLockCompactionRange(0)).thenReturn(false); + + SnapshotHandler handler = new SnapshotHandler(mockEngine); + SnapshotWriter stubWriter = stubWriter("/tmp/snapshot"); + + HgStoreException ex = assertThrows( + "onSnapshotSave must throw when the compaction-range lock is held", + HgStoreException.class, + () -> handler.onSnapshotSave(stubWriter)); + + assertTrue("Exception message must mention the partition", + ex.getMessage().contains("0")); + assertTrue("Exception message must mention compaction is in progress", + ex.getMessage().contains("compaction in progress")); + verify(mockBusinessHandler, never()).saveSnapshot(any(), any(), anyInt()); + } + + /** + * When the compaction-range lock is free, onSnapshotSave must reserve it, call saveSnapshot, + * and release the lock afterwards. + */ + @Test + public void testOnSnapshotSaveCallsSaveSnapshotWhenNotBusy() throws Exception { + PartitionEngine mockEngine = mock(PartitionEngine.class); + HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); + BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); + + final String snapshotPath = tmpDir.newFolder("snap-not-busy").getAbsolutePath(); + + when(mockEngine.getGroupId()).thenReturn(0); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + when(mockBusinessHandler.tryLockCompactionRange(0)).thenReturn(true); + + SnapshotHandler handler = new SnapshotHandler(mockEngine); + SnapshotWriter stubWriter = stubWriter(snapshotPath); + + handler.onSnapshotSave(stubWriter); + + // Verify: saveSnapshot was called with concrete path containing expected data dir + String expectedDataDir = snapshotPath + File.separator + "data"; + verify(mockBusinessHandler).saveSnapshot( + contains(expectedDataDir), // Must contain the snapshot path + /data + eq(""), // graphName (empty string) + eq(0)); // groupId (partition 0) + verify(mockBusinessHandler).unlockCompactionRange(0); + } + + /** + * When should_not_load is absent (the common corruption variant: leader crashed + * mid-checkpoint with no flag ever written) and data/ is missing, onSnapshotLoad + * must throw a diagnostic naming the corrupt snapshot directory, rather than + * falling through to businessHandler.loadSnapshot. + */ + @Test + public void testOnSnapshotLoadThrowsWhenShouldNotLoadAbsentAndDataMissing() throws Exception { + PartitionEngine mockEngine = mock(PartitionEngine.class); + HgStoreEngine mockStoreEngine = mock(HgStoreEngine.class); + BusinessHandler mockBusinessHandler = mock(BusinessHandler.class); + + when(mockEngine.getGroupId()).thenReturn(3); + when(mockEngine.getStoreEngine()).thenReturn(mockStoreEngine); + when(mockStoreEngine.getBusinessHandler()).thenReturn(mockBusinessHandler); + + String snapshotPath = tmpDir.newFolder("snapshot-no-flag-no-data").getAbsolutePath(); + // should_not_load deliberately not created; data/ deliberately not created + + SnapshotHandler handler = new SnapshotHandler(mockEngine); + SnapshotReader stubReader = stubReader(snapshotPath); + + HgStoreException ex = assertThrows( + "onSnapshotLoad must throw when data/ is missing, flag or no flag", + HgStoreException.class, + () -> handler.onSnapshotLoad(stubReader, 0L)); + + assertTrue("Exception message must name the corrupt snapshot directory", + ex.getMessage().contains(snapshotPath)); + } + + private static SnapshotWriter stubWriter(String path) { + return new SnapshotWriter() { + @Override public boolean saveMeta(RaftOutter.SnapshotMeta meta) { return false; } + @Override public boolean addFile(String fileName, Message fileMeta) { return false; } + @Override public boolean removeFile(String fileName) { return false; } + @Override public void close(boolean keepDataOnError) {} + @Override public boolean init(Void opts) { return false; } + @Override public void shutdown() {} + @Override public String getPath() { return path; } + @Override public Set listFiles() { return null; } + @Override public Message getFileMeta(String fileName) { return null; } + @Override public void close() {} + }; + } + + private static SnapshotReader stubReader(String path) { + return new SnapshotReader() { + @Override public RaftOutter.SnapshotMeta load() { return null; } + @Override public String generateURIForCopy() { return null; } + @Override public boolean init(Void opts) { return false; } + @Override public void shutdown() {} + @Override public String getPath() { return path; } + @Override public Set listFiles() { return null; } + @Override public Message getFileMeta(String fileName) { return null; } + @Override public void close() {} + }; + } +} diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/raftcore/RaftSuiteTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/raftcore/RaftSuiteTest.java index f3b1f31d29..3721000ec5 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/raftcore/RaftSuiteTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/raftcore/RaftSuiteTest.java @@ -17,13 +17,15 @@ package org.apache.hugegraph.store.raftcore; +import org.apache.hugegraph.store.core.snapshot.SnapshotHandlerTest; import org.junit.runner.RunWith; import org.junit.runners.Suite; @RunWith(Suite.class) @Suite.SuiteClasses({ BytesCarrierTest.class, - ZeroByteStringHelperTest.class + ZeroByteStringHelperTest.class, + SnapshotHandlerTest.class }) public class RaftSuiteTest {