You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ni...@apache.org on 2025/02/19 09:33:26 UTC
(ignite) branch master updated: IGNITE-24481 Use SnapshotFileTree in LocalSnapshotSender (#11879)
This is an automated email from the ASF dual-hosted git repository.
nizhikov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 2cc0302c68b IGNITE-24481 Use SnapshotFileTree in LocalSnapshotSender (#11879)
2cc0302c68b is described below
commit 2cc0302c68bd2360f3d2ef44bbafd37758b479a0
Author: Nikolay <ni...@apache.org>
AuthorDate: Wed Feb 19 12:33:16 2025 +0300
IGNITE-24481 Use SnapshotFileTree in LocalSnapshotSender (#11879)
---
.../java/org/apache/ignite/dump/DumpReader.java | 4 +-
.../internal/MarshallerMappingFileStore.java | 3 +-
.../apache/ignite/internal/binary/BinaryUtils.java | 20 ++-----
.../org/apache/ignite/internal/cdc/CdcMain.java | 5 +-
.../cache/binary/BinaryMetadataFileStore.java | 3 +-
.../cache/persistence/filename/NodeFileTree.java | 16 +++++
.../snapshot/IgniteSnapshotManager.java | 68 ++++++++++------------
.../snapshot/IncrementalSnapshotFutureTask.java | 7 +--
.../IncrementalSnapshotVerificationTask.java | 11 +---
.../snapshot/SnapshotRestoreProcess.java | 2 +-
.../snapshot/AbstractSnapshotSelfTest.java | 6 +-
.../snapshot/EncryptedSnapshotTest.java | 7 ++-
.../snapshot/IgniteClusterSnapshotDeltaTest.java | 9 +--
.../snapshot/IgniteClusterSnapshotSelfTest.java | 14 ++---
.../snapshot/IgniteSnapshotManagerSelfTest.java | 24 +++++---
.../persistence/snapshot/PlainSnapshotTest.java | 6 +-
16 files changed, 103 insertions(+), 102 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/dump/DumpReader.java b/modules/core/src/main/java/org/apache/ignite/dump/DumpReader.java
index 72499b499f8..55cebeb7360 100644
--- a/modules/core/src/main/java/org/apache/ignite/dump/DumpReader.java
+++ b/modules/core/src/main/java/org/apache/ignite/dump/DumpReader.java
@@ -32,8 +32,8 @@ import java.util.stream.Collectors;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.GridLoggerProxy;
-import org.apache.ignite.internal.binary.BinaryUtils;
import org.apache.ignite.internal.cdc.CdcMain;
+import org.apache.ignite.internal.processors.cache.persistence.filename.NodeFileTree;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadata;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.Dump;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.Dump.DumpedPartitionIterator;
@@ -86,7 +86,7 @@ public class DumpReader implements Runnable {
cnsmr.start();
try {
- File[] files = F.first(dump.fileTrees()).marshaller().listFiles(BinaryUtils::notTmpFile);
+ File[] files = F.first(dump.fileTrees()).marshaller().listFiles(NodeFileTree::notTmpFile);
if (files != null)
cnsmr.onMappings(CdcMain.typeMappingIterator(files, tm -> true));
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/MarshallerMappingFileStore.java b/modules/core/src/main/java/org/apache/ignite/internal/MarshallerMappingFileStore.java
index ddc40c37c79..d838846d996 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/MarshallerMappingFileStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/MarshallerMappingFileStore.java
@@ -29,6 +29,7 @@ import java.util.concurrent.locks.Lock;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.binary.BinaryUtils;
+import org.apache.ignite.internal.processors.cache.persistence.filename.NodeFileTree;
import org.apache.ignite.internal.processors.cache.persistence.filename.SharedFileTree;
import org.apache.ignite.internal.util.GridStripedLock;
import org.apache.ignite.internal.util.IgniteUtils;
@@ -167,7 +168,7 @@ final class MarshallerMappingFileStore {
* @param marshCtx Marshaller context to register mappings.
*/
void restoreMappings(MarshallerContext marshCtx) throws IgniteCheckedException {
- File[] files = mappingDir.listFiles(BinaryUtils::notTmpFile);
+ File[] files = mappingDir.listFiles(NodeFileTree::notTmpFile);
if (files == null)
return;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
index 2f5ed83df9a..f106a50dcd8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryUtils.java
@@ -68,6 +68,7 @@ import org.apache.ignite.binary.Binarylizable;
import org.apache.ignite.internal.binary.builder.BinaryLazyValue;
import org.apache.ignite.internal.binary.streams.BinaryInputStream;
import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.persistence.filename.NodeFileTree;
import org.apache.ignite.internal.util.MutableSingletonList;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -78,15 +79,12 @@ import org.jetbrains.annotations.Nullable;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_BINARY_MARSHALLER_USE_STRING_SERIALIZATION_VER_2;
-import static org.apache.ignite.internal.processors.cache.persistence.filename.NodeFileTree.TMP_SUFFIX;
+import static org.apache.ignite.internal.processors.cache.persistence.filename.NodeFileTree.FILE_SUFFIX;
/**
* Binary utils.
*/
public class BinaryUtils {
- /** Binary metadata file suffix. */
- public static final String METADATA_FILE_SUFFIX = ".bin";
-
/**
* Actual file name "{type_id}.classname{platform_id}".
* Where {@code type_id} is integer type id and {@code platform_id} is byte from {@link PlatformType}
@@ -2547,7 +2545,7 @@ public class BinaryUtils {
* @return Binary metadata file name.
*/
public static String binaryMetaFileName(int typeId) {
- return typeId + METADATA_FILE_SUFFIX;
+ return typeId + FILE_SUFFIX;
}
/** @param fileName Name of file with marshaller mapping information. */
@@ -2598,22 +2596,14 @@ public class BinaryUtils {
return typeId + MAPPING_FILE_EXTENSION + platformId;
}
- /**
- * @param f File.
- * @return {@code True} if file is regular(not temporary).
- */
- public static boolean notTmpFile(File f) {
- return !f.getName().endsWith(TMP_SUFFIX);
- }
-
/**
* @param fileName File name.
* @return Type id
* @see #binaryMetaFileName(int)
- * @see #METADATA_FILE_SUFFIX
+ * @see NodeFileTree#FILE_SUFFIX
*/
public static int typeId(String fileName) {
- return Integer.parseInt(fileName.substring(0, fileName.length() - METADATA_FILE_SUFFIX.length()));
+ return Integer.parseInt(fileName.substring(0, fileName.length() - FILE_SUFFIX.length()));
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java
index c2d572ca7c8..62c2ab37dfd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java
@@ -81,7 +81,6 @@ import static org.apache.ignite.internal.IgniteKernal.SITE;
import static org.apache.ignite.internal.IgniteVersionUtils.ACK_VER_STR;
import static org.apache.ignite.internal.IgniteVersionUtils.COPYRIGHT;
import static org.apache.ignite.internal.IgnitionEx.initializeDefaultMBeanServer;
-import static org.apache.ignite.internal.binary.BinaryUtils.METADATA_FILE_SUFFIX;
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.CDC_DATA_RECORD;
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.CDC_MANAGER_RECORD;
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.CDC_MANAGER_STOP_RECORD;
@@ -678,7 +677,7 @@ public class CdcMain implements Runnable {
return;
Iterator<BinaryType> changedTypes = Arrays.stream(files)
- .filter(p -> p.toString().endsWith(METADATA_FILE_SUFFIX))
+ .filter(NodeFileTree::binFile)
.map(f -> {
int typeId = BinaryUtils.typeId(f.getName());
long lastModified = f.lastModified();
@@ -719,7 +718,7 @@ public class CdcMain implements Runnable {
/** Search for new or changed {@link TypeMapping} and notifies the consumer. */
private void updateMappings() {
try {
- File[] files = ft.marshaller().listFiles(BinaryUtils::notTmpFile);
+ File[] files = ft.marshaller().listFiles(NodeFileTree::notTmpFile);
if (files == null)
return;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataFileStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataFileStore.java
index 75de1d67028..ce3f352d886 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataFileStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/BinaryMetadataFileStore.java
@@ -39,6 +39,7 @@ import org.apache.ignite.internal.binary.BinaryMetadata;
import org.apache.ignite.internal.binary.BinaryUtils;
import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
+import org.apache.ignite.internal.processors.cache.persistence.filename.NodeFileTree;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -210,7 +211,7 @@ class BinaryMetadataFileStore {
if (!enabled)
return;
- for (File file : metadataDir.listFiles(BinaryUtils::notTmpFile))
+ for (File file : metadataDir.listFiles(NodeFileTree::notTmpFile))
restoreMetadata(file);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/filename/NodeFileTree.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/filename/NodeFileTree.java
index 4599ea3dcd8..1fc71de27fa 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/filename/NodeFileTree.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/filename/NodeFileTree.java
@@ -606,6 +606,14 @@ public class NodeFileTree extends SharedFileTree {
return f.getName().equals(CACHE_DATA_FILENAME);
}
+ /**
+ * @param f File.
+ * @return {@code True} if file conforms cache config file name pattern.
+ */
+ public static boolean binFile(File f) {
+ return f.getName().endsWith(FILE_SUFFIX);
+ }
+
/**
* @param f File.
* @return {@code True} if file conforms temp cache storage name pattern.
@@ -614,6 +622,14 @@ public class NodeFileTree extends SharedFileTree {
return f.isDirectory() && f.getName().startsWith(TMP_CACHE_DIR_PREFIX);
}
+ /**
+ * @param f File.
+ * @return {@code True} if file is regular(not temporary).
+ */
+ public static boolean notTmpFile(File f) {
+ return !f.getName().endsWith(TMP_SUFFIX);
+ }
+
/**
* @param f Temporary cache directory.
* @return Cache or group id.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
index 62271e8d735..5d59fe5e2d5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
@@ -376,7 +376,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
private volatile ReadWriteMetastorage metaStorage;
/** Local snapshot sender factory. */
- private BiFunction<String, String, SnapshotSender> locSndrFactory = LocalSnapshotSender::new;
+ private Function<SnapshotFileTree, SnapshotSender> locSndrFactory = LocalSnapshotSender::new;
/** Remote snapshot sender factory. */
private BiFunction<String, UUID, SnapshotSender> rmtSndrFactory = this::remoteSnapshotSenderFactory;
@@ -1040,7 +1040,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
req.dump(),
req.compress(),
req.encrypt(),
- locSndrFactory.apply(req.snapshotName(), req.snapshotPath())
+ locSndrFactory.apply(req.snapshotFileTree())
);
if (withMetaStorage) {
@@ -1059,9 +1059,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
.map(n -> cctx.discovery().node(n).consistentId().toString())
.collect(Collectors.toSet());
- File snpDir = snapshotLocalDir(req.snapshotName(), req.snapshotPath());
-
- snpDir.mkdirs();
+ req.snapshotFileTree().root().mkdirs();
SnapshotFutureTaskResult res = (SnapshotFutureTaskResult)task0.result();
@@ -1087,16 +1085,14 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
encKey == null ? null : encSpi.encryptKey(encKey)
);
- SnapshotHandlerContext ctx = new SnapshotHandlerContext(meta, req.groups(), cctx.localNode(), snpDir,
+ SnapshotHandlerContext ctx = new SnapshotHandlerContext(meta, req.groups(), cctx.localNode(), req.snapshotFileTree().root(),
req.streamerWarning(), true);
req.meta(meta);
- File smf = new File(snpDir, snapshotMetaFileName(cctx.localNode().consistentId().toString()));
-
- storeSnapshotMeta(req.meta(), smf);
+ storeSnapshotMeta(req.meta(), req.snapshotFileTree().meta());
- log.info("Snapshot metafile has been created: " + smf.getAbsolutePath());
+ log.info("Snapshot metafile has been created: " + req.snapshotFileTree().meta().getAbsolutePath());
return new SnapshotOperationResponse(handlers.invokeAll(SnapshotHandlerType.CREATE, ctx));
}
@@ -2643,14 +2639,14 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/**
* @param factory Factory which produces {@link LocalSnapshotSender} implementation.
*/
- void localSnapshotSenderFactory(BiFunction<String, String, SnapshotSender> factory) {
+ void localSnapshotSenderFactory(Function<SnapshotFileTree, SnapshotSender> factory) {
locSndrFactory = factory;
}
/**
* @return Factory which produces {@link LocalSnapshotSender} implementation.
*/
- BiFunction<String, String, SnapshotSender> localSnapshotSenderFactory() {
+ Function<SnapshotFileTree, SnapshotSender> localSnapshotSenderFactory() {
return locSndrFactory;
}
@@ -3930,11 +3926,8 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
* Snapshot sender which writes all data to local directory.
*/
private class LocalSnapshotSender extends SnapshotSender {
- /** Local snapshot directory. */
- private final File snpLocDir;
-
- /** Local node snapshot directory calculated on snapshot directory. */
- private File dbDir;
+ /** Snapshot file tree. */
+ private final SnapshotFileTree sft;
/** Size of page. */
private final int pageSize;
@@ -3944,29 +3937,26 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
sequentialWrite() ? DeltaSortedIterator::new : DeltaIterator::new;
/**
- * @param snpName Snapshot name.
- * @param snpPath Snapshot directory path.
+ * @param sft Snapshot file tree.
*/
- public LocalSnapshotSender(String snpName, @Nullable String snpPath) {
+ public LocalSnapshotSender(SnapshotFileTree sft) {
super(IgniteSnapshotManager.this.log, cctx.kernalContext().pools().getSnapshotExecutorService());
- snpLocDir = snapshotLocalDir(snpName, snpPath);
+ this.sft = sft;
pageSize = cctx.kernalContext().config().getDataStorageConfiguration().getPageSize();
}
/** {@inheritDoc} */
@Override protected void init(int partsCnt) {
- dbDir = new File(snpLocDir, databaseRelativePath(pdsSettings.folderName()));
-
- if (dbDir.exists()) {
+ if (sft.nodeStorage().exists()) {
throw new IgniteException("Snapshot with given name already exists " +
- "[snpName=" + snpLocDir.getName() + ", absPath=" + dbDir.getAbsolutePath() + ']');
+ "[snpName=" + sft.root().getName() + ", absPath=" + sft.nodeStorage().getAbsolutePath() + ']');
}
- writeSnapshotDirectoryToMetastorage(snpLocDir);
+ writeSnapshotDirectoryToMetastorage(sft.root());
try {
- U.ensureDirectory(dbDir, "snapshot work directory for a local snapshot sender", log);
+ U.ensureDirectory(sft.nodeStorage(), "snapshot work directory for a local snapshot sender", log);
}
catch (IgniteCheckedException e) {
throw new IgniteException(e);
@@ -3975,10 +3965,10 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/** {@inheritDoc} */
@Override public void sendCacheConfig0(File ccfg, String cacheDirName) {
- assert dbDir != null;
-
try {
- File cacheDir = U.resolveWorkDirectory(dbDir.getAbsolutePath(), cacheDirName, false);
+ File cacheDir = sft.cacheStorage(cacheDirName);
+
+ U.mkdirs(cacheDir);
File targetCacheCfg = new File(cacheDir, ccfg.getName());
@@ -4007,7 +3997,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
return;
try {
- saveMappings(cctx.kernalContext(), mappings, snpLocDir);
+ saveMappings(cctx.kernalContext(), mappings, sft.root());
}
catch (IgniteCheckedException e) {
throw new IgniteException(e);
@@ -4019,7 +4009,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
if (types == null)
return;
- cctx.kernalContext().cacheObjects().saveMetadata(types, snpLocDir);
+ cctx.kernalContext().cacheObjects().saveMetadata(types, sft.root());
}
/** {@inheritDoc} */
@@ -4028,7 +4018,9 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
if (len == 0)
return;
- File cacheDir = U.resolveWorkDirectory(dbDir.getAbsolutePath(), cacheDirName, false);
+ File cacheDir = sft.cacheStorage(cacheDirName);
+
+ U.mkdirs(cacheDir);
File snpPart = new File(cacheDir, part.getName());
@@ -4038,19 +4030,19 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
copy(ioFactory, part, snpPart, len, transferRateLimiter);
if (log.isDebugEnabled()) {
- log.debug("Partition has been snapshot [snapshotDir=" + dbDir.getAbsolutePath() +
+ log.debug("Partition has been snapshot [snapshotDir=" + sft.nodeStorage().getAbsolutePath() +
", cacheDirName=" + cacheDirName + ", part=" + part.getName() +
", length=" + part.length() + ", snapshot=" + snpPart.getName() + ']');
}
}
- catch (IOException | IgniteCheckedException ex) {
+ catch (IOException ex) {
throw new IgniteException(ex);
}
}
/** {@inheritDoc} */
@Override public void sendDelta0(File delta, String cacheDirName, GroupPartitionId pair) {
- File snpPart = partitionFile(dbDir, cacheDirName, pair.getPartitionId());
+ File snpPart = sft.partitionFile(cacheDirName, pair.getPartitionId());
if (log.isDebugEnabled()) {
log.debug("Start partition snapshot recovery with the given delta page file [part=" + snpPart +
@@ -4089,10 +4081,10 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
@Override protected void close0(@Nullable Throwable th) {
if (th == null) {
if (log.isInfoEnabled())
- log.info("The Local snapshot sender closed. All resources released [dbNodeSnpDir=" + dbDir + ']');
+ log.info("The Local snapshot sender closed. All resources released [dbNodeSnpDir=" + sft.nodeStorage() + ']');
}
else {
- deleteSnapshot(snpLocDir);
+ deleteSnapshot(sft.root());
if (log.isDebugEnabled())
log.debug("Local snapshot sender closed due to an error occurred: " + th.getMessage());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotFutureTask.java
index b72632d9bd1..ec47d7375d9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotFutureTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotFutureTask.java
@@ -29,7 +29,6 @@ import java.util.function.BiConsumer;
import org.apache.ignite.IgniteException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
-import org.apache.ignite.internal.binary.BinaryUtils;
import org.apache.ignite.internal.pagemem.wal.record.IncrementalSnapshotFinishRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.ClusterSnapshotRecord;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
@@ -39,8 +38,6 @@ import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPa
import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
import org.apache.ignite.internal.util.typedef.internal.CU;
-import static org.apache.ignite.internal.binary.BinaryUtils.METADATA_FILE_SUFFIX;
-
/** */
class IncrementalSnapshotFutureTask extends AbstractSnapshotFutureTask<Void> implements BiConsumer<String, File> {
/** Incremental file tree. */
@@ -132,13 +129,13 @@ class IncrementalSnapshotFutureTask extends AbstractSnapshotFutureTask<Void> imp
copyFiles(
ft.marshaller(),
ift.marshaller(),
- BinaryUtils::notTmpFile
+ NodeFileTree::notTmpFile
);
copyFiles(
ft.binaryMeta(),
ift.binaryMeta(),
- file -> file.getName().endsWith(METADATA_FILE_SUFFIX)
+ NodeFileTree::binFile
);
onDone();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTask.java
index 2cb7d81f302..485f3e799bb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTask.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.processors.cache.persistence.snapshot;
-import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -56,7 +55,6 @@ import org.apache.ignite.transactions.TransactionState;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.managers.discovery.ConsistentIdMapper.ALL_NODES;
-import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.databaseRelativePath;
/** */
@GridInternal
@@ -371,8 +369,7 @@ public class IncrementalSnapshotVerificationTask extends AbstractSnapshotVerific
private void checkBaseline(BaselineTopology blt) throws IgniteCheckedException, IOException {
IgniteSnapshotManager snpMgr = ignite.context().cache().context().snapshotMgr();
- File snpDir = snpMgr.snapshotLocalDir(snpName, snpPath);
- SnapshotMetadata meta = snpMgr.readSnapshotMetadata(snpDir, ignite.localNode().consistentId().toString());
+ SnapshotMetadata meta = snpMgr.readSnapshotMetadata(sft.meta());
if (!F.eqNotOrdered(blt.consistentIds(), meta.baselineNodes())) {
throw new IgniteCheckedException("Topologies of snapshot and current cluster are different [snp=" +
@@ -382,12 +379,8 @@ public class IncrementalSnapshotVerificationTask extends AbstractSnapshotVerific
/** @return Collection of snapshotted transactional caches, key is a cache ID. */
private Map<Integer, StoredCacheData> readTxCachesData() throws IgniteCheckedException, IOException {
- File snpDir = ignite.context().cache().context().snapshotMgr().snapshotLocalDir(snpName, snpPath);
-
- String folderName = ignite.context().pdsFolderResolver().resolveFolders().folderName();
-
return GridLocalConfigManager.readCachesData(
- new File(snpDir, databaseRelativePath(folderName)),
+ sft.nodeStorage(),
ignite.context().marshallerContext().jdkMarshaller(),
ignite.configuration())
.values().stream()
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java
index 1973f2ef738..c5098106e0e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java
@@ -1176,7 +1176,7 @@ public class SnapshotRestoreProcess {
/** Restore registered mappings for user classes. */
private void restoreMappings(File marshallerDir, BooleanSupplier stopChecker) throws IgniteCheckedException {
- File[] mappings = marshallerDir.listFiles(BinaryUtils::notTmpFile);
+ File[] mappings = marshallerDir.listFiles(NodeFileTree::notTmpFile);
if (mappings == null)
throw new IgniteException("Failed to list marshaller directory [dir=" + marshallerDir + ']');
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java
index e5d7871cc3a..3add1d6252c 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java
@@ -744,13 +744,13 @@ public abstract class AbstractSnapshotSelfTest extends GridCommonAbstractTest {
for (Ignite grid : grids) {
IgniteSnapshotManager mgr = snp((IgniteEx)grid);
- BiFunction<String, String, SnapshotSender> old = mgr.localSnapshotSenderFactory();
+ Function<SnapshotFileTree, SnapshotSender> old = mgr.localSnapshotSenderFactory();
BlockingExecutor block = new BlockingExecutor(mgr.snapshotExecutorService());
execs.add(block);
- mgr.localSnapshotSenderFactory((snpName, snpPath) ->
- new DelegateSnapshotSender(log, block, old.apply(snpName, snpPath)));
+ mgr.localSnapshotSenderFactory(sft ->
+ new DelegateSnapshotSender(log, block, old.apply(sft)));
}
return execs;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/EncryptedSnapshotTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/EncryptedSnapshotTest.java
index e68470e8c1f..13b97637eb2 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/EncryptedSnapshotTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/EncryptedSnapshotTest.java
@@ -29,6 +29,7 @@ import org.apache.ignite.cluster.ClusterState;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.encryption.AbstractEncryptionTest;
+import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
import org.apache.ignite.internal.util.distributed.FullMessage;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
@@ -282,9 +283,11 @@ public class EncryptedSnapshotTest extends AbstractSnapshotSelfTest {
// Start grid node with data before each test.
IgniteEx ig = startGridsWithCache(1, CACHE_KEYS_RANGE, valueBuilder(), dfltCacheCfg);
+ SnapshotFileTree sft = snapshotFileTree(ig, SNAPSHOT_NAME);
+
assertThrowsAnyCause(log,
() -> snp(ig).registerSnapshotTask(
- snapshotFileTree(ig, SNAPSHOT_NAME),
+ sft,
ig.localNode().id(),
null,
F.asMap(CU.cacheId(dfltCacheCfg.getName()), null),
@@ -292,7 +295,7 @@ public class EncryptedSnapshotTest extends AbstractSnapshotSelfTest {
false,
false,
false,
- snp(ig).localSnapshotSenderFactory().apply(SNAPSHOT_NAME, null)
+ snp(ig).localSnapshotSenderFactory().apply(sft)
).get(TIMEOUT),
IgniteCheckedException.class,
"Metastore is required because it holds encryption keys");
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotDeltaTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotDeltaTest.java
index f1e116bef55..90877ba6e8d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotDeltaTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotDeltaTest.java
@@ -27,7 +27,7 @@ import java.util.Collection;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
-import java.util.function.BiFunction;
+import java.util.function.Function;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.configuration.CacheConfiguration;
@@ -37,6 +37,7 @@ import org.apache.ignite.internal.processors.cache.persistence.file.FileIODecora
import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
import org.apache.ignite.internal.processors.cache.persistence.filename.NodeFileTree;
+import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
@@ -113,13 +114,13 @@ public class IgniteClusterSnapshotDeltaTest extends AbstractSnapshotSelfTest {
IgniteSnapshotManager mgr = snp(srv);
- BiFunction<String, String, SnapshotSender> old = mgr.localSnapshotSenderFactory();
+ Function<SnapshotFileTree, SnapshotSender> old = mgr.localSnapshotSenderFactory();
CountDownLatch partStart = new CountDownLatch(partCnt);
CountDownLatch deltaApply = new CountDownLatch(1);
- mgr.localSnapshotSenderFactory((rqId, nodeId) -> new DelegateSnapshotSender(log,
- mgr.snapshotExecutorService(), old.apply(rqId, nodeId)) {
+ mgr.localSnapshotSenderFactory(sft -> new DelegateSnapshotSender(log,
+ mgr.snapshotExecutorService(), old.apply(sft)) {
@Override public void sendPart0(File part, String cacheDirName, GroupPartitionId pair, Long length) {
if (cacheDir.equals(cacheDirName))
partStart.countDown();
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotSelfTest.java
index b0fdfa76f90..82ea14e7fcc 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotSelfTest.java
@@ -35,8 +35,8 @@ import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.BiFunction;
import java.util.function.Consumer;
+import java.util.function.Function;
import java.util.function.Predicate;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
@@ -1215,12 +1215,12 @@ public class IgniteClusterSnapshotSelfTest extends AbstractSnapshotSelfTest {
IgniteEx clnt = startClientGrid(2);
IgniteSnapshotManager mgr = snp(grid);
- BiFunction<String, String, SnapshotSender> old = mgr.localSnapshotSenderFactory();
+ Function<SnapshotFileTree, SnapshotSender> old = mgr.localSnapshotSenderFactory();
BlockingExecutor block = new BlockingExecutor(mgr.snapshotExecutorService());
- mgr.localSnapshotSenderFactory((snpName, snpPath) ->
- new DelegateSnapshotSender(log, block, old.apply(snpName, snpPath)));
+ mgr.localSnapshotSenderFactory(sft ->
+ new DelegateSnapshotSender(log, block, old.apply(sft)));
IgniteFuture<Void> fut = snp(grid).createSnapshot(SNAPSHOT_NAME, null, false, onlyPrimary);
@@ -1313,13 +1313,13 @@ public class IgniteClusterSnapshotSelfTest extends AbstractSnapshotSelfTest {
* @param blocked Latch to await delta partition processing.
* @return Factory which produces local snapshot senders.
*/
- private BiFunction<String, String, SnapshotSender> blockingLocalSnapshotSender(IgniteEx ignite,
+ private Function<SnapshotFileTree, SnapshotSender> blockingLocalSnapshotSender(IgniteEx ignite,
CountDownLatch started,
CountDownLatch blocked
) {
- BiFunction<String, String, SnapshotSender> old = snp(ignite).localSnapshotSenderFactory();
+ Function<SnapshotFileTree, SnapshotSender> old = snp(ignite).localSnapshotSenderFactory();
- return (snpName, snpPath) -> new DelegateSnapshotSender(log, snp(ignite).snapshotExecutorService(), old.apply(snpName, null)) {
+ return sft -> new DelegateSnapshotSender(log, snp(ignite).snapshotExecutorService(), old.apply(sft)) {
@Override public void sendDelta0(File delta, String cacheDirName, GroupPartitionId pair) {
if (log.isInfoEnabled())
log.info("Processing delta file has been blocked: " + delta.getName());
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
index 0c56acaa691..cba652e86ec 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
@@ -144,9 +144,11 @@ public class IgniteSnapshotManagerSelfTest extends AbstractSnapshotSelfTest {
}
}, 5, "cache-loader-");
+ SnapshotFileTree sft = snapshotFileTree(ig, SNAPSHOT_NAME);
+
// Register task but not schedule it on the checkpoint.
SnapshotFutureTask snpFutTask = (SnapshotFutureTask)mgr.registerSnapshotTask(
- snapshotFileTree(ig, SNAPSHOT_NAME),
+ sft,
cctx.localNodeId(),
null,
F.asMap(CU.cacheId(DEFAULT_CACHE_NAME), null),
@@ -154,7 +156,7 @@ public class IgniteSnapshotManagerSelfTest extends AbstractSnapshotSelfTest {
false,
false,
false,
- new DelegateSnapshotSender(log, mgr.snapshotExecutorService(), mgr.localSnapshotSenderFactory().apply(SNAPSHOT_NAME, null)) {
+ new DelegateSnapshotSender(log, mgr.snapshotExecutorService(), mgr.localSnapshotSenderFactory().apply(sft)) {
@Override public void sendPart0(File part, String cacheDirName, GroupPartitionId pair, Long length) {
try {
U.await(slowCopy);
@@ -272,11 +274,13 @@ public class IgniteSnapshotManagerSelfTest extends AbstractSnapshotSelfTest {
}
});
+ SnapshotFileTree sft = snapshotFileTree(ig, SNAPSHOT_NAME);
+
IgniteInternalFuture<?> snpFut = startLocalSnapshotTask(cctx0,
- snapshotFileTree(ig, SNAPSHOT_NAME),
+ sft,
F.asMap(CU.cacheId(DEFAULT_CACHE_NAME), null),
encryption,
- mgr.localSnapshotSenderFactory().apply(SNAPSHOT_NAME, null));
+ mgr.localSnapshotSenderFactory().apply(sft));
// Check the right exception thrown.
assertThrowsAnyCause(log,
@@ -296,12 +300,14 @@ public class IgniteSnapshotManagerSelfTest extends AbstractSnapshotSelfTest {
IgniteSnapshotManager mgr0 = snp(ig);
+ SnapshotFileTree sft = snapshotFileTree(ig, SNAPSHOT_NAME);
+
IgniteInternalFuture<?> fut = startLocalSnapshotTask(ig.context().cache().context(),
- snapshotFileTree(ig, SNAPSHOT_NAME),
+ sft,
parts,
encryption,
new DelegateSnapshotSender(log, mgr0.snapshotExecutorService(),
- mgr0.localSnapshotSenderFactory().apply(SNAPSHOT_NAME, null)) {
+ mgr0.localSnapshotSenderFactory().apply(sft)) {
@Override public void sendPart0(File part, String cacheDirName, GroupPartitionId pair, Long length) {
if (pair.getPartitionId() == 0)
throw new IgniteException(err_msg + pair);
@@ -332,11 +338,13 @@ public class IgniteSnapshotManagerSelfTest extends AbstractSnapshotSelfTest {
CountDownLatch cpLatch = new CountDownLatch(1);
+ SnapshotFileTree sft = snapshotFileTree(ig, SNAPSHOT_NAME);
+
IgniteInternalFuture<?> snpFut = startLocalSnapshotTask(cctx0,
- snapshotFileTree(ig, SNAPSHOT_NAME),
+ sft,
F.asMap(CU.cacheId(DEFAULT_CACHE_NAME), null),
encryption,
- new DelegateSnapshotSender(log, mgr.snapshotExecutorService(), mgr.localSnapshotSenderFactory().apply(SNAPSHOT_NAME, null)) {
+ new DelegateSnapshotSender(log, mgr.snapshotExecutorService(), mgr.localSnapshotSenderFactory().apply(sft)) {
@Override public void sendPart0(File part, String cacheDirName, GroupPartitionId pair, Long length) {
try {
U.await(cpLatch);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/PlainSnapshotTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/PlainSnapshotTest.java
index b8d283b85cc..10439d75651 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/PlainSnapshotTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/PlainSnapshotTest.java
@@ -88,18 +88,18 @@ public class PlainSnapshotTest extends AbstractSnapshotSelfTest {
GridCacheSharedContext<?, ?> cctx = ig.context().cache().context();
IgniteSnapshotManager mgr = snp(ig);
+ SnapshotFileTree sft = snapshotFileTree(ig, SNAPSHOT_NAME);
// Collection of pairs group and appropriate cache partition to be snapshot.
IgniteInternalFuture<?> snpFut = startLocalSnapshotTask(cctx,
- snapshotFileTree(ig, SNAPSHOT_NAME),
+ sft,
F.asMap(CU.cacheId(DEFAULT_CACHE_NAME), null),
- false, mgr.localSnapshotSenderFactory().apply(SNAPSHOT_NAME, null));
+ false, mgr.localSnapshotSenderFactory().apply(sft));
snpFut.get();
// Calculate CRCs.
NodeFileTree ft = ig.context().pdsFolderResolver().fileTree();
- SnapshotFileTree sft = snapshotFileTree(ig, SNAPSHOT_NAME);
// Checkpoint forces on cluster deactivation (currently only single node in cluster),
// so we must have the same data in snapshot partitions and those which left