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 2023/09/22 07:38:43 UTC
[ignite] branch cache_dumps updated: IGNITE-19950 Creation of cache dump implemented (#10936)
This is an automated email from the ASF dual-hosted git repository.
nizhikov pushed a commit to branch cache_dumps
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/cache_dumps by this push:
new bd81eda1888 IGNITE-19950 Creation of cache dump implemented (#10936)
bd81eda1888 is described below
commit bd81eda18882ecde7c2b0d31a53ba2d889f82192
Author: Nikolay <ni...@apache.org>
AuthorDate: Fri Sep 22 10:38:34 2023 +0300
IGNITE-19950 Creation of cache dump implemented (#10936)
---
.../ignite/jdbc/thin/JdbcThinMetadataSelfTest.java | 8 +
.../java/org/apache/ignite/IgniteSnapshot.java | 12 +
.../processors/cache/GridCacheContext.java | 17 +
.../processors/cache/GridCacheMapEntry.java | 40 ++
.../processors/cache/GridLocalConfigManager.java | 8 +-
.../cache/binary/BinaryMetadataFileStore.java | 13 +-
.../binary/CacheObjectBinaryProcessorImpl.java | 11 +-
.../persistence/file/FilePageStoreManager.java | 11 +-
.../snapshot/AbstractCreateSnapshotFutureTask.java | 250 ++++++++++
.../snapshot/AbstractSnapshotFutureTask.java | 16 +-
.../snapshot/IgniteSnapshotManager.java | 161 ++++--
.../snapshot/IncrementalSnapshotFutureTask.java | 5 -
.../snapshot/SnapshotFinishedFutureTask.java | 2 +-
.../persistence/snapshot/SnapshotFutureTask.java | 281 +++--------
.../snapshot/SnapshotFutureTaskResult.java | 4 +-
.../persistence/snapshot/SnapshotMetadata.java | 13 +-
.../snapshot/SnapshotOperationRequest.java | 13 +-
.../snapshot/SnapshotPartitionsVerifyHandler.java | 21 +-
.../snapshot/SnapshotResponseRemoteFutureTask.java | 7 +-
.../snapshot/SnapshotRestoreProcess.java | 8 +-
.../cache/persistence/snapshot/SnapshotSender.java | 2 +-
.../snapshot/dump/CreateDumpFutureTask.java | 544 +++++++++++++++++++++
.../cache/persistence/snapshot/dump/Dump.java | 263 ++++++++++
.../DumpEntry.java} | 34 +-
.../DumpEntryChangeListener.java} | 36 +-
.../snapshot/dump/DumpEntrySerializer.java | 226 +++++++++
.../internal/processors/pool/PoolProcessor.java | 22 +-
.../ignite/spi/systemview/view/SnapshotView.java | 7 +-
.../snapshot/AbstractSnapshotSelfTest.java | 2 +-
.../snapshot/EncryptedSnapshotTest.java | 2 +-
.../snapshot/IgniteSnapshotManagerSelfTest.java | 1 +
.../snapshot/dump/AbstractCacheDumpTest.java | 434 ++++++++++++++++
.../snapshot/dump/IgniteCacheDumpSelf2Test.java | 212 ++++++++
.../snapshot/dump/IgniteCacheDumpSelfTest.java | 503 +++++++++++++++++++
.../java/org/apache/ignite/platform/model/Key.java | 5 +
.../org/apache/ignite/platform/model/Role.java | 15 +
.../org/apache/ignite/platform/model/User.java | 15 +
.../ignite/testsuites/IgniteSnapshotTestSuite.java | 4 +
.../org/apache/ignite/internal/cdc/SqlCdcTest.java | 2 +-
.../ignite/internal/dump/DumpCacheConfigTest.java | 125 +++++
.../IgniteBinaryCacheQueryTestSuite3.java | 5 +-
41 files changed, 3004 insertions(+), 356 deletions(-)
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java
index 22b07cf579a..2a7547d59fb 100644
--- a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java
@@ -419,6 +419,7 @@ public class JdbcThinMetadataSelfTest extends JdbcThinAbstractSelfTest {
new TreeSet<>(Arrays.asList(
"SYS.METRICS",
"SYS.SERVICES",
+ "SYS.SNAPSHOT",
"SYS.CACHE_GROUPS",
"SYS.CACHES",
"SYS.TASKS",
@@ -896,6 +897,13 @@ public class JdbcThinMetadataSelfTest extends JdbcThinAbstractSelfTest {
"SYS.SERVICES.STATICALLY_CONFIGURED.null",
"SYS.SERVICES.SERVICE_ID.null",
"SYS.SERVICES.TOPOLOGY_SNAPSHOT.null",
+ "SYS.SNAPSHOT.BASELINE_NODES.null",
+ "SYS.SNAPSHOT.CACHE_GROUPS.null",
+ "SYS.SNAPSHOT.CONSISTENT_ID.null",
+ "SYS.SNAPSHOT.INCREMENT_INDEX.null",
+ "SYS.SNAPSHOT.NAME.null",
+ "SYS.SNAPSHOT.SNAPSHOT_RECORD_SEGMENT.null",
+ "SYS.SNAPSHOT.TYPE.null",
"SYS.TASKS.AFFINITY_CACHE_NAME.null",
"SYS.TASKS.INTERNAL.null",
"SYS.TASKS.END_TIME.null",
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteSnapshot.java b/modules/core/src/main/java/org/apache/ignite/IgniteSnapshot.java
index 2f6cc9f4988..e66d742ba17 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSnapshot.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSnapshot.java
@@ -34,6 +34,8 @@ import org.jetbrains.annotations.Nullable;
* <li>Concurrent updates from {@link IgniteDataStreamer} with default {@link IgniteDataStreamer#allowOverwrite()}
* setting (false) into a persistent cache can cause that cache data stored inconsistent.</li>
* </ul>
+ * You can create cache dump, also. Cache dump is consistent files consist of entry by entry cache content.
+ * This type of snapshots supports in-memory caches.
*/
public interface IgniteSnapshot {
/**
@@ -61,6 +63,16 @@ public interface IgniteSnapshot {
*/
public IgniteFuture<Void> cancelSnapshot(String name);
+ /**
+ * Creates cache groups dump.
+ * Dump is consistent entry by entry backup of cache group content.
+ * In-memory cache groups also supported.
+ *
+ * @param name Dump name.
+ * @return Future which will be completed when dump ends.
+ */
+ public IgniteFuture<Void> createDump(String name);
+
/**
* Restore cache group(s) from the snapshot.
* <p>
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
index 6136c24ce72..b08621a310b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java
@@ -79,6 +79,7 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTran
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrManager;
import org.apache.ignite.internal.processors.cache.jta.CacheJtaManagerAdapter;
import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.DumpEntryChangeListener;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryManager;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryManager;
import org.apache.ignite.internal.processors.cache.store.CacheStoreManager;
@@ -284,6 +285,9 @@ public class GridCacheContext<K, V> implements Externalizable {
/** Recovery mode flag. */
private volatile boolean recoveryMode;
+ /** Dump callback. */
+ private volatile DumpEntryChangeListener dumpLsnr;
+
/** */
private final boolean disableTriggeringCacheInterceptorOnConflict =
Boolean.parseBoolean(System.getProperty(IGNITE_DISABLE_TRIGGERING_CACHE_INTERCEPTOR_ON_CONFLICT, "false"));
@@ -2341,6 +2345,19 @@ public class GridCacheContext<K, V> implements Externalizable {
return lastRmvAllJobFut;
}
+ /** */
+ public DumpEntryChangeListener dumpListener() {
+ return dumpLsnr;
+ }
+
+ /** */
+ public void dumpListener(DumpEntryChangeListener dumpEntryChangeLsnr) {
+ assert this.dumpLsnr == null || dumpEntryChangeLsnr == null;
+ assert cacheType == CacheType.USER;
+
+ this.dumpLsnr = dumpEntryChangeLsnr;
+ }
+
/** {@inheritDoc} */
@Override public void writeExternal(ObjectOutput out) throws IOException {
U.writeString(out, igniteInstanceName());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index cf7815a465b..85708085efe 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -66,6 +66,7 @@ import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRowAdapter;
import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
import org.apache.ignite.internal.processors.cache.persistence.StorageException;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.DumpEntryChangeListener;
import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryListener;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
@@ -1466,6 +1467,11 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
assert ttl >= 0 : ttl;
assert expireTime >= 0 : expireTime;
+ DumpEntryChangeListener dumpLsnr = cctx.dumpListener();
+
+ if (dumpLsnr != null)
+ dumpLsnr.beforeChange(cctx, key, old, extras == null ? CU.EXPIRE_TIME_ETERNAL : extras.expireTime(), ver);
+
// Detach value before index update.
val = cctx.kernalContext().cacheObjects().prepareForCache(val, cctx);
@@ -1659,6 +1665,11 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
+ DumpEntryChangeListener dumpLsnr = cctx.dumpListener();
+
+ if (dumpLsnr != null)
+ dumpLsnr.beforeChange(cctx, key, old, extras == null ? CU.EXPIRE_TIME_ETERNAL : extras.expireTime(), ver);
+
removeValue();
update(null, 0, 0, newVer, true);
@@ -3602,6 +3613,11 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (mvccExtras() != null)
return false;
+ DumpEntryChangeListener dumpLsnr = cctx.dumpListener();
+
+ if (dumpLsnr != null)
+ dumpLsnr.beforeChange(cctx, key, expiredVal, extras == null ? CU.TTL_MINIMUM : extras.expireTime(), ver);
+
if (cctx.deferredDelete() && !detached() && !isInternal()) {
if (!deletedUnlocked() && !isStartVersion()) {
update(null, 0L, 0L, ver, true);
@@ -6008,6 +6024,18 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
+ DumpEntryChangeListener dumpLsnr = cctx.dumpListener();
+
+ if (dumpLsnr != null) {
+ dumpLsnr.beforeChange(
+ cctx,
+ entry.key,
+ oldVal,
+ entry.extras == null ? CU.EXPIRE_TIME_ETERNAL : entry.extras.expireTime(),
+ entry.ver
+ );
+ }
+
updated = cctx.kernalContext().cacheObjects().prepareForCache(updated, cctx);
if (writeThrough)
@@ -6111,6 +6139,18 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
+ DumpEntryChangeListener dumpLsnr = cctx.dumpListener();
+
+ if (dumpLsnr != null) {
+ dumpLsnr.beforeChange(
+ cctx,
+ entry.key,
+ oldVal,
+ entry.extras == null ? CU.EXPIRE_TIME_ETERNAL : entry.extras.expireTime(),
+ entry.ver
+ );
+ }
+
if (writeThrough)
// Must persist inside synchronization in non-tx mode.
cctx.store().remove(null, entry.key);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridLocalConfigManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridLocalConfigManager.java
index 846d7128a99..67cb6b02117 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridLocalConfigManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridLocalConfigManager.java
@@ -541,8 +541,12 @@ public class GridLocalConfigManager {
public File cacheConfigurationFile(CacheConfiguration<?, ?> ccfg) {
File cacheWorkDir = cacheWorkDir(ccfg);
- return ccfg.getGroupName() == null ? new File(cacheWorkDir, CACHE_DATA_FILENAME) :
- new File(cacheWorkDir, ccfg.getName() + CACHE_DATA_FILENAME);
+ return new File(cacheWorkDir, cacheDataFilename(ccfg));
+ }
+
+ /** @return Name of cache data filename. */
+ public static String cacheDataFilename(CacheConfiguration<?, ?> ccfg) {
+ return ccfg.getGroupName() == null ? CACHE_DATA_FILENAME : (ccfg.getName() + CACHE_DATA_FILENAME);
}
/**
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 b994c4b7d06..db43a7a0c11 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
@@ -82,35 +82,40 @@ class BinaryMetadataFileStore {
* @param log Logger.
* @param binaryMetadataFileStoreDir Path to binary metadata store configured by user, should include binary_meta
* and consistentId.
+ * @param forceEnabled If {@code true} then will write files even if persistence and CDC disabled.
*/
BinaryMetadataFileStore(
final ConcurrentMap<Integer, BinaryMetadataHolder> metadataLocCache,
final GridKernalContext ctx,
final IgniteLogger log,
- final File binaryMetadataFileStoreDir
+ final File binaryMetadataFileStoreDir,
+ final boolean forceEnabled
) throws IgniteCheckedException {
this.metadataLocCache = metadataLocCache;
this.ctx = ctx;
- enabled = CU.isPersistenceEnabled(ctx.config()) || CU.isCdcEnabled(ctx.config());
+ enabled = forceEnabled || CU.isPersistenceEnabled(ctx.config()) || CU.isCdcEnabled(ctx.config());
this.log = log;
if (!enabled)
return;
- fileIOFactory = ctx.config().getDataStorageConfiguration().getFileIOFactory();
+ DataStorageConfiguration dsCfg = ctx.config().getDataStorageConfiguration();
+
+ fileIOFactory = dsCfg == null ? new DataStorageConfiguration().getFileIOFactory() : dsCfg.getFileIOFactory();
final String nodeFolderName = ctx.pdsFolderResolver().resolveFolders().folderName();
if (binaryMetadataFileStoreDir != null)
metadataDir = binaryMetadataFileStoreDir;
- else
+ else {
metadataDir = new File(U.resolveWorkDirectory(
ctx.config().getWorkDirectory(),
DataStorageConfiguration.DFLT_BINARY_METADATA_PATH,
false
), nodeFolderName);
+ }
fixLegacyFolder(nodeFolderName);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
index db8bb33c8b4..7ca52e66449 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java
@@ -261,7 +261,8 @@ public class CacheObjectBinaryProcessorImpl extends GridProcessorAdapter impleme
CU.isPersistenceEnabled(ctx.config()) && binaryMetadataFileStoreDir == null ?
resolveBinaryWorkDir(ctx.config().getWorkDirectory(),
ctx.pdsFolderResolver().resolveFolders().folderName()) :
- binaryMetadataFileStoreDir);
+ binaryMetadataFileStoreDir,
+ false);
metadataFileStore.start();
}
@@ -1019,7 +1020,9 @@ public class CacheObjectBinaryProcessorImpl extends GridProcessorAdapter impleme
ctx,
log,
resolveBinaryWorkDir(dir.getAbsolutePath(),
- ctx.pdsFolderResolver().resolveFolders().folderName()));
+ ctx.pdsFolderResolver().resolveFolders().folderName()),
+ true
+ );
for (BinaryType type : types)
writer.mergeAndWriteMetadata(((BinaryTypeImpl)type).metadata());
@@ -1037,7 +1040,7 @@ public class CacheObjectBinaryProcessorImpl extends GridProcessorAdapter impleme
try {
ConcurrentMap<Integer, BinaryMetadataHolder> metaCache = new ConcurrentHashMap<>();
- new BinaryMetadataFileStore(metaCache, ctx, log, metadataDir)
+ new BinaryMetadataFileStore(metaCache, ctx, log, metadataDir, false)
.restoreMetadata();
Collection<BinaryMetadata> metadata = F.viewReadOnly(metaCache.values(), BinaryMetadataHolder::metadata);
@@ -1073,7 +1076,7 @@ public class CacheObjectBinaryProcessorImpl extends GridProcessorAdapter impleme
ConcurrentMap<Integer, BinaryMetadataHolder> metaCache = new ConcurrentHashMap<>();
- new BinaryMetadataFileStore(metaCache, ctx, log, metadataDir).restoreMetadata(typeId);
+ new BinaryMetadataFileStore(metaCache, ctx, log, metadataDir, false).restoreMetadata(typeId);
addMetaLocally(typeId, metaCache.get(typeId).metadata().wrap(binaryContext()), false);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
index af0a4975d1e..ee2d6d87ea0 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
@@ -916,6 +916,15 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
* @return List of cache partitions in given directory.
*/
public static List<File> cachePartitionFiles(File cacheDir) {
+ return cachePartitionFiles(cacheDir, FILE_SUFFIX);
+ }
+
+ /**
+ * @param cacheDir Cache directory to check.
+ * @param ext File extension.
+ * @return List of cache partitions in given directory.
+ */
+ public static List<File> cachePartitionFiles(File cacheDir, String ext) {
File[] files = cacheDir.listFiles();
if (files == null)
@@ -923,7 +932,7 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
return Arrays.stream(files)
.filter(File::isFile)
- .filter(f -> f.getName().endsWith(FILE_SUFFIX))
+ .filter(f -> f.getName().endsWith(ext))
.collect(Collectors.toList());
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractCreateSnapshotFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractCreateSnapshotFutureTask.java
new file mode 100644
index 00000000000..89ae911cf5d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractCreateSnapshotFutureTask.java
@@ -0,0 +1,250 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.snapshot;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.binary.BinaryType;
+import org.apache.ignite.internal.IgniteFutureCancelledCheckedException;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointListener;
+import org.apache.ignite.internal.processors.marshaller.MappedName;
+import org.apache.ignite.internal.util.lang.IgniteThrowableRunner;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+
+import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION;
+
+/**
+ *
+ */
+public abstract class AbstractCreateSnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskResult> {
+ /**
+ * Cache group and corresponding partitions collected under the PME lock.
+ * For full snapshot additional checkpoint write lock required.
+ * @see SnapshotFutureTask#onMarkCheckpointBegin(CheckpointListener.Context)
+ */
+ protected final Map<Integer, Set<Integer>> processed = new HashMap<>();
+
+ /** Future which will be completed when task requested to be closed. Will be executed on system pool. */
+ protected volatile CompletableFuture<Void> closeFut;
+
+ /**
+ * @param cctx Shared context.
+ * @param srcNodeId Node id which cause snapshot task creation.
+ * @param reqId Snapshot operation request ID.
+ * @param snpName Snapshot name.
+ * @param snpSndr Factory which produces snapshot sender instance.
+ * @param parts Partitions to be processed.
+ */
+ protected AbstractCreateSnapshotFutureTask(
+ GridCacheSharedContext<?, ?> cctx,
+ UUID srcNodeId,
+ UUID reqId,
+ String snpName,
+ SnapshotSender snpSndr,
+ Map<Integer, Set<Integer>> parts
+ ) {
+ super(cctx, srcNodeId, reqId, snpName, snpSndr, parts);
+ }
+
+ /** */
+ protected abstract List<CompletableFuture<Void>> saveCacheConfigs();
+
+ /** */
+ protected abstract List<CompletableFuture<Void>> saveGroup(int grpId, Set<Integer> grpParts) throws IgniteCheckedException;
+
+ /** {@inheritDoc} */
+ @Override public boolean cancel() {
+ super.cancel();
+
+ try {
+ closeAsync().get();
+ }
+ catch (InterruptedException | ExecutionException e) {
+ U.error(log, "SnapshotFutureTask cancellation failed", e);
+
+ return false;
+ }
+
+ return true;
+ }
+
+ /** @return Future which will be completed when operations truly stopped. */
+ protected abstract CompletableFuture<Void> closeAsync();
+
+ /**
+ * @return {@code true} if current task requested to be stopped.
+ */
+ protected boolean stopping() {
+ return err.get() != null;
+ }
+
+ /** */
+ protected void processPartitions() throws IgniteCheckedException {
+ for (Map.Entry<Integer, Set<Integer>> e : parts.entrySet()) {
+ int grpId = e.getKey();
+ Set<Integer> grpParts = e.getValue();
+
+ CacheGroupContext gctx = cctx.cache().cacheGroup(grpId);
+
+ Iterator<GridDhtLocalPartition> iter;
+
+ if (grpParts == null)
+ iter = gctx.topology().currentLocalPartitions().iterator();
+ else {
+ if (grpParts.contains(INDEX_PARTITION)) {
+ throw new IgniteCheckedException("Index partition cannot be included into snapshot if " +
+ " set of cache group partitions has been explicitly provided [grpId=" + grpId + ']');
+ }
+
+ iter = F.iterator(grpParts, gctx.topology()::localPartition, false);
+ }
+
+ Set<Integer> owning = new HashSet<>();
+ Set<Integer> missed = new HashSet<>();
+
+ // Iterate over partitions in particular cache group.
+ while (iter.hasNext()) {
+ GridDhtLocalPartition part = iter.next();
+
+ // Partition can be in MOVING\RENTING states.
+ // Index partition will be excluded if not all partition OWNING.
+ // There is no data assigned to partition, thus it haven't been created yet.
+ if (part.state() == GridDhtPartitionState.OWNING)
+ owning.add(part.id());
+ else
+ missed.add(part.id());
+ }
+
+ boolean affNode = gctx.nodeFilter() == null || gctx.nodeFilter().apply(cctx.localNode());
+
+ if (grpParts != null) {
+ // Partition has been provided for cache group, but some of them are not in OWNING state.
+ // Exit with an error.
+ if (!missed.isEmpty()) {
+ throw new IgniteCheckedException("Snapshot operation cancelled due to " +
+ "not all of requested partitions has OWNING state on local node [grpId=" + grpId +
+ ", missed=" + S.toStringSortedDistinct(missed) + ']');
+ }
+ }
+ else {
+ // Partitions have not been provided for snapshot task and all partitions have
+ // OWNING state, so index partition must be included into snapshot.
+ if (!missed.isEmpty()) {
+ log.warning("All local cache group partitions in OWNING state have been included into a snapshot. " +
+ "Partitions which have different states skipped. Index partitions has also been skipped " +
+ "[snpName=" + snpName + ", grpId=" + grpId + ", missed=" + S.toStringSortedDistinct(missed) + ']');
+ }
+ else if (affNode && missed.isEmpty() && cctx.kernalContext().query().moduleEnabled())
+ owning.add(INDEX_PARTITION);
+ }
+
+ processed.put(grpId, owning);
+ }
+ }
+
+ /** Starts async execution of all tasks required to create snapshot. */
+ protected void saveSnapshotData() {
+ try {
+ // Submit all tasks for partitions and deltas processing.
+ List<CompletableFuture<Void>> futs = new ArrayList<>();
+
+ Collection<BinaryType> binTypesCopy = cctx.kernalContext()
+ .cacheObjects()
+ .metadata();
+
+ List<Map<Integer, MappedName>> mappingsCopy = cctx.kernalContext()
+ .marshallerContext()
+ .getCachedMappings();
+
+ // Process binary meta.
+ futs.add(runAsync(() -> snpSndr.sendBinaryMeta(binTypesCopy)));
+ // Process marshaller meta.
+ futs.add(runAsync(() -> snpSndr.sendMarshallerMeta(mappingsCopy)));
+ futs.addAll(saveCacheConfigs());
+
+ for (Map.Entry<Integer, Set<Integer>> grpParts : processed.entrySet())
+ futs.addAll(saveGroup(grpParts.getKey(), grpParts.getValue()));
+
+ int futsSize = futs.size();
+
+ CompletableFuture.allOf(futs.toArray(new CompletableFuture[futsSize])).whenComplete((res, t) -> {
+ assert t == null : "Exception must never be thrown since a wrapper is used " +
+ "for each snapshot task: " + t;
+
+ closeAsync();
+ });
+ }
+ catch (IgniteCheckedException e) {
+ acceptException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void acceptException(Throwable th) {
+ if (th == null)
+ return;
+
+ if (!(th instanceof IgniteFutureCancelledCheckedException))
+ U.error(log, "Snapshot task has accepted exception to stop", th);
+
+ if (err.compareAndSet(null, th))
+ closeAsync();
+ }
+
+ /**
+ * @param exec Runnable task to execute.
+ * @return Wrapped task.
+ */
+ Runnable wrapExceptionIfStarted(IgniteThrowableRunner exec) {
+ return () -> {
+ if (stopping())
+ return;
+
+ try {
+ exec.run();
+ }
+ catch (Throwable t) {
+ acceptException(t);
+ }
+ };
+ }
+
+ /** */
+ protected CompletableFuture<Void> runAsync(IgniteThrowableRunner task) {
+ return CompletableFuture.runAsync(
+ wrapExceptionIfStarted(task),
+ snpSndr.executor()
+ );
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotFutureTask.java
index 7ccc2f3944c..12d0f755a6a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotFutureTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotFutureTask.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.processors.cache.persistence.snapshot;
-import java.io.File;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -25,7 +24,6 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.IgniteFutureCancelledCheckedException;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
-import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -49,12 +47,6 @@ abstract class AbstractSnapshotFutureTask<T> extends GridFutureAdapter<T> {
/** Unique identifier of snapshot process. */
protected final String snpName;
- /** Snapshot working directory on file system. */
- protected final File tmpSnpWorkDir;
-
- /** IO factory which will be used for creating snapshot delta-writers. */
- protected final FileIOFactory ioFactory;
-
/** Snapshot data sender. */
@GridToStringExclude
protected final SnapshotSender snpSndr;
@@ -70,8 +62,6 @@ abstract class AbstractSnapshotFutureTask<T> extends GridFutureAdapter<T> {
* @param srcNodeId Node id which cause snapshot task creation.
* @param reqId Snapshot operation request ID.
* @param snpName Unique identifier of snapshot process.
- * @param tmpWorkDir Working directory for intermediate snapshot results.
- * @param ioFactory Factory to working with snapshot files.
* @param snpSndr Factory which produces snapshot receiver instance.
* @param parts Partition to be processed.
*/
@@ -80,8 +70,6 @@ abstract class AbstractSnapshotFutureTask<T> extends GridFutureAdapter<T> {
UUID srcNodeId,
UUID reqId,
String snpName,
- File tmpWorkDir,
- FileIOFactory ioFactory,
SnapshotSender snpSndr,
Map<Integer, Set<Integer>> parts
) {
@@ -90,12 +78,10 @@ abstract class AbstractSnapshotFutureTask<T> extends GridFutureAdapter<T> {
assert snpSndr.executor() != null : "Executor service must be not null.";
this.cctx = cctx;
- this.log = cctx.logger(AbstractSnapshotFutureTask.class);
+ this.log = cctx.logger(this.getClass());
this.srcNodeId = srcNodeId;
this.reqId = reqId;
this.snpName = snpName;
- this.tmpSnpWorkDir = new File(tmpWorkDir, snpName);
- this.ioFactory = ioFactory;
this.snpSndr = snpSndr;
this.parts = parts;
}
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 157a1ea5c57..4bbb285757e 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
@@ -141,6 +141,7 @@ import org.apache.ignite.internal.processors.cache.persistence.metastorage.Metas
import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadOnlyMetastorage;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadWriteMetastorage;
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.CreateDumpFutureTask;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPageIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.DataPagePayload;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
@@ -253,7 +254,8 @@ import static org.apache.ignite.spi.systemview.view.SnapshotView.SNAPSHOT_SYS_VI
* <p>
* These major actions available:
* <ul>
- * <li>Create snapshot of the whole cluster cache groups by triggering PME to achieve consistency.</li>
+ * <li>Create snapshot of the whole persistent cluster cache groups by triggering PME to achieve consistency.</li>
+ * <li>Create cache dump - snapshot of cluster cache groups including in-memory.</li>
* <li>Create incremental snapshot using lightweight, non-blocking Consistent Cut algorithm.</li>
* </ul>
*/
@@ -311,7 +313,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
public static final int SNAPSHOT_LIMITED_TRANSFER_BLOCK_SIZE_BYTES = 64 * 1024;
/** Metastorage key to save currently running snapshot directory path. */
- private static final String SNP_RUNNING_DIR_KEY = "snapshot-running-dir";
+ public static final String SNP_RUNNING_DIR_KEY = "snapshot-running-dir";
/** Prefix for meta store records which means that incremental snapshot creation is disabled for a cache group. */
private static final String INC_SNP_DISABLED_KEY_PREFIX = "grp-inc-snp-disabled-";
@@ -359,6 +361,9 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/** Pattern for incremental snapshot directory names. */
public static final Pattern INC_SNP_NAME_PATTERN = U.fixedLengthNumberNamePattern(null);
+ /** Lock file for dump directory. */
+ public static final String DUMP_LOCK = "dump.lock";
+
/**
* Local buffer to perform copy-on-write operations with pages for {@code SnapshotFutureTask.PageStoreSerialWriter}s.
* It is important to have only one buffer per thread (instead of creating each buffer per
@@ -416,7 +421,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
private volatile FileIOFactory ioFactory = new RandomAccessFileIOFactory();
/** File store manager to create page store for restore. */
- private volatile FilePageStoreManager storeMgr;
+ private volatile @Nullable FilePageStoreManager storeMgr;
/** File store manager to create page store for restore. */
private volatile GridLocalConfigManager locCfgMgr;
@@ -525,18 +530,13 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
if (ctx.clientNode())
return;
- if (!CU.isPersistenceEnabled(ctx.config()))
- return;
-
- assert cctx.pageStore() instanceof FilePageStoreManager;
-
storeMgr = (FilePageStoreManager)cctx.pageStore();
locCfgMgr = cctx.cache().configManager();
pdsSettings = cctx.kernalContext().pdsFolderResolver().resolveFolders();
locSnpDir = resolveSnapshotWorkDirectory(ctx.config());
- tmpWorkDir = U.resolveWorkDirectory(storeMgr.workDir().getAbsolutePath(), DFLT_SNAPSHOT_TMP_DIR, true);
+ tmpWorkDir = U.resolveWorkDirectory(pdsSettings.persistentStoreNodePath().getAbsolutePath(), DFLT_SNAPSHOT_TMP_DIR, true);
U.ensureDirectory(locSnpDir, "snapshot work directory", log);
U.ensureDirectory(tmpWorkDir, "temp directory for snapshot creation", log);
@@ -690,6 +690,20 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
return views;
})),
Function.identity());
+
+ Arrays.stream(locSnpDir.listFiles())
+ .filter(File::isDirectory)
+ .map(dumpDir ->
+ Paths.get(dumpDir.getAbsolutePath(), DB_DEFAULT_FOLDER, pdsSettings.folderName(), DUMP_LOCK).toFile())
+ .filter(File::exists)
+ .map(File::getParentFile)
+ .forEach(lockedDumpDir -> {
+ log.warning("Found locked dump dir. " +
+ "This means, dump creation not finished prior to node fail. " +
+ "Directory will be deleted: " + lockedDumpDir);
+
+ U.delete(lockedDumpDir);
+ });
}
/** {@inheritDoc} */
@@ -740,19 +754,21 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/**
* @param snpDir Snapshot dir.
- * @param folderName Local node folder name (see {@link U#maskForFileName} with consistent id).
+ * @param pdsSettings PDS settings.
*/
- public void deleteSnapshot(File snpDir, String folderName) {
+ public void deleteSnapshot(File snpDir, PdsFolderSettings<?> pdsSettings) {
if (!snpDir.exists())
return;
if (!snpDir.isDirectory())
return;
+ String folderName = pdsSettings.folderName();
+
try {
File binDir = binaryWorkDir(snpDir.getAbsolutePath(), folderName);
File nodeDbDir = new File(snpDir.getAbsolutePath(), databaseRelativePath(folderName));
- File smf = new File(snpDir, snapshotMetaFileName(folderName));
+ File smf = new File(snpDir, snapshotMetaFileName(U.maskForFileName(pdsSettings.consistentId().toString())));
U.delete(binDir);
U.delete(nodeDbDir);
@@ -1027,8 +1043,6 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
meta,
req.snapshotPath(),
req.incrementIndex(),
- tmpWorkDir,
- ioFactory,
lowPtr,
markWalFut
)).chain(fut -> {
@@ -1150,6 +1164,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
req.requestId(),
parts,
withMetaStorage,
+ req.dump(),
locSndrFactory.apply(req.snapshotName(), req.snapshotPath()));
if (withMetaStorage && task0 instanceof SnapshotFutureTask) {
@@ -1184,7 +1199,8 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
res.parts(),
res.snapshotPointer(),
cctx.gridConfig().getEncryptionSpi().masterKeyDigest(),
- req.onlyPrimary()
+ req.onlyPrimary(),
+ req.dump()
);
SnapshotHandlerContext ctx = new SnapshotHandlerContext(meta, req.groups(), cctx.localNode(), snpDir,
@@ -1382,7 +1398,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
if (req.incremental())
U.delete(incrementalSnapshotLocalDir(req.snapshotName(), req.snapshotPath(), req.incrementIndex()));
else
- deleteSnapshot(snapshotLocalDir(req.snapshotName(), req.snapshotPath()), pdsSettings.folderName());
+ deleteSnapshot(snapshotLocalDir(req.snapshotName(), req.snapshotPath()), pdsSettings);
}
else if (!F.isEmpty(req.warnings())) {
// Pass the warnings further to the next stage for the case when snapshot started from not coordinator.
@@ -1394,12 +1410,16 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
storeWarnings(snpReq);
}
- removeLastMetaStorageKey();
+ if (req.dump())
+ removeDumpLock(req.snapshotName());
+ else {
+ removeLastMetaStorageKey();
- if (req.error() == null) {
- Collection<Integer> grpIds = req.groups().stream().map(CU::cacheId).collect(Collectors.toList());
+ if (req.error() == null) {
+ Collection<Integer> grpIds = req.groups().stream().map(CU::cacheId).collect(Collectors.toList());
- enableIncrementalSnapshotsCreation(grpIds);
+ enableIncrementalSnapshotsCreation(grpIds);
+ }
}
}
catch (Exception e) {
@@ -1755,6 +1775,11 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
return new IgniteFutureImpl<>(cancelSnapshot0(name));
}
+ /** {@inheritDoc} */
+ @Override public IgniteFuture<Void> createDump(String name) {
+ return createSnapshot(name, null, false, false, true);
+ }
+
/**
* @param name Snapshot name.
*
@@ -2139,10 +2164,31 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
@Nullable String snpPath,
boolean incremental,
boolean onlyPrimary
+ ) {
+ return createSnapshot(name, snpPath, incremental, onlyPrimary, false);
+ }
+
+ /**
+ * Create a consistent copy of all persistence cache groups from the whole cluster.
+ *
+ * @param name Snapshot unique name which satisfies the following name pattern [a-zA-Z0-9_].
+ * @param snpPath Snapshot directory path.
+ * @param incremental Incremental snapshot flag.
+ * @param onlyPrimary If {@code true} snapshot only primary copies of partitions.
+ * @param dump If {@code true} cache dump must be created.
+ * @return Future which will be completed when a process ends.
+ */
+ public IgniteFutureImpl<Void> createSnapshot(
+ String name,
+ @Nullable String snpPath,
+ boolean incremental,
+ boolean onlyPrimary,
+ boolean dump
) {
A.notNullOrEmpty(name, "Snapshot name cannot be null or empty.");
A.ensure(U.alphanumericUnderscore(name), "Snapshot name must satisfy the following name pattern: a-zA-Z0-9_");
A.ensure(!(incremental && onlyPrimary), "Only primary not supported for incremental snapshots");
+ A.ensure(!(dump && incremental), "Incremental dump not supported");
try {
cctx.kernalContext().security().authorize(ADMIN_SNAPSHOT);
@@ -2167,12 +2213,12 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
return new IgniteSnapshotFutureImpl(cctx.kernalContext().closure()
.callAsync(
BALANCE,
- new CreateSnapshotCallable(name, incremental, onlyPrimary),
+ new CreateSnapshotCallable(name, incremental, onlyPrimary, dump),
options(Collections.singletonList(crd)).withFailoverDisabled()
));
}
- if (!CU.isPersistenceEnabled(cctx.gridConfig())) {
+ if (!CU.isPersistenceEnabled(cctx.gridConfig()) && !dump) {
throw new IgniteException("Create snapshot request has been rejected. " +
"Snapshots on an in-memory clusters are not allowed.");
}
@@ -2227,12 +2273,15 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
lastSeenSnpFut = snpFut0;
}
- List<String> grps = cctx.cache().persistentGroups().stream()
+ List<String> grps = (dump ? cctx.cache().cacheGroupDescriptors().values() : cctx.cache().persistentGroups()).stream()
.filter(g -> cctx.cache().cacheType(g.cacheOrGroupName()) == CacheType.USER)
.map(CacheGroupDescriptor::cacheOrGroupName)
.collect(Collectors.toList());
- grps.add(METASTORAGE_CACHE_NAME);
+ if (!dump)
+ grps.add(METASTORAGE_CACHE_NAME);
+ else if (grps.isEmpty())
+ throw new IgniteException("Dump operation has been rejected. No cache group defined in cluster");
List<ClusterNode> srvNodes = cctx.discovery().serverNodes(AffinityTopologyVersion.NONE);
@@ -2259,7 +2308,8 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
bltNodeIds,
incremental,
incIdx,
- onlyPrimary
+ onlyPrimary,
+ dump
));
String msg =
@@ -2408,7 +2458,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
if (INC_SNP_NAME_PATTERN.matcher(snpDir.getName()).matches() && snpDir.getAbsolutePath().contains(INC_SNP_DIR))
U.delete(snpDir);
else
- deleteSnapshot(snpDir, pdsSettings.folderName());
+ deleteSnapshot(snpDir, pdsSettings);
if (log.isInfoEnabled()) {
log.info("Previous attempt to create snapshot fail due to the local node crash. All resources " +
@@ -2504,7 +2554,10 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
* @param grps List of cache groups which will be destroyed.
*/
public void onCacheGroupsStopped(List<Integer> grps) {
- for (AbstractSnapshotFutureTask<?> sctx : F.view(locSnpTasks.values(), t -> t instanceof SnapshotFutureTask)) {
+ Collection<AbstractSnapshotFutureTask<?>> tasks =
+ F.view(locSnpTasks.values(), t -> t instanceof SnapshotFutureTask || t instanceof CreateDumpFutureTask);
+
+ for (AbstractSnapshotFutureTask<?> sctx : tasks) {
Set<Integer> retain = new HashSet<>(grps);
retain.retainAll(sctx.affectedCacheGroups());
@@ -2636,6 +2689,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
* @param requestId Snapshot operation request ID.
* @param parts Collection of pairs group and appropriate cache partition to be snapshot.
* @param withMetaStorage {@code true} if all metastorage data must be also included into snapshot.
+ * @param dump {@code true} if cache group dump must be created.
* @param snpSndr Factory which produces snapshot receiver instance.
* @return Snapshot operation task which should be registered on checkpoint to run.
*/
@@ -2645,10 +2699,12 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
UUID requestId,
Map<Integer, Set<Integer>> parts,
boolean withMetaStorage,
+ boolean dump,
SnapshotSender snpSndr
) {
- AbstractSnapshotFutureTask<?> task = registerTask(snpName, new SnapshotFutureTask(cctx, srcNodeId, requestId,
- snpName, tmpWorkDir, ioFactory, snpSndr, parts, withMetaStorage, locBuff));
+ AbstractSnapshotFutureTask<?> task = registerTask(snpName, dump
+ ? new CreateDumpFutureTask(cctx, srcNodeId, requestId, snpName, snapshotLocalDir(snpName, null), ioFactory, snpSndr, parts)
+ : new SnapshotFutureTask(cctx, srcNodeId, requestId, snpName, tmpWorkDir, ioFactory, snpSndr, parts, withMetaStorage, locBuff));
if (!withMetaStorage) {
for (Integer grpId : parts.keySet()) {
@@ -2784,6 +2840,22 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
}
}
+ /** */
+ private void removeDumpLock(String dumpName) throws IgniteCheckedException {
+ File lock = new File(nodeDumpDirectory(snapshotLocalDir(dumpName, null), cctx), DUMP_LOCK);
+
+ if (!lock.exists())
+ return;
+
+ if (!lock.delete())
+ throw new IgniteCheckedException("Lock file can't be deleted: " + lock);
+ }
+
+ /** */
+ public static File nodeDumpDirectory(File dumpDir, GridCacheSharedContext<?, ?> cctx) throws IgniteCheckedException {
+ return new File(dumpDir, databaseRelativePath(cctx.kernalContext().pdsFolderResolver().resolveFolders().folderName()));
+ }
+
/**
* Disables creation of incremental snapshots for the given cache group.
*
@@ -3775,8 +3847,6 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
reqMsg0.requestId(),
snpName,
reqMsg0.snapshotPath(),
- tmpWorkDir,
- ioFactory,
rmtSndrFactory.apply(rqId, nodeId),
reqMsg0.parts()));
@@ -4220,7 +4290,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
log.info("The Local snapshot sender closed. All resources released [dbNodeSnpDir=" + dbDir + ']');
}
else {
- deleteSnapshot(snpLocDir, pdsSettings.folderName());
+ deleteSnapshot(snpLocDir, pdsSettings);
if (log.isDebugEnabled())
log.debug("Local snapshot sender closed due to an error occurred: " + th.getMessage());
@@ -4538,25 +4608,44 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/** If {@code true} snapshot only primary copies of partitions. */
private final boolean onlyPrimary;
+ /** If {@code true} create cache dump. */
+ private final boolean dump;
+
/** Auto-injected grid instance. */
@IgniteInstanceResource
private transient IgniteEx ignite;
/**
* @param snpName Snapshot name.
+ * @param incremental If {@code true} then incremental snapshot must be created.
+ * @param onlyPrimary If {@code true} then only copy of primary partitions will be created.
+ * @param dump If {@code true} then cache dump must be created.
*/
- public CreateSnapshotCallable(String snpName, boolean incremental, boolean onlyPrimary) {
+ public CreateSnapshotCallable(
+ String snpName,
+ boolean incremental,
+ boolean onlyPrimary,
+ boolean dump
+ ) {
this.snpName = snpName;
this.incremental = incremental;
this.onlyPrimary = onlyPrimary;
+ this.dump = dump;
}
/** {@inheritDoc} */
- @Override public Void call() throws Exception {
+ @Override public Void call() {
if (incremental)
ignite.snapshot().createIncrementalSnapshot(snpName).get();
- else
- ignite.context().cache().context().snapshotMgr().createSnapshot(snpName, null, false, onlyPrimary).get();
+ else {
+ ignite.context().cache().context().snapshotMgr().createSnapshot(
+ snpName,
+ null,
+ false,
+ onlyPrimary,
+ dump
+ ).get();
+ }
return null;
}
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 1ffb31fe9a5..6d877b760d3 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
@@ -35,7 +35,6 @@ 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;
-import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -75,8 +74,6 @@ class IncrementalSnapshotFutureTask extends AbstractSnapshotFutureTask<Void> imp
SnapshotMetadata meta,
@Nullable String snpPath,
int incIdx,
- File tmpWorkDir,
- FileIOFactory ioFactory,
WALPointer lowPtr,
IgniteInternalFuture<WALPointer> highPtrFut
) {
@@ -85,8 +82,6 @@ class IncrementalSnapshotFutureTask extends AbstractSnapshotFutureTask<Void> imp
srcNodeId,
reqNodeId,
meta.snapshotName(),
- tmpWorkDir,
- ioFactory,
new SnapshotSender(
cctx.logger(IncrementalSnapshotFutureTask.class),
cctx.kernalContext().pools().getSnapshotExecutorService()
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java
index e32ddc1d00b..1ba5100f66a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java
@@ -25,7 +25,7 @@ public class SnapshotFinishedFutureTask extends AbstractSnapshotFutureTask<Void>
* @param e Finished snapshot task future with particular exception.
*/
public SnapshotFinishedFutureTask(IgniteCheckedException e) {
- super(null, null, null, null, null, null, null, null);
+ super(null, null, null, null, null, null);
onDone(e);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTask.java
index 793316949f2..ddae6250e03 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTask.java
@@ -24,10 +24,7 @@ import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -35,7 +32,6 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicIntegerArray;
import java.util.concurrent.atomic.AtomicLong;
@@ -47,9 +43,7 @@ import java.util.function.BiConsumer;
import java.util.function.BooleanSupplier;
import java.util.stream.Collectors;
import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.binary.BinaryType;
import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.internal.IgniteFutureCancelledCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.pagemem.PageIdUtils;
import org.apache.ignite.internal.pagemem.store.PageStore;
@@ -57,8 +51,6 @@ import org.apache.ignite.internal.pagemem.store.PageWriteListener;
import org.apache.ignite.internal.pagemem.wal.record.delta.ClusterSnapshotRecord;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
-import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
-import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointListener;
import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
@@ -70,20 +62,16 @@ import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
import org.apache.ignite.internal.processors.cache.persistence.wal.crc.FastCrc;
import org.apache.ignite.internal.processors.compress.CompressionProcessor;
-import org.apache.ignite.internal.processors.marshaller.MappedName;
import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageImpl;
import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.lang.IgniteThrowableRunner;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.C3;
-import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.jetbrains.annotations.Nullable;
-import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cacheWorkDir;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.getPartitionFile;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.copy;
@@ -99,13 +87,19 @@ import static org.apache.ignite.internal.processors.cache.persistence.snapshot.I
* If partitions for particular cache group are not provided that they will be collected and added
* on checkpoint under the write-lock.
*/
-class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskResult> implements CheckpointListener {
+class SnapshotFutureTask extends AbstractCreateSnapshotFutureTask implements CheckpointListener {
/** File page store manager for accessing cache group associated files. */
private final FilePageStoreManager pageStore;
/** Local buffer to perform copy-on-write operations for {@link PageStoreSerialWriter}. */
private final ThreadLocal<ByteBuffer> locBuff;
+ /** Snapshot working directory on file system. */
+ private final File tmpSnpWorkDir;
+
+ /** IO factory which will be used for creating snapshot delta-writers. */
+ private final FileIOFactory ioFactory;
+
/**
* The length of file size per each cache partition file.
* Partition has value greater than zero only for partitions in OWNING state.
@@ -130,9 +124,6 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
/** {@code true} if all metastorage data must be also included into snapshot. */
private final boolean withMetaStorage;
- /** Cache group and corresponding partitions collected under the checkpoint write lock. */
- private final Map<Integer, Set<Integer>> processed = new HashMap<>();
-
/** Checkpoint end future. */
private final CompletableFuture<Boolean> cpEndFut = new CompletableFuture<>();
@@ -142,9 +133,6 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
/** Absolute path to save intermediate results of cache partitions of this node. */
private volatile File tmpConsIdDir;
- /** Future which will be completed when task requested to be closed. Will be executed on system pool. */
- private volatile CompletableFuture<Void> closeFut;
-
/** Pointer to {@link ClusterSnapshotRecord}. */
private volatile @Nullable WALPointer snpPtr;
@@ -184,7 +172,7 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
boolean withMetaStorage,
ThreadLocal<ByteBuffer> locBuff
) {
- super(cctx, srcNodeId, reqId, snpName, tmpWorkDir, ioFactory, snpSndr, parts);
+ super(cctx, srcNodeId, reqId, snpName, snpSndr, parts);
assert snpName != null : "Snapshot name cannot be empty or null.";
assert snpSndr != null : "Snapshot sender which handles execution tasks must be not null.";
@@ -192,6 +180,8 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
assert cctx.pageStore() instanceof FilePageStoreManager : "Snapshot task can work only with physical files.";
assert !parts.containsKey(MetaStorage.METASTORAGE_CACHE_ID) : "The withMetaStorage must be used instead.";
+ this.tmpSnpWorkDir = new File(tmpWorkDir, snpName);
+ this.ioFactory = ioFactory;
this.withMetaStorage = withMetaStorage;
this.pageStore = (FilePageStoreManager)cctx.pageStore();
this.locBuff = locBuff;
@@ -204,13 +194,9 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
if (th == null)
return;
- if (err.compareAndSet(null, th))
- closeAsync();
+ super.acceptException(th);
startedFut.onDone(th);
-
- if (!(th instanceof IgniteFutureCancelledCheckedException))
- U.error(log, "Snapshot task has accepted exception to stop", th);
}
/** {@inheritDoc} */
@@ -248,13 +234,6 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
return startedFut;
}
- /**
- * @return {@code true} if current task requested to be stopped.
- */
- private boolean stopping() {
- return err.get() != null;
- }
-
/**
* Initiates snapshot task.
*
@@ -359,66 +338,7 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
ctx.walFlush(true);
}
- for (Map.Entry<Integer, Set<Integer>> e : parts.entrySet()) {
- int grpId = e.getKey();
- Set<Integer> grpParts = e.getValue();
-
- CacheGroupContext gctx = cctx.cache().cacheGroup(grpId);
-
- Iterator<GridDhtLocalPartition> iter;
-
- if (grpParts == null)
- iter = gctx.topology().currentLocalPartitions().iterator();
- else {
- if (grpParts.contains(INDEX_PARTITION)) {
- throw new IgniteCheckedException("Index partition cannot be included into snapshot if " +
- " set of cache group partitions has been explicitly provided [grpId=" + grpId + ']');
- }
-
- iter = F.iterator(grpParts, gctx.topology()::localPartition, false);
- }
-
- Set<Integer> owning = new HashSet<>();
- Set<Integer> missed = new HashSet<>();
-
- // Iterate over partitions in particular cache group.
- while (iter.hasNext()) {
- GridDhtLocalPartition part = iter.next();
-
- // Partition can be in MOVING\RENTING states.
- // Index partition will be excluded if not all partition OWNING.
- // There is no data assigned to partition, thus it haven't been created yet.
- if (part.state() == GridDhtPartitionState.OWNING)
- owning.add(part.id());
- else
- missed.add(part.id());
- }
-
- boolean affNode = gctx.nodeFilter() == null || gctx.nodeFilter().apply(cctx.localNode());
-
- if (grpParts != null) {
- // Partition has been provided for cache group, but some of them are not in OWNING state.
- // Exit with an error.
- if (!missed.isEmpty()) {
- throw new IgniteCheckedException("Snapshot operation cancelled due to " +
- "not all of requested partitions has OWNING state on local node [grpId=" + grpId +
- ", missed=" + S.toStringSortedDistinct(missed) + ']');
- }
- }
- else {
- // Partitions have not been provided for snapshot task and all partitions have
- // OWNING state, so index partition must be included into snapshot.
- if (!missed.isEmpty()) {
- log.warning("All local cache group partitions in OWNING state have been included into a snapshot. " +
- "Partitions which have different states skipped. Index partitions has also been skipped " +
- "[snpName=" + snpName + ", grpId=" + grpId + ", missed=" + S.toStringSortedDistinct(missed) + ']');
- }
- else if (affNode && missed.isEmpty() && cctx.kernalContext().query().moduleEnabled())
- owning.add(INDEX_PARTITION);
- }
-
- processed.put(grpId, owning);
- }
+ processPartitions();
List<CacheConfiguration<?, ?>> ccfgs = new ArrayList<>();
@@ -465,118 +385,81 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
if (!startedFut.onDone())
return;
- // Submit all tasks for partitions and deltas processing.
- List<CompletableFuture<Void>> futs = new ArrayList<>();
-
if (log.isInfoEnabled()) {
log.info("Submit partition processing tasks to the snapshot execution pool " +
"[map=" + groupByGroupId(partFileLengths.keySet()) +
", totalSize=" + U.humanReadableByteCount(partFileLengths.values().stream().mapToLong(v -> v).sum()) + ']');
}
- Collection<BinaryType> binTypesCopy = cctx.kernalContext()
- .cacheObjects()
- .metadata(Collections.emptyList())
- .values();
-
- // Process binary meta.
- futs.add(CompletableFuture.runAsync(
- wrapExceptionIfStarted(() -> snpSndr.sendBinaryMeta(binTypesCopy)),
- snpSndr.executor()));
-
- List<Map<Integer, MappedName>> mappingsCopy = cctx.kernalContext()
- .marshallerContext()
- .getCachedMappings();
-
- // Process marshaller meta.
- futs.add(CompletableFuture.runAsync(
- wrapExceptionIfStarted(() -> snpSndr.sendMarshallerMeta(mappingsCopy)),
- snpSndr.executor()));
-
- // Send configuration files of all cache groups.
- for (CacheConfigurationSender ccfgSndr : ccfgSndrs)
- futs.add(CompletableFuture.runAsync(wrapExceptionIfStarted(ccfgSndr::sendCacheConfig), snpSndr.executor()));
+ saveSnapshotData();
+ }
- try {
- for (Map.Entry<Integer, Set<Integer>> e : processed.entrySet()) {
- int grpId = e.getKey();
- String cacheDirName = pageStore.cacheDirName(grpId);
+ /** {@inheritDoc} */
+ @Override protected List<CompletableFuture<Void>> saveGroup(int grpId, Set<Integer> grpParts) throws IgniteCheckedException {
+ String cacheDirName = pageStore.cacheDirName(grpId);
- // Process partitions for a particular cache group.
- for (int partId : e.getValue()) {
- GroupPartitionId pair = new GroupPartitionId(grpId, partId);
+ // Process partitions for a particular cache group.
+ return grpParts.stream().map(partId -> {
+ GroupPartitionId pair = new GroupPartitionId(grpId, partId);
- Long partLen = partFileLengths.get(pair);
+ Long partLen = partFileLengths.get(pair);
- totalSize.addAndGet(partLen);
+ totalSize.addAndGet(partLen);
- CompletableFuture<Void> fut0 = CompletableFuture.runAsync(
- wrapExceptionIfStarted(() -> {
- snpSndr.sendPart(
- getPartitionFile(pageStore.workDir(), cacheDirName, partId),
- cacheDirName,
- pair,
- partLen);
+ return runAsync(() -> {
+ snpSndr.sendPart(
+ getPartitionFile(pageStore.workDir(), cacheDirName, partId),
+ cacheDirName,
+ pair,
+ partLen);
- // Stop partition writer.
- partDeltaWriters.get(pair).markPartitionProcessed();
+ // Stop partition writer.
+ partDeltaWriters.get(pair).markPartitionProcessed();
- processedSize.addAndGet(partLen);
- }),
- snpSndr.executor())
- // Wait for the completion of both futures - checkpoint end, copy partition.
- .runAfterBothAsync(cpEndFut,
- wrapExceptionIfStarted(() -> {
- PageStoreSerialWriter writer = partDeltaWriters.get(pair);
+ processedSize.addAndGet(partLen);
- writer.close();
+ // Wait for the completion of both futures - checkpoint end, copy partition.
+ }).runAfterBothAsync(cpEndFut, wrapExceptionIfStarted(() -> {
+ PageStoreSerialWriter writer = partDeltaWriters.get(pair);
- File delta = writer.deltaFile;
+ writer.close();
- try {
- // Atomically creates a new, empty delta file if and only if
- // a file with this name does not yet exist.
- delta.createNewFile();
- }
- catch (IOException ex) {
- throw new IgniteCheckedException(ex);
- }
+ File delta = writer.deltaFile;
- snpSndr.sendDelta(delta, cacheDirName, pair);
+ try {
+ // Atomically creates a new, empty delta file if and only if
+ // a file with this name does not yet exist.
+ delta.createNewFile();
+ }
+ catch (IOException ex) {
+ throw new IgniteCheckedException(ex);
+ }
- processedSize.addAndGet(delta.length());
+ snpSndr.sendDelta(delta, cacheDirName, pair);
- boolean deleted = delta.delete();
+ processedSize.addAndGet(delta.length());
- assert deleted;
+ boolean deleted = delta.delete();
- File deltaIdx = partDeltaIndexFile(delta);
+ assert deleted;
- if (deltaIdx.exists()) {
- deleted = deltaIdx.delete();
+ File deltaIdx = partDeltaIndexFile(delta);
- assert deleted;
- }
- }),
- snpSndr.executor());
+ if (deltaIdx.exists()) {
+ deleted = deltaIdx.delete();
- futs.add(fut0);
+ assert deleted;
}
- }
-
- int futsSize = futs.size();
-
- CompletableFuture.allOf(futs.toArray(new CompletableFuture[futsSize]))
- .whenComplete((res, t) -> {
- assert t == null : "Exception must never be thrown since a wrapper is used " +
- "for each snapshot task: " + t;
+ }), snpSndr.executor());
+ }).collect(Collectors.toList());
+ }
- closeAsync();
- });
- }
- catch (IgniteCheckedException e) {
- acceptException(e);
- }
+ /** {@inheritDoc} */
+ @Override protected List<CompletableFuture<Void>> saveCacheConfigs() {
+ // Send configuration files of all cache groups.
+ return ccfgSndrs.stream()
+ .map(ccfgSndr -> runAsync(ccfgSndr::sendCacheConfig))
+ .collect(Collectors.toList());
}
/**
@@ -585,7 +468,7 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
* @param dirName Directory name to init.
* @throws IgniteCheckedException If fails.
*/
- private void addPartitionWriters(int grpId, Set<Integer> parts, String dirName) throws IgniteCheckedException {
+ void addPartitionWriters(int grpId, Set<Integer> parts, String dirName) throws IgniteCheckedException {
Integer encGrpId = cctx.cache().isEncrypted(grpId) ? grpId : null;
for (int partId : parts) {
@@ -600,28 +483,8 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
}
}
- /**
- * @param exec Runnable task to execute.
- * @return Wrapped task.
- */
- private Runnable wrapExceptionIfStarted(IgniteThrowableRunner exec) {
- return () -> {
- if (stopping())
- return;
-
- try {
- exec.run();
- }
- catch (Throwable t) {
- acceptException(t);
- }
- };
- }
-
- /**
- * @return Future which will be completed when operations truly stopped.
- */
- public synchronized CompletableFuture<Void> closeAsync() {
+ /** {@inheritDoc} */
+ @Override public synchronized CompletableFuture<Void> closeAsync() {
if (closeFut == null) {
Throwable err0 = err.get();
@@ -648,22 +511,6 @@ class SnapshotFutureTask extends AbstractSnapshotFutureTask<SnapshotFutureTaskRe
return processedSize.get();
}
- /** {@inheritDoc} */
- @Override public boolean cancel() {
- super.cancel();
-
- try {
- closeAsync().get();
- }
- catch (InterruptedException | ExecutionException e) {
- U.error(log, "SnapshotFutureTask cancellation failed", e);
-
- return false;
- }
-
- return true;
- }
-
/**
* @param grps List of processing pairs.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTaskResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTaskResult.java
index 0f2acac059e..1203825b723 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTaskResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTaskResult.java
@@ -27,7 +27,7 @@ import org.jetbrains.annotations.Nullable;
/**
* Represents result of {@link SnapshotFutureTask}.
*/
-class SnapshotFutureTaskResult {
+public class SnapshotFutureTaskResult {
/** Partitions for which snapshot was created. */
private final Set<GroupPartitionId> parts;
@@ -35,7 +35,7 @@ class SnapshotFutureTaskResult {
private final @Nullable WALPointer snpPtr;
/** */
- SnapshotFutureTaskResult(Set<GroupPartitionId> parts, @Nullable WALPointer snpPtr) {
+ public SnapshotFutureTaskResult(Set<GroupPartitionId> parts, @Nullable WALPointer snpPtr) {
this.parts = Collections.unmodifiableSet(parts);
this.snpPtr = snpPtr;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java
index 2f2a3abb9dd..cd04a5f80ef 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java
@@ -101,6 +101,9 @@ public class SnapshotMetadata implements Serializable {
/** If {@code true} snapshot only primary copies of partitions. */
private boolean onlyPrimary;
+ /** If {@code true} cache group dump stored. */
+ private boolean dump;
+
/**
* @param rqId Unique request id.
* @param snpName Snapshot name.
@@ -112,6 +115,7 @@ public class SnapshotMetadata implements Serializable {
* @param snpRecPtr WAL pointer to {@link ClusterSnapshotRecord} if exists.
* @param masterKeyDigest Master key digest for encrypted caches.
* @param onlyPrimary If {@code true} snapshot only primary copies of partitions.
+ * @param dump If {@code true} cache group dump stored.
*/
public SnapshotMetadata(
UUID rqId,
@@ -125,7 +129,8 @@ public class SnapshotMetadata implements Serializable {
Set<GroupPartitionId> pairs,
@Nullable WALPointer snpRecPtr,
@Nullable byte[] masterKeyDigest,
- boolean onlyPrimary
+ boolean onlyPrimary,
+ boolean dump
) {
this.rqId = rqId;
this.snpName = snpName;
@@ -137,6 +142,7 @@ public class SnapshotMetadata implements Serializable {
this.snpRecPtr = snpRecPtr;
this.masterKeyDigest = masterKeyDigest;
this.onlyPrimary = onlyPrimary;
+ this.dump = dump;
if (!F.isEmpty(compGrpIds)) {
hasComprGrps = true;
@@ -228,6 +234,11 @@ public class SnapshotMetadata implements Serializable {
return onlyPrimary;
}
+ /** @return If {@code true} then metadata describes cache dump. */
+ public boolean dump() {
+ return dump;
+ }
+
/** Save the state of this <tt>HashMap</tt> partitions and cache groups to a stream. */
private void writeObject(java.io.ObjectOutputStream s)
throws java.io.IOException {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationRequest.java
index 05008820bf4..90777b1cc56 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationRequest.java
@@ -91,6 +91,9 @@ public class SnapshotOperationRequest implements Serializable {
/** If {@code true} snapshot only primary copies of partitions. */
private final boolean onlyPrimary;
+ /** If {@code true} then create dump. */
+ private final boolean dump;
+
/**
* @param reqId Request ID.
* @param opNodeId Operational node ID.
@@ -101,6 +104,7 @@ public class SnapshotOperationRequest implements Serializable {
* @param incremental {@code True} if incremental snapshot requested.
* @param incIdx Incremental snapshot index.
* @param onlyPrimary If {@code true} snapshot only primary copies of partitions.
+ * @param dump If {@code true} then create dump.
*/
public SnapshotOperationRequest(
UUID reqId,
@@ -111,7 +115,8 @@ public class SnapshotOperationRequest implements Serializable {
Set<UUID> nodes,
boolean incremental,
int incIdx,
- boolean onlyPrimary
+ boolean onlyPrimary,
+ boolean dump
) {
this.reqId = reqId;
this.opNodeId = opNodeId;
@@ -122,6 +127,7 @@ public class SnapshotOperationRequest implements Serializable {
this.incremental = incremental;
this.incIdx = incIdx;
this.onlyPrimary = onlyPrimary;
+ this.dump = dump;
startTime = U.currentTimeMillis();
}
@@ -196,6 +202,11 @@ public class SnapshotOperationRequest implements Serializable {
return onlyPrimary;
}
+ /** @return If {@code true} then create dump. */
+ public boolean dump() {
+ return dump;
+ }
+
/** @return Start time. */
public long startTime() {
return startTime;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java
index 3547d5edf30..65da2f35ada 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java
@@ -68,12 +68,14 @@ import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.fromOrdinal;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DATA_FILENAME;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.FILE_SUFFIX;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cacheDirectories;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cacheGroupName;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cachePartitionFiles;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.partId;
import static org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId.getTypeByPartId;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.databaseRelativePath;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.CreateDumpFutureTask.DUMP_FILE_EXT;
import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.calculatePartitionHash;
import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.checkPartitionsPageCrcSum;
@@ -132,7 +134,7 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
Set<Integer> parts = meta.partitions().get(grpId) == null ? Collections.emptySet() :
new HashSet<>(meta.partitions().get(grpId));
- for (File part : cachePartitionFiles(dir)) {
+ for (File part : cachePartitionFiles(dir, meta.dump() ? DUMP_FILE_EXT : FILE_SUFFIX)) {
int partId = partId(part.getName());
if (!parts.remove(partId))
@@ -156,6 +158,7 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
", meta=" + meta + ']');
}
+ // This will throw if compression disabled. Calculation before other checks.
boolean punchHoleEnabled = isPunchHoleEnabled(opCtx, grpDirs.keySet());
if (!opCtx.check()) {
@@ -164,6 +167,17 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
return Collections.emptyMap();
}
+ return meta.dump() ? checkDumpFiles() : checkSnapshotFiles(opCtx, grpDirs, meta, partFiles, punchHoleEnabled);
+ }
+
+ /** */
+ private Map<PartitionKeyV2, PartitionHashRecordV2> checkSnapshotFiles(
+ SnapshotHandlerContext opCtx,
+ Map<Integer, File> grpDirs,
+ SnapshotMetadata meta,
+ Set<File> partFiles,
+ boolean punchHoleEnabled
+ ) throws IgniteCheckedException {
Map<PartitionKeyV2, PartitionHashRecordV2> res = new ConcurrentHashMap<>();
ThreadLocal<ByteBuffer> buff = ThreadLocal.withInitial(() -> ByteBuffer.allocateDirect(meta.pageSize())
.order(ByteOrder.nativeOrder()));
@@ -294,6 +308,11 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
return res;
}
+ /** */
+ private Map<PartitionKeyV2, PartitionHashRecordV2> checkDumpFiles() {
+ return Collections.emptyMap();
+ }
+
/** {@inheritDoc} */
@Override public void complete(String name,
Collection<SnapshotHandlerResult<Map<PartitionKeyV2, PartitionHashRecordV2>>> results) throws IgniteCheckedException {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotResponseRemoteFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotResponseRemoteFutureTask.java
index 90e562f4265..ddb8332e2f4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotResponseRemoteFutureTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotResponseRemoteFutureTask.java
@@ -28,7 +28,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import org.apache.ignite.IgniteException;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
-import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
import org.apache.ignite.internal.util.typedef.F;
import org.jetbrains.annotations.Nullable;
@@ -48,8 +47,6 @@ public class SnapshotResponseRemoteFutureTask extends AbstractSnapshotFutureTask
* @param reqId Snapshot operation request ID.
* @param snpName Unique identifier of snapshot process.
* @param snpPath Snapshot directory path.
- * @param tmpWorkDir Working directory for intermediate snapshot results.
- * @param ioFactory Factory to working with snapshot files.
* @param snpSndr Factory which produces snapshot receiver instance.
* @param parts Partition to be processed.
*/
@@ -59,12 +56,10 @@ public class SnapshotResponseRemoteFutureTask extends AbstractSnapshotFutureTask
UUID reqId,
String snpName,
String snpPath,
- File tmpWorkDir,
- FileIOFactory ioFactory,
SnapshotSender snpSndr,
Map<Integer, Set<Integer>> parts
) {
- super(cctx, srcNodeId, reqId, snpName, tmpWorkDir, ioFactory, snpSndr, parts);
+ super(cctx, srcNodeId, reqId, snpName, snpSndr, parts);
this.snpPath = snpPath;
}
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 074a1b7227e..7fa931d5864 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
@@ -420,7 +420,8 @@ public class SnapshotRestoreProcess {
new HashSet<>(bltNodes),
false,
incIdx,
- onlyPrimary
+ onlyPrimary,
+ false
);
prepareRestoreProc.start(req.requestId(), req);
@@ -791,8 +792,9 @@ public class SnapshotRestoreProcess {
ctx.compress().checkPageCompressionSupported(path.toPath(), meta.pageSize());
}
catch (Exception e) {
- String grpWithCompr = req.groups().stream().filter(s -> meta.isGroupWithCompresion(CU.cacheId(grpName)))
- .collect(Collectors.joining(", "));
+ String grpWithCompr = F.isEmpty(req.groups()) ? ""
+ : req.groups().stream().filter(s -> meta.isGroupWithCompresion(CU.cacheId(grpName)))
+ .collect(Collectors.joining(", "));
String msg = "Requested cache groups [" + grpWithCompr + "] for restore " +
"from snapshot '" + meta.snapshotName() + "' are compressed while " +
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotSender.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotSender.java
index c48f899fc7d..010ade1952f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotSender.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotSender.java
@@ -33,7 +33,7 @@ import org.jetbrains.annotations.Nullable;
/**
*
*/
-abstract class SnapshotSender {
+public abstract class SnapshotSender {
/** Busy processing lock. */
private final ReadWriteLock lock = new ReentrantReadWriteLock();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/CreateDumpFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/CreateDumpFutureTask.java
new file mode 100644
index 00000000000..413e24ba852
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/CreateDumpFutureTask.java
@@ -0,0 +1,544 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.snapshot.dump;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.concurrent.locks.LockSupport;
+import java.util.stream.Collectors;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.StoredCacheData;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+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.partstate.GroupPartitionId;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.AbstractCreateSnapshotFutureTask;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFutureTaskResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotSender;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.GridConcurrentHashSet;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.internal.util.lang.GridCloseableIterator;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION;
+import static org.apache.ignite.internal.processors.cache.GridLocalConfigManager.cacheDataFilename;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DIR_PREFIX;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_GRP_DIR_PREFIX;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.PART_FILE_PREFIX;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.DUMP_LOCK;
+
+/**
+ * Task creates cache group dump.
+ * Dump is a consistent snapshot of cache entries.
+ * Directories structure is same as a full snapshot but each partitions saved in "part-0.dump" file.
+ * Files structure is a set of {@link DumpEntry} written one by one.
+ *
+ * @see Dump
+ * @see DumpEntry
+ */
+public class CreateDumpFutureTask extends AbstractCreateSnapshotFutureTask implements DumpEntryChangeListener {
+ /** Dump files name. */
+ public static final String DUMP_FILE_EXT = ".dump";
+
+ /** Root dump directory. */
+ private final File dumpDir;
+
+ /** */
+ private final FileIOFactory ioFactory;
+
+ /**
+ * Dump contextes.
+ * Key is [group_id, partition_id] combined in single long value.
+ *
+ * @see #toLong(int, int)
+ */
+ private final Map<Long, PartitionDumpContext> dumpCtxs = new ConcurrentHashMap<>();
+
+ /** Local node is primary for set of group partitions. */
+ private final Map<Integer, Set<Integer>> grpPrimaries = new ConcurrentHashMap<>();
+
+ /**
+ * @param cctx Cache context.
+ * @param srcNodeId Node id which cause snapshot task creation.
+ * @param reqId Snapshot operation request ID.
+ * @param dumpName Dump name.
+ * @param ioFactory IO factory.
+ * @param snpSndr Snapshot sender.
+ * @param parts Parts to dump.
+ */
+ public CreateDumpFutureTask(
+ GridCacheSharedContext<?, ?> cctx,
+ UUID srcNodeId,
+ UUID reqId,
+ String dumpName,
+ File dumpDir,
+ FileIOFactory ioFactory,
+ SnapshotSender snpSndr,
+ Map<Integer, Set<Integer>> parts
+ ) {
+ super(
+ cctx,
+ srcNodeId,
+ reqId,
+ dumpName,
+ snpSndr,
+ parts
+ );
+
+ this.dumpDir = dumpDir;
+ this.ioFactory = ioFactory;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean start() {
+ try {
+ if (log.isInfoEnabled())
+ log.info("Start cache dump [name=" + snpName + ", grps=" + parts.keySet() + ']');
+
+ createDumpLock();
+
+ processPartitions();
+
+ prepare();
+
+ saveSnapshotData();
+ }
+ catch (IgniteCheckedException | IOException e) {
+ acceptException(e);
+ }
+
+ return false; // Don't wait for checkpoint.
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void processPartitions() throws IgniteCheckedException {
+ super.processPartitions();
+
+ processed.values().forEach(parts -> parts.remove(INDEX_PARTITION));
+ }
+
+ /** Prepares all data structures to dump entries. */
+ private void prepare() throws IOException, IgniteCheckedException {
+ for (Map.Entry<Integer, Set<Integer>> e : processed.entrySet()) {
+ int grp = e.getKey();
+
+ File grpDumpDir = groupDirectory(cctx.cache().cacheGroup(grp));
+
+ if (!grpDumpDir.mkdirs())
+ throw new IgniteCheckedException("Dump directory can't be created: " + grpDumpDir);
+
+ CacheGroupContext gctx = cctx.cache().cacheGroup(grp);
+
+ for (GridCacheContext<?, ?> cctx : gctx.caches())
+ cctx.dumpListener(this);
+
+ grpPrimaries.put(
+ grp,
+ gctx.affinity().primaryPartitions(gctx.shared().kernalContext().localNodeId(), gctx.affinity().lastVersion())
+ );
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected List<CompletableFuture<Void>> saveCacheConfigs() {
+ return parts.keySet().stream().map(grp -> runAsync(() -> {
+ CacheGroupContext gctx = cctx.cache().cacheGroup(grp);
+
+ File grpDir = groupDirectory(gctx);
+
+ IgniteUtils.ensureDirectory(grpDir, "dump group directory", null);
+
+ for (GridCacheContext<?, ?> cacheCtx : gctx.caches()) {
+ DynamicCacheDescriptor desc = cctx.kernalContext().cache().cacheDescriptor(cacheCtx.cacheId());
+
+ StoredCacheData cacheData = new StoredCacheData(new CacheConfiguration(desc.cacheConfiguration()));
+
+ cacheData.queryEntities(desc.schema().entities());
+ cacheData.sql(desc.sql());
+
+ cctx.cache().configManager().writeCacheData(cacheData, new File(grpDir, cacheDataFilename(cacheData.config())));
+ }
+ })).collect(Collectors.toList());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected List<CompletableFuture<Void>> saveGroup(int grp, Set<Integer> grpParts) {
+ long start = System.currentTimeMillis();
+
+ AtomicLong entriesCnt = new AtomicLong();
+ AtomicLong changedEntriesCnt = new AtomicLong();
+
+ String name = cctx.cache().cacheGroup(grp).cacheOrGroupName();
+
+ CacheGroupContext gctx = cctx.kernalContext().cache().cacheGroup(grp);
+
+ if (log.isInfoEnabled())
+ log.info("Start group dump [name=" + name + ", id=" + grp + ']');
+
+ List<CompletableFuture<Void>> futs = grpParts.stream().map(part -> runAsync(() -> {
+ long entriesCnt0 = 0;
+
+ try (PartitionDumpContext dumpCtx = dumpContext(grp, part)) {
+ try (GridCloseableIterator<CacheDataRow> rows = gctx.offheap().reservedIterator(part, dumpCtx.topVer)) {
+ if (rows == null)
+ throw new IgniteCheckedException("Partition missing [part=" + part + ']');
+
+ while (rows.hasNext()) {
+ CacheDataRow row = rows.next();
+
+ assert row.partition() == part;
+
+ int cache = row.cacheId() == 0 ? grp : row.cacheId();
+
+ if (dumpCtx.writeForIterator(cache, row.expireTime(), row.key(), row.value(), row.version()))
+ entriesCnt0++;
+ }
+ }
+
+ entriesCnt.addAndGet(entriesCnt0);
+ changedEntriesCnt.addAndGet(dumpCtx.changedCnt.intValue());
+
+ if (log.isDebugEnabled()) {
+ log.debug("Finish group partition dump [name=" + name +
+ ", id=" + grp +
+ ", part=" + part +
+ ", time=" + (System.currentTimeMillis() - start) +
+ ", iteratorEntriesCount=" + entriesCnt +
+ ", changedEntriesCount=" + changedEntriesCnt + ']');
+
+ }
+ }
+ })).collect(Collectors.toList());
+
+ int futsSize = futs.size();
+
+ CompletableFuture.allOf(futs.toArray(new CompletableFuture[futsSize])).whenComplete((res, t) -> {
+ clearDumpListener(gctx);
+
+ if (log.isInfoEnabled()) {
+ log.info("Finish group dump [name=" + name +
+ ", id=" + grp +
+ ", time=" + (System.currentTimeMillis() - start) +
+ ", iteratorEntriesCount=" + entriesCnt +
+ ", changedEntriesCount=" + changedEntriesCnt + ']');
+ }
+ });
+
+ return futs;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void beforeChange(GridCacheContext cctx, KeyCacheObject key, CacheObject val, long expireTime, GridCacheVersion ver) {
+ try {
+ assert key.partition() != -1;
+
+ dumpContext(cctx.groupId(), key.partition()).writeChanged(cctx.cacheId(), expireTime, key, val, ver);
+ }
+ catch (IgniteException e) {
+ acceptException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CompletableFuture<Void> closeAsync() {
+ if (closeFut == null) {
+ dumpCtxs.values().forEach(PartitionDumpContext::close);
+
+ Throwable err0 = err.get();
+
+ Set<GroupPartitionId> taken = new HashSet<>();
+
+ for (Map.Entry<Integer, Set<Integer>> e : processed.entrySet()) {
+ int grp = e.getKey();
+
+ clearDumpListener(cctx.cache().cacheGroup(grp));
+
+ for (Integer part : e.getValue())
+ taken.add(new GroupPartitionId(grp, part));
+ }
+
+ closeFut = CompletableFuture.runAsync(
+ () -> onDone(new SnapshotFutureTaskResult(taken, null), err0),
+ cctx.kernalContext().pools().getSystemExecutorService()
+ );
+ }
+
+ return closeFut;
+ }
+
+ /** */
+ private void clearDumpListener(CacheGroupContext gctx) {
+ for (GridCacheContext<?, ?> cctx : gctx.caches())
+ cctx.dumpListener(null);
+ }
+
+ /** */
+ private void createDumpLock() throws IgniteCheckedException, IOException {
+ File nodeDumpDir = IgniteSnapshotManager.nodeDumpDirectory(dumpDir, cctx);
+
+ if (!nodeDumpDir.mkdirs())
+ throw new IgniteCheckedException("Can't create node dump directory: " + nodeDumpDir.getAbsolutePath());
+
+ File lock = new File(nodeDumpDir, DUMP_LOCK);
+
+ if (!lock.createNewFile())
+ throw new IgniteCheckedException("Lock file can't be created or already exists: " + lock.getAbsolutePath());
+ }
+
+ /** */
+ private PartitionDumpContext dumpContext(int grp, int part) {
+ return dumpCtxs.computeIfAbsent(
+ toLong(grp, part),
+ key -> new PartitionDumpContext(cctx.kernalContext().cache().cacheGroup(grp), part)
+ );
+ }
+
+ /**
+ * Context of dump single partition.
+ */
+ private class PartitionDumpContext implements Closeable {
+ /** Group id. */
+ final int grp;
+
+ /** Partition id. */
+ final int part;
+
+ /** Hashes of cache keys of entries changed by the user during partition dump. */
+ final Map<Integer, Set<KeyCacheObject>> changed;
+
+ /** Count of entries changed during dump creation. */
+ LongAdder changedCnt = new LongAdder();
+
+ /** Partition dump file. Lazily initialized to prevent creation files for empty partitions. */
+ final FileIO file;
+
+ /** Last version on time of dump start. Can be used only for primary. */
+ @Nullable final GridCacheVersion startVer;
+
+ /** Topology Version. */
+ private final AffinityTopologyVersion topVer;
+
+ /** Partition serializer. */
+ DumpEntrySerializer serdes;
+
+ /** If {@code true} context is closed. */
+ volatile boolean closed;
+
+ /** Count of writers. When count becomes {@code 0} context must be closed. */
+ private final AtomicInteger writers = new AtomicInteger(1); // Iterator writing entries to this context, by default.
+
+ /**
+ * @param gctx Group context.
+ * @param part Partition id.
+ */
+ public PartitionDumpContext(CacheGroupContext gctx, int part) {
+ assert gctx != null;
+
+ try {
+ this.part = part;
+ grp = gctx.groupId();
+ topVer = gctx.topology().lastTopologyChangeVersion();
+
+ startVer = grpPrimaries.get(gctx.groupId()).contains(part) ? gctx.shared().versions().last() : null;
+
+ serdes = new DumpEntrySerializer();
+ changed = new HashMap<>();
+
+ for (int cache : gctx.cacheIds())
+ changed.put(cache, new GridConcurrentHashSet<>());
+
+ File dumpFile = new File(groupDirectory(gctx), PART_FILE_PREFIX + part + DUMP_FILE_EXT);
+
+ if (!dumpFile.createNewFile())
+ throw new IgniteException("Dump file can't be created: " + dumpFile);
+
+ file = ioFactory.create(dumpFile);
+ }
+ catch (IOException | IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ /**
+ * Writes entry that changed while dump creation in progress.
+ * @param cache Cache id.
+ * @param expireTime Expire time.
+ * @param key Key.
+ * @param val Value before change.
+ * @param ver Version before change.
+ */
+ public void writeChanged(
+ int cache,
+ long expireTime,
+ KeyCacheObject key,
+ CacheObject val,
+ GridCacheVersion ver
+ ) {
+ String reasonToSkip = null;
+
+ if (closed) // Partition already saved in dump.
+ reasonToSkip = "partition already saved";
+ else {
+ writers.getAndIncrement();
+
+ try {
+ if (startVer != null && ver.isGreater(startVer))
+ reasonToSkip = "greater version";
+ else if (!changed.get(cache).add(key)) // Entry changed several time during dump.
+ reasonToSkip = "changed several times";
+ else if (val == null)
+ reasonToSkip = "newly created or already removed"; // Previous value is null. Entry created after dump start, skip.
+ else {
+ synchronized (this) {
+ if (closed) // Partition already saved in dump.
+ reasonToSkip = "partition already saved";
+ else {
+ write(cache, expireTime, key, val);
+
+ changedCnt.increment();
+ }
+ }
+ }
+ }
+ finally {
+ writers.decrementAndGet();
+ }
+ }
+
+ if (log.isTraceEnabled()) {
+ log.trace("Listener [grp=" + grp +
+ ", cache=" + cache +
+ ", part=" + part +
+ ", key=" + key +
+ ", written=" + (reasonToSkip == null ? "true" : reasonToSkip) + ']');
+ }
+ }
+
+ /**
+ * Writes entry fetched from partition iterator.
+ *
+ * @param cache Cache id.
+ * @param expireTime Expire time.
+ * @param key Key.
+ * @param val Value.
+ * @param ver Version.
+ * @return {@code True} if entry was written in dump,
+ * {@code false} if it was already written by {@link #writeChanged(int, long, KeyCacheObject, CacheObject, GridCacheVersion)}.
+ */
+ public boolean writeForIterator(
+ int cache,
+ long expireTime,
+ KeyCacheObject key,
+ CacheObject val,
+ GridCacheVersion ver
+ ) {
+ boolean written = true;
+
+ if (startVer != null && ver.isGreater(startVer))
+ written = false;
+ else if (changed.get(cache).contains(key))
+ written = false;
+ else {
+ synchronized (this) {
+ write(cache, expireTime, key, val);
+ }
+ }
+
+ if (log.isTraceEnabled()) {
+ log.trace("Iterator [" +
+ "grp=" + grp +
+ ", cache=" + cache +
+ ", part=" + part +
+ ", key=" + key +
+ ", written=" + written + ']');
+ }
+
+ return written;
+ }
+
+ /** */
+ private void write(int cache, long expireTime, KeyCacheObject key, CacheObject val) {
+ try {
+ ByteBuffer buf = serdes.writeToBuffer(cache, expireTime, key, val, cctx.cacheObjectContext(cache));
+
+ if (file.writeFully(buf) != buf.limit())
+ throw new IgniteException("Can't write row");
+ }
+ catch (IOException | IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() {
+ synchronized (this) {
+ if (closed)
+ return;
+
+ closed = true;
+ }
+
+ writers.decrementAndGet();
+
+ while (writers.get() > 0) // Waiting for all on the fly listeners to complete.
+ LockSupport.parkNanos(1_000_000);
+
+ U.closeQuiet(file);
+
+ serdes = null;
+ }
+ }
+
+ /** */
+ public static long toLong(int high, int low) {
+ return (((long)high) << Integer.SIZE) | (low & 0xffffffffL);
+ }
+
+ /** */
+ private File groupDirectory(CacheGroupContext grpCtx) throws IgniteCheckedException {
+ return new File(
+ IgniteSnapshotManager.nodeDumpDirectory(dumpDir, cctx),
+ (grpCtx.caches().size() > 1 ? CACHE_GRP_DIR_PREFIX : CACHE_DIR_PREFIX) + grpCtx.cacheOrGroupName()
+ );
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/Dump.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/Dump.java
new file mode 100644
index 00000000000..70a2e26de49
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/Dump.java
@@ -0,0 +1,263 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.snapshot.dump;
+
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.StoredCacheData;
+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.file.FilePageStoreManager;
+import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadata;
+import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.MarshallerUtils;
+import org.apache.ignite.marshaller.jdk.JdkMarshaller;
+
+import static org.apache.ignite.configuration.DataStorageConfiguration.DFLT_BINARY_METADATA_PATH;
+import static org.apache.ignite.configuration.DataStorageConfiguration.DFLT_MARSHALLER_PATH;
+import static org.apache.ignite.internal.processors.cache.GridLocalConfigManager.readCacheData;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DIR_PREFIX;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_GRP_DIR_PREFIX;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.DFLT_STORE_DIR;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.PART_FILE_PREFIX;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNAPSHOT_METAFILE_EXT;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.CreateDumpFutureTask.DUMP_FILE_EXT;
+
+/**
+ * This class provides ability to work with saved cache dump.
+ */
+public class Dump {
+ /** Dump directory. */
+ private final File dumpDir;
+
+ /** Kernal context. */
+ private final GridKernalContext cctx;
+
+ /**
+ * @param dumpDir Dump directory.
+ */
+ public Dump(GridKernalContext cctx, File dumpDir) throws IgniteCheckedException {
+ this.cctx = cctx;
+ this.dumpDir = dumpDir;
+
+ File binaryMeta = new File(dumpDir, DFLT_BINARY_METADATA_PATH);
+ File marshaller = new File(dumpDir, DFLT_MARSHALLER_PATH);
+
+ A.ensure(dumpDir != null, "dump directory is null");
+ A.ensure(dumpDir.exists(), "dump directory not exists");
+ A.ensure(binaryMeta.exists(), "binary metadata directory not exists");
+ A.ensure(marshaller.exists(), "marshaller directory not exists");
+ }
+
+ /** @return List of node directories. */
+ public List<String> nodesDirectories() {
+ return Arrays.stream(new File(dumpDir, DFLT_STORE_DIR).listFiles(f -> f.isDirectory() &&
+ !(f.getAbsolutePath().endsWith(DFLT_BINARY_METADATA_PATH)
+ || f.getAbsolutePath().endsWith(DFLT_MARSHALLER_PATH)))).map(File::getName).collect(Collectors.toList());
+ }
+
+ /** @return List of snapshot metadata saved in {@link #dumpDir}. */
+ public List<SnapshotMetadata> metadata() throws IOException, IgniteCheckedException {
+ JdkMarshaller marsh = MarshallerUtils.jdkMarshaller(cctx.igniteInstanceName());
+
+ ClassLoader clsLdr = U.resolveClassLoader(cctx.config());
+
+ return Arrays.stream(dumpDir.listFiles(f -> f.getName().endsWith(SNAPSHOT_METAFILE_EXT))).map(meta -> {
+ try (InputStream in = new BufferedInputStream(Files.newInputStream(meta.toPath()))) {
+ return marsh.<SnapshotMetadata>unmarshal(in, clsLdr);
+ }
+ catch (IOException | IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }).filter(SnapshotMetadata::dump).collect(Collectors.toList());
+ }
+
+ /**
+ * @param node Node directory name.
+ * @param group Group id.
+ * @return List of cache configs saved in dump for group.
+ */
+ public List<StoredCacheData> configs(String node, int group) {
+ JdkMarshaller marsh = MarshallerUtils.jdkMarshaller(cctx.igniteInstanceName());
+
+ return Arrays.stream(FilePageStoreManager.cacheDataFiles(dumpGroupDirectory(node, group))).map(f -> {
+ try {
+ return readCacheData(f, marsh, cctx.config());
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }).collect(Collectors.toList());
+ }
+
+ /**
+ * @param node Node directory name.
+ * @param group Group id.
+ * @return Dump iterator.
+ */
+ public List<Integer> partitions(String node, int group) {
+ return Arrays.stream(dumpGroupDirectory(node, group)
+ .listFiles(f -> f.getName().startsWith(PART_FILE_PREFIX) && f.getName().endsWith(DUMP_FILE_EXT)))
+ .map(partFile -> Integer.parseInt(partFile.getName().replace(PART_FILE_PREFIX, "").replace(DUMP_FILE_EXT, "")))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * @param node Node directory name.
+ * @param group Group id.
+ * @return Dump iterator.
+ */
+ public DumpedPartitionIterator iterator(String node, int group, int part) {
+ return iterator(node, group, part, true);
+ }
+
+ /**
+ * @param node Node directory name.
+ * @param group Group id.
+ * @param excludeDuplicates Skip entries that was dumped twice - by iterator and change listener.
+ * @return Dump iterator.
+ */
+ DumpedPartitionIterator iterator(String node, int group, int part, boolean excludeDuplicates) {
+ FileIOFactory ioFactory = new RandomAccessFileIOFactory();
+
+ FileIO dumpFile;
+
+ try {
+ dumpFile = ioFactory.create(new File(dumpGroupDirectory(node, group), PART_FILE_PREFIX + part + DUMP_FILE_EXT));
+ }
+ catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ DumpEntrySerializer serializer = new DumpEntrySerializer();
+
+ serializer.kernalContext(cctx);
+
+ return new DumpedPartitionIterator() {
+ DumpEntry next;
+
+ Set<KeyCacheObject> partKeys = new HashSet<>();
+
+ /** {@inheritDoc} */
+ @Override public boolean hasNext() {
+ advance();
+
+ return next != null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public DumpEntry next() {
+ advance();
+
+ if (next == null)
+ throw new NoSuchElementException();
+
+ DumpEntry next0 = next;
+
+ next = null;
+
+ return next0;
+ }
+
+ /** */
+ private void advance() {
+ if (next != null)
+ return;
+
+ try {
+ next = serializer.read(dumpFile, group, part);
+
+ /*
+ * During dumping entry can be dumped twice: by partition iterator and change listener.
+ * Excluding duplicates keys from iteration.
+ */
+ while (next != null && !partKeys.add(next.key()))
+ next = serializer.read(dumpFile, group, part);
+
+ if (next == null)
+ partKeys = null; // Let GC do the rest.
+ }
+ catch (IOException | IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() {
+ U.closeQuiet(dumpFile);
+
+ partKeys = null;
+ }
+ };
+ }
+
+ /** @return Root dump directory. */
+ public File dumpDirectory() {
+ return dumpDir;
+ }
+
+ /** */
+ private File dumpGroupDirectory(String node, int groupId) {
+ File nodeDir = Paths.get(dumpDir.getAbsolutePath(), DFLT_STORE_DIR, node).toFile();
+
+ assert nodeDir.exists() && nodeDir.isDirectory();
+
+ File[] grpDirs = nodeDir.listFiles(f -> {
+ if (!f.isDirectory()
+ || (!f.getName().startsWith(CACHE_DIR_PREFIX)
+ && !f.getName().startsWith(CACHE_GRP_DIR_PREFIX)))
+ return false;
+
+ String grpName = f.getName().startsWith(CACHE_DIR_PREFIX)
+ ? f.getName().replace(CACHE_DIR_PREFIX, "")
+ : f.getName().replace(CACHE_GRP_DIR_PREFIX, "");
+
+ return groupId == CU.cacheId(grpName);
+ });
+
+ if (grpDirs.length != 1)
+ throw new IgniteException("Wrong number of group directories: " + grpDirs.length);
+
+ return grpDirs[0];
+ }
+
+ /**
+ * Closeable dump iterator.
+ */
+ public interface DumpedPartitionIterator extends Iterator<DumpEntry>, AutoCloseable {
+ // No-op.
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/DumpEntry.java
similarity index 59%
copy from modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java
copy to modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/DumpEntry.java
index e32ddc1d00b..5054c97806c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/DumpEntry.java
@@ -15,28 +15,24 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.persistence.snapshot;
+package org.apache.ignite.internal.processors.cache.persistence.snapshot.dump;
-import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-/** */
-public class SnapshotFinishedFutureTask extends AbstractSnapshotFutureTask<Void> {
- /**
- * @param e Finished snapshot task future with particular exception.
- */
- public SnapshotFinishedFutureTask(IgniteCheckedException e) {
- super(null, null, null, null, null, null, null, null);
+/**
+ * Single cache entry from dump.
+ */
+public interface DumpEntry {
+ /** @return Cache id. */
+ public int cacheId();
- onDone(e);
- }
+ /** @return Expiration time. */
+ public long expireTime();
- /** {@inheritDoc} */
- @Override public boolean start() {
- return false;
- }
+ /** @return Key. */
+ public KeyCacheObject key();
- /** {@inheritDoc} */
- @Override public void acceptException(Throwable th) {
- onDone(th);
- }
+ /** @return Value. */
+ public CacheObject value();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/DumpEntryChangeListener.java
similarity index 52%
copy from modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java
copy to modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/DumpEntryChangeListener.java
index e32ddc1d00b..40669266a8c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFinishedFutureTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/DumpEntryChangeListener.java
@@ -15,28 +15,24 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.cache.persistence.snapshot;
+package org.apache.ignite.internal.processors.cache.persistence.snapshot.dump;
-import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.jetbrains.annotations.Nullable;
-/** */
-public class SnapshotFinishedFutureTask extends AbstractSnapshotFutureTask<Void> {
+/**
+ * Copy on write listener that handles changes of entries during dump.
+ */
+public interface DumpEntryChangeListener {
/**
- * @param e Finished snapshot task future with particular exception.
+ * @param cctx Cache context.
+ * @param key Key.
+ * @param val Value BEFORE change.
+ * @param expireTime Expire time BEFORE change.
+ * @param ver Version of entry BEFORE change.
*/
- public SnapshotFinishedFutureTask(IgniteCheckedException e) {
- super(null, null, null, null, null, null, null, null);
-
- onDone(e);
- }
-
- /** {@inheritDoc} */
- @Override public boolean start() {
- return false;
- }
-
- /** {@inheritDoc} */
- @Override public void acceptException(Throwable th) {
- onDone(th);
- }
+ public void beforeChange(GridCacheContext cctx, KeyCacheObject key, @Nullable CacheObject val, long expireTime, GridCacheVersion ver);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/DumpEntrySerializer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/DumpEntrySerializer.java
new file mode 100644
index 00000000000..aaf7df24bbd
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/DumpEntrySerializer.java
@@ -0,0 +1,226 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.snapshot.dump;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.processors.cache.CacheObject;
+import org.apache.ignite.internal.processors.cache.CacheObjectContext;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
+import org.apache.ignite.internal.processors.cache.persistence.wal.crc.FastCrc;
+import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Serialization logic for dump.
+ */
+public class DumpEntrySerializer {
+ /** sizeOf(CRC) + sizeOf(Data size) */
+ public static final int HEADER_SZ = Integer.BYTES + Integer.BYTES;
+
+ /** */
+ private ByteBuffer buf;
+
+ /** */
+ private final FastCrc crc = new FastCrc();
+
+ /** Kernal context. */
+ private @Nullable GridKernalContext cctx;
+
+ /** Cache object processor. */
+ private IgniteCacheObjectProcessor co;
+
+ /** */
+ public DumpEntrySerializer() {
+ buf = ByteBuffer.allocate(100);
+ }
+
+ /** */
+ public void kernalContext(GridKernalContext cctx) {
+ this.cctx = cctx;
+ co = cctx.cacheObjects();
+ }
+
+ /**
+ * Dump entry structure:
+ * <pre>
+ * +---------+-----------+----------+-----------------+-----+-------+
+ * | 4 bytes | 4 bytes | 4 bytes | 8 bytes | | |
+ * +---------+-----------+----------+-----------------+-----+-------+
+ * | CRC | Data size | cache ID | expiration time | key | value |
+ * +---------+-----------+----------+-----------------+-----+-------+
+ * </pre>
+ *
+ * @param cache Cache id.
+ * @param expireTime Expire time.
+ * @param key Key.
+ * @param val Value.
+ * @param coCtx Cache object context.
+ * @return Buffer with serialized entry.
+ * @throws IgniteCheckedException If failed
+ */
+ public ByteBuffer writeToBuffer(
+ int cache,
+ long expireTime,
+ KeyCacheObject key,
+ CacheObject val,
+ CacheObjectContext coCtx
+ ) throws IgniteCheckedException {
+ int keySz = key.valueBytesLength(coCtx);
+ int valSz = val.valueBytesLength(coCtx);
+ int dataSz = /*cache ID*/Integer.BYTES + /*expire time*/Long.BYTES + /*key*/keySz + /*value*/valSz;
+
+ int fullSz = dataSz + /*extra bytes for row size*/Integer.BYTES + /*CRC*/Integer.BYTES;
+
+ if (buf.capacity() < fullSz)
+ buf = ByteBuffer.allocate(fullSz);
+ else
+ buf.limit(fullSz);
+
+ buf.position(Integer.BYTES); // CRC value.
+ buf.putInt(dataSz);
+ buf.putInt(cache);
+ buf.putLong(expireTime);
+
+ if (!key.putValue(buf))
+ throw new IgniteCheckedException("Can't write key");
+
+ if (!val.putValue(buf))
+ throw new IgniteCheckedException("Can't write value");
+
+ assert buf.position() == fullSz;
+
+ buf.position(Integer.BYTES);
+
+ crc.reset();
+ crc.update(buf, fullSz - Integer.BYTES);
+
+ buf.position(0);
+ buf.putInt(crc.getValue());
+
+ buf.position(0);
+
+ return buf;
+ }
+
+ /**
+ * @param dumpFile File to read data from.
+ * @param grp Cache group.
+ * @return dump entry.
+ */
+ public DumpEntry read(FileIO dumpFile, int grp, int part) throws IOException, IgniteCheckedException {
+ assert cctx != null : "Set kernalContext first";
+
+ buf.position(0);
+ buf.limit(HEADER_SZ);
+
+ int read = dumpFile.readFully(buf);
+
+ if (read < HEADER_SZ)
+ return null;
+
+ buf.position(0);
+
+ int crc = buf.getInt();
+ int dataSz = buf.getInt();
+
+ if (buf.capacity() < dataSz + HEADER_SZ) {
+ buf = ByteBuffer.allocate(dataSz + HEADER_SZ);
+
+ buf.position(HEADER_SZ - Integer.BYTES);
+ buf.putInt(dataSz); // Required for CRC check.
+ }
+ else
+ buf.limit(dataSz + HEADER_SZ);
+
+ read = dumpFile.readFully(buf);
+
+ if (read != dataSz)
+ throw new IgniteException("Expected to read " + dataSz + " bytes but read only " + read);
+
+ buf.position(HEADER_SZ - Integer.BYTES);
+
+ checkCRC(crc, dataSz, buf);
+
+ buf.position(HEADER_SZ);
+
+ int cache = buf.getInt();
+ long expireTime = buf.getLong();
+
+ int keySz = buf.getInt();
+
+ byte keyType = buf.get();
+
+ byte[] keyBytes = new byte[keySz];
+
+ buf.get(keyBytes, 0, keyBytes.length);
+
+ GridCacheContext<?, ?> cacheCtx = Objects.requireNonNull(
+ cctx.cache().cacheGroup(grp).shared().cacheContext(cache),
+ "Can't find cache context!"
+ );
+
+ CacheObjectContext coCtx = Objects.requireNonNull(cacheCtx.cacheObjectContext(), "Can't find cache object context!");
+
+ KeyCacheObject key = co.toKeyCacheObject(coCtx, keyType, keyBytes);
+
+ if (key.partition() == -1)
+ key.partition(part);
+
+ int valSz = buf.getInt();
+ byte valType = buf.get();
+ byte[] valBytes = new byte[valSz];
+
+ buf.get(valBytes, 0, valBytes.length);
+
+ CacheObject val = co.toCacheObject(coCtx, valType, valBytes);
+
+ return new DumpEntry() {
+ @Override public int cacheId() {
+ return cache;
+ }
+
+ @Override public long expireTime() {
+ return expireTime;
+ }
+
+ @Override public KeyCacheObject key() {
+ return key;
+ }
+
+ @Override public CacheObject value() {
+ return val;
+ }
+ };
+ }
+
+ /** */
+ private void checkCRC(int expCrc, int dataSz, ByteBuffer buf) {
+ crc.reset();
+ crc.update(buf, dataSz + Integer.BYTES /*dataSz field included in CRC calculation*/);
+
+ if (expCrc != crc.getValue())
+ throw new IgniteException("Data corrupted [expCrc=" + expCrc + ", crc=" + crc + ']');
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
index c7985236c1a..d0c3ddd28c6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/pool/PoolProcessor.java
@@ -516,19 +516,19 @@ public class PoolProcessor extends GridProcessorAdapter {
rebalanceExecSvc.allowCoreThreadTimeOut(true);
- if (CU.isPersistenceEnabled(ctx.config())) {
- snpExecSvc = createExecutorService(
- SNAPSHOT_RUNNER_THREAD_PREFIX,
- cfg.getIgniteInstanceName(),
- cfg.getSnapshotThreadPoolSize(),
- cfg.getSnapshotThreadPoolSize(),
- DFLT_THREAD_KEEP_ALIVE_TIME,
- new LinkedBlockingQueue<>(),
- GridIoPolicy.UNDEFINED,
- excHnd);
+ snpExecSvc = createExecutorService(
+ SNAPSHOT_RUNNER_THREAD_PREFIX,
+ cfg.getIgniteInstanceName(),
+ cfg.getSnapshotThreadPoolSize(),
+ cfg.getSnapshotThreadPoolSize(),
+ DFLT_THREAD_KEEP_ALIVE_TIME,
+ new LinkedBlockingQueue<>(),
+ GridIoPolicy.UNDEFINED,
+ excHnd);
- snpExecSvc.allowCoreThreadTimeOut(true);
+ snpExecSvc.allowCoreThreadTimeOut(true);
+ if (CU.isPersistenceEnabled(ctx.config())) {
reencryptExecSvc = createExecutorService(
"reencrypt",
ctx.igniteInstanceName(),
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/SnapshotView.java b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/SnapshotView.java
index a7078204bcd..06f582f4230 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/SnapshotView.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/SnapshotView.java
@@ -63,7 +63,7 @@ public class SnapshotView {
SnapshotMetadata meta,
Collection<String> cacheGrps
) {
- type = SnapshotType.FULL;
+ type = meta.dump() ? SnapshotType.DUMP : SnapshotType.FULL;
name = meta.snapshotName();
consistentId = meta.consistentId();
baselineNodes = F.concat(meta.baselineNodes(), ",");
@@ -149,6 +149,9 @@ public class SnapshotView {
FULL,
/** Incremental snapshot. */
- INCREMENTAL
+ INCREMENTAL,
+
+ /** Dump. */
+ DUMP
}
}
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 802497a2d72..7890402f2ec 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
@@ -824,7 +824,7 @@ public abstract class AbstractSnapshotSelfTest extends GridCommonAbstractTest {
SnapshotSender snpSndr
) throws IgniteCheckedException {
AbstractSnapshotFutureTask<?> task = cctx.snapshotMgr().registerSnapshotTask(snpName, cctx.localNodeId(), null,
- parts, withMetaStorage, snpSndr);
+ parts, withMetaStorage, false, snpSndr);
if (!(task instanceof SnapshotFutureTask))
throw new IgniteCheckedException("Snapshot task hasn't been registered: " + task);
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 d16d6074f52..d1d51093c21 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
@@ -291,7 +291,7 @@ public class EncryptedSnapshotTest extends AbstractSnapshotSelfTest {
GridTestUtils.assertThrowsAnyCause(log,
() -> snp(ig).registerSnapshotTask(SNAPSHOT_NAME, ig.localNode().id(),
- null, F.asMap(CU.cacheId(dfltCacheCfg.getName()), null), false,
+ null, F.asMap(CU.cacheId(dfltCacheCfg.getName()), null), false, false,
snp(ig).localSnapshotSenderFactory().apply(SNAPSHOT_NAME, null)).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/IgniteSnapshotManagerSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
index 79b3699b6b0..478cbb750e2 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
@@ -148,6 +148,7 @@ public class IgniteSnapshotManagerSelfTest extends AbstractSnapshotSelfTest {
null,
F.asMap(CU.cacheId(DEFAULT_CACHE_NAME), null),
encryption,
+ false,
new DelegateSnapshotSender(log, mgr.snapshotExecutorService(), mgr.localSnapshotSenderFactory().apply(SNAPSHOT_NAME, null)) {
@Override public void sendPart0(File part, String cacheDirName, GroupPartitionId pair, Long length) {
try {
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/AbstractCacheDumpTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/AbstractCacheDumpTest.java
new file mode 100644
index 00000000000..f2cabfa3250
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/AbstractCacheDumpTest.java
@@ -0,0 +1,434 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.snapshot.dump;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.function.IntConsumer;
+import java.util.function.IntFunction;
+import java.util.stream.IntStream;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterState;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.processors.cache.CacheObjectContext;
+import org.apache.ignite.internal.processors.cache.StoredCacheData;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadata;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.Dump.DumpedPartitionIterator;
+import org.apache.ignite.internal.util.typedef.T2;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.platform.model.ACL;
+import org.apache.ignite.platform.model.Key;
+import org.apache.ignite.platform.model.Role;
+import org.apache.ignite.platform.model.User;
+import org.apache.ignite.platform.model.Value;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.internal.processors.cache.GridCacheUtils.UTILITY_CACHE_NAME;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNP_RUNNING_DIR_KEY;
+import static org.apache.ignite.platform.model.AccessLevel.SUPER;
+import static org.apache.ignite.testframework.GridTestUtils.runAsync;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/** */
+@RunWith(Parameterized.class)
+public abstract class AbstractCacheDumpTest extends GridCommonAbstractTest {
+ /** */
+ public static final String GRP = "grp";
+
+ /** */
+ public static final String CACHE_0 = "cache-0";
+
+ /** */
+ public static final String CACHE_1 = "cache-1";
+
+ /** */
+ public static final int KEYS_CNT = 1000;
+
+ /** */
+ public static final String DMP_NAME = "dump";
+
+ /** */
+ protected static final IntFunction<User> USER_FACTORY = i ->
+ new User(i, ACL.values()[Math.abs(i) % ACL.values().length], new Role("Role" + i, SUPER));
+
+ /** */
+ @Parameterized.Parameter
+ public int nodes;
+
+ /** */
+ @Parameterized.Parameter(1)
+ public int backups;
+
+ /** */
+ @Parameterized.Parameter(2)
+ public boolean persistence;
+
+ /** */
+ @Parameterized.Parameter(3)
+ public CacheAtomicityMode mode;
+
+ /** */
+ @Parameterized.Parameters(name = "nodes={0},backups={1},persistence={2},mode={3}")
+ public static List<Object[]> params() {
+ List<Object[]> params = new ArrayList<>();
+
+ for (int nodes : new int[]{1, 3})
+ for (int backups : new int[]{0, 1})
+ for (boolean persistence : new boolean[]{true, false})
+ for (CacheAtomicityMode mode : CacheAtomicityMode._values()) {
+ if (nodes == 1 && backups != 0)
+ continue;
+
+ params.add(new Object[]{nodes, backups, persistence, mode});
+ }
+
+ return params;
+ }
+
+ /** */
+ protected int snpPoolSz = 1;
+
+ /** */
+ protected IgniteEx cli;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setSnapshotThreadPoolSize(snpPoolSz)
+ .setDataStorageConfiguration(new DataStorageConfiguration()
+ .setDefaultDataRegionConfiguration(new DataRegionConfiguration().setPersistenceEnabled(persistence)))
+ .setCacheConfiguration(
+ new CacheConfiguration<>()
+ .setName(DEFAULT_CACHE_NAME)
+ .setBackups(backups)
+ .setAtomicityMode(mode)
+ .setWriteSynchronizationMode(FULL_SYNC)
+ .setAffinity(new RendezvousAffinityFunction().setPartitions(20)),
+ new CacheConfiguration<>()
+ .setGroupName(GRP)
+ .setName(CACHE_0)
+ .setBackups(backups)
+ .setAtomicityMode(mode)
+ .setWriteSynchronizationMode(FULL_SYNC)
+ .setAffinity(new RendezvousAffinityFunction().setPartitions(20)),
+ new CacheConfiguration<>()
+ .setGroupName(GRP)
+ .setName(CACHE_1)
+ .setBackups(backups)
+ .setAtomicityMode(mode)
+ .setWriteSynchronizationMode(FULL_SYNC)
+ .setAffinity(new RendezvousAffinityFunction().setPartitions(20))
+ );
+ }
+
+ /** */
+ protected IgniteEx startGridAndFillCaches() throws Exception {
+ IgniteEx ign = (IgniteEx)startGridsMultiThreaded(nodes);
+
+ cli = startClientGrid(nodes);
+
+ ign.cluster().state(ClusterState.ACTIVE);
+
+ putData(ign.cache(DEFAULT_CACHE_NAME), ign.cache(CACHE_0), ign.cache(CACHE_1));
+
+ return ign;
+ }
+
+ /** */
+ protected T2<CountDownLatch, IgniteInternalFuture<?>> runDumpAsyncAndStopBeforeStart() throws IgniteInterruptedCheckedException {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ List<Ignite> ignites = Ignition.allGrids();
+
+ for (Ignite ign : ignites) {
+ ((IgniteEx)ign).context().pools().getSnapshotExecutorService().submit(() -> {
+ try {
+ latch.await();
+ }
+ catch (InterruptedException e) {
+ throw new IgniteException(e);
+ }
+ });
+ }
+
+ IgniteInternalFuture<Object> dumpFut = runAsync(() -> createDump((IgniteEx)ignites.get(0)));
+
+ // Waiting while dump will be setup: task planned after change listener set.
+ assertTrue(waitForCondition(() -> {
+ for (Ignite ign : ignites) {
+ if (ign.configuration().isClientMode() == Boolean.TRUE)
+ continue;
+
+ if (((ThreadPoolExecutor)((IgniteEx)ign).context().pools().getSnapshotExecutorService()).getTaskCount() <= 1)
+ return false;
+ }
+
+ return true;
+ }, 10 * 1000));
+
+ return new T2<>(latch, dumpFut);
+ }
+
+ /** */
+ protected void putData(
+ IgniteCache<Object, Object> cache,
+ IgniteCache<Object, Object> grpCache0,
+ IgniteCache<Object, Object> grpCache1
+ ) {
+ IntStream.range(0, KEYS_CNT).forEach(i -> {
+ cache.put(i, i);
+ grpCache0.put(i, USER_FACTORY.apply(i));
+ grpCache1.put(new Key(i), new Value(String.valueOf(i)));
+ });
+ }
+
+ /** */
+ void checkDump(IgniteEx ign) throws Exception {
+ checkDump(ign, DMP_NAME);
+ }
+
+ /** */
+ void checkDump(IgniteEx ign, String name) throws Exception {
+ if (persistence)
+ assertNull(ign.context().cache().context().database().metaStorage().read(SNP_RUNNING_DIR_KEY));
+
+ Dump dump = dump(ign, name);
+
+ List<SnapshotMetadata> metadata = dump.metadata();
+
+ assertNotNull(metadata);
+ assertEquals(nodes, metadata.size());
+
+ for (SnapshotMetadata meta : metadata) {
+ assertEquals(name, meta.snapshotName());
+ assertTrue(meta.dump());
+ assertFalse(meta.cacheGroupIds().contains(CU.cacheId(UTILITY_CACHE_NAME)));
+ }
+
+ List<String> nodesDirs = dump.nodesDirectories();
+
+ assertEquals(nodes, nodesDirs.size());
+
+ Set<Integer> keys = new HashSet<>();
+ int dfltDumpSz = 0;
+ int grpDumpSz = 0;
+
+ CacheObjectContext coCtx = ign.context().cache().context().cacheObjectContext(CU.cacheId(DEFAULT_CACHE_NAME));
+ CacheObjectContext coCtx0 = ign.context().cache().context().cacheObjectContext(CU.cacheId(CACHE_0));
+ CacheObjectContext coCtx1 = ign.context().cache().context().cacheObjectContext(CU.cacheId(CACHE_1));
+
+ for (String nodeDir : nodesDirs) {
+ List<StoredCacheData> ccfgs = dump.configs(nodeDir, CU.cacheId(DEFAULT_CACHE_NAME));
+
+ assertNotNull(ccfgs);
+ assertEquals(1, ccfgs.size());
+
+ assertEquals(DEFAULT_CACHE_NAME, ccfgs.get(0).configuration().getName());
+ assertFalse(ccfgs.get(0).sql());
+ assertTrue(ccfgs.get(0).queryEntities().isEmpty());
+
+ ccfgs = dump.configs(nodeDir, CU.cacheId(GRP));
+
+ assertNotNull(ccfgs);
+ assertEquals(2, ccfgs.size());
+
+ ccfgs.sort(Comparator.comparing(d -> d.config().getName()));
+
+ CacheConfiguration ccfg0 = ccfgs.get(0).configuration();
+ CacheConfiguration ccfg1 = ccfgs.get(1).configuration();
+
+ assertEquals(GRP, ccfg0.getGroupName());
+ assertEquals(CACHE_0, ccfg0.getName());
+
+ assertEquals(GRP, ccfg1.getGroupName());
+ assertEquals(CACHE_1, ccfg1.getName());
+
+ assertFalse(ccfgs.get(0).sql());
+ assertFalse(ccfgs.get(1).sql());
+ assertTrue(ccfgs.get(0).queryEntities().isEmpty());
+ assertTrue(ccfgs.get(1).queryEntities().isEmpty());
+
+ List<Integer> parts = dump.partitions(nodeDir, CU.cacheId(DEFAULT_CACHE_NAME));
+
+ for (int part : parts) {
+ try (DumpedPartitionIterator iter = dump.iterator(nodeDir, CU.cacheId(DEFAULT_CACHE_NAME), part)) {
+ while (iter.hasNext()) {
+ DumpEntry e = iter.next();
+
+ checkDefaultCacheEntry(e, coCtx);
+
+ keys.add(e.key().<Integer>value(coCtx, true));
+
+ dfltDumpSz++;
+ }
+ }
+ }
+
+ parts = dump.partitions(nodeDir, CU.cacheId(GRP));
+
+ for (int part : parts) {
+ try (DumpedPartitionIterator iter = dump.iterator(nodeDir, CU.cacheId(GRP), part)) {
+ while (iter.hasNext()) {
+ DumpEntry e = iter.next();
+
+ checkGroupEntry(e, coCtx0, coCtx1);
+
+ grpDumpSz++;
+ }
+ }
+ }
+ }
+
+ assertEquals(KEYS_CNT + KEYS_CNT * backups, dfltDumpSz);
+ assertEquals(2 * (KEYS_CNT + KEYS_CNT * backups), grpDumpSz);
+
+ IntStream.range(0, KEYS_CNT).forEach(key -> assertTrue(keys.contains(key)));
+ }
+
+ /** */
+ protected void checkDefaultCacheEntry(DumpEntry e, CacheObjectContext coCtx) {
+ assertNotNull(e);
+
+ Integer key = e.key().<Integer>value(coCtx, true);
+
+ assertEquals(key, e.value().<Integer>value(coCtx, true));
+ }
+
+ /** */
+ protected void checkGroupEntry(DumpEntry e, CacheObjectContext coCtx0, CacheObjectContext coCtx1) {
+ assertNotNull(e);
+
+ if (e.cacheId() == CU.cacheId(CACHE_0))
+ assertEquals(USER_FACTORY.apply(e.key().value(coCtx0, true)), e.value().value(coCtx0, true));
+ else {
+ assertNotNull(e.key().<Key>value(coCtx1, true));
+ assertNotNull(e.value().<Value>value(coCtx1, true));
+ }
+ }
+
+ /** */
+ protected void insertOrUpdate(IgniteEx ignite, int i) {
+ insertOrUpdate(ignite, i, i);
+ }
+
+ /** */
+ protected void insertOrUpdate(IgniteEx ignite, int i, int val) {
+ ignite.cache(DEFAULT_CACHE_NAME).put(i, val);
+ IgniteCache<Object, Object> cache = ignite.cache(CACHE_0);
+ IgniteCache<Object, Object> cache1 = ignite.cache(CACHE_1);
+
+ if (mode == CacheAtomicityMode.TRANSACTIONAL) {
+ try (Transaction tx = ignite.transactions().txStart()) {
+ cache.put(i, USER_FACTORY.apply(val));
+
+ tx.commit();
+ }
+
+ try (Transaction tx = ignite.transactions().txStart()) {
+ cache1.put(new Key(i), new Value(String.valueOf(val)));
+
+ tx.commit();
+ }
+ }
+ else {
+ cache.put(i, USER_FACTORY.apply(val));
+
+ cache1.put(new Key(i), new Value(String.valueOf(val)));
+ }
+ }
+
+ /** */
+ protected void remove(IgniteEx ignite, int i) {
+ ignite.cache(DEFAULT_CACHE_NAME).remove(i);
+
+ IgniteCache<Object, Object> cache = ignite.cache(CACHE_0);
+ IgniteCache<Object, Object> cache1 = ignite.cache(CACHE_1);
+
+ IntConsumer moreRemovals = j -> {
+ cache.remove(j);
+ cache.remove(j);
+ cache1.remove(new Key(j));
+ cache1.remove(new Key(j));
+ };
+
+ if (mode == CacheAtomicityMode.TRANSACTIONAL) {
+ try (Transaction tx = ignite.transactions().txStart()) {
+ moreRemovals.accept(i);
+
+ tx.commit();
+ }
+ }
+ else
+ moreRemovals.accept(i);
+ }
+
+ /** */
+ void createDump(IgniteEx ign) {
+ createDump(ign, DMP_NAME);
+ }
+
+ /** */
+ public static Dump dump(IgniteEx ign, String name) throws IgniteCheckedException {
+ return new Dump(
+ ign.context(),
+ new File(U.resolveWorkDirectory(U.defaultWorkDirectory(), ign.configuration().getSnapshotPath(), false), name)
+ );
+ }
+
+ /** */
+ void createDump(IgniteEx ign, String name) {
+ ign.snapshot().createDump(name).get();
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/IgniteCacheDumpSelf2Test.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/IgniteCacheDumpSelf2Test.java
new file mode 100644
index 00000000000..c15edfb3e96
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/IgniteCacheDumpSelf2Test.java
@@ -0,0 +1,212 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.snapshot.dump;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterState;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+import static java.nio.file.StandardOpenOption.READ;
+import static java.nio.file.StandardOpenOption.WRITE;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DATA_FILENAME;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DIR_PREFIX;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.PART_FILE_PREFIX;
+import static org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderResolver.DB_DEFAULT_FOLDER;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.DUMP_LOCK;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.AbstractCacheDumpTest.DMP_NAME;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.AbstractCacheDumpTest.KEYS_CNT;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.AbstractCacheDumpTest.dump;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.CreateDumpFutureTask.DUMP_FILE_EXT;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.DumpEntrySerializer.HEADER_SZ;
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+
+/** */
+public class IgniteCacheDumpSelf2Test extends GridCommonAbstractTest {
+ /** */
+ private LogListener lsnr;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ if (lsnr != null) {
+ ListeningTestLogger testLog = new ListeningTestLogger(log);
+
+ testLog.registerListener(lsnr);
+
+ cfg.setGridLogger(testLog);
+ }
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ stopAllGrids();
+
+ cleanPersistenceDir();
+ }
+
+ /** */
+ @Test
+ public void testDumpFailIfNoCaches() throws Exception {
+ try (IgniteEx ign = startGrid(new IgniteConfiguration())) {
+ ign.cluster().state(ClusterState.ACTIVE);
+
+ assertThrows(
+ null,
+ () -> ign.snapshot().createDump("dump").get(),
+ IgniteException.class,
+ "Dump operation has been rejected. No cache group defined in cluster"
+ );
+ }
+ }
+
+ /** */
+ @Test
+ public void testUnreadyDumpCleared() throws Exception {
+ IgniteEx ign = (IgniteEx)startGridsMultiThreaded(2);
+
+ ign.cluster().state(ClusterState.ACTIVE);
+
+ IgniteCache<Integer, Integer> cache = ign.createCache(DEFAULT_CACHE_NAME);
+
+ IntStream.range(0, KEYS_CNT).forEach(i -> cache.put(i, i));
+
+ ign.snapshot().createDump(DMP_NAME).get(getTestTimeout());
+
+ stopAllGrids();
+
+ Dump dump = dump(ign, DMP_NAME);
+
+ List<String> nodes = dump.nodesDirectories();
+
+ assertNotNull(nodes);
+ assertEquals(2, nodes.size());
+
+ File nodeDumpDir = new File(dump.dumpDirectory(), DB_DEFAULT_FOLDER + File.separator + nodes.get(0));
+
+ assertTrue(new File(nodeDumpDir, DUMP_LOCK).createNewFile());
+
+ lsnr = LogListener.matches("Found locked dump dir. " +
+ "This means, dump creation not finished prior to node fail. " +
+ "Directory will be deleted: " + nodeDumpDir.getAbsolutePath()).build();
+
+ startGridsMultiThreaded(2);
+
+ assertFalse(nodeDumpDir.exists());
+ assertTrue(lsnr.check());
+ }
+
+ /** */
+ @Test
+ public void testDumpIteratorFaileOnWrongCrc() throws Exception {
+ try (IgniteEx ign = startGrid(new IgniteConfiguration())) {
+ ign.cluster().state(ClusterState.ACTIVE);
+
+ IgniteCache<Integer, Integer> cache = ign.createCache(DEFAULT_CACHE_NAME);
+
+ for (int key : partitionKeys(cache, 0, KEYS_CNT, 0))
+ cache.put(key, key);
+
+ ign.snapshot().createDump(DMP_NAME).get();
+
+ Dump dump = dump(ign, DMP_NAME);
+
+ List<String> nodes = dump.nodesDirectories();
+
+ assertNotNull(nodes);
+ assertEquals(1, nodes.size());
+
+ File cacheDumpDir = new File(
+ dump.dumpDirectory(),
+ DB_DEFAULT_FOLDER + File.separator + nodes.get(0) + File.separator + CACHE_DIR_PREFIX + DEFAULT_CACHE_NAME
+ );
+
+ assertTrue(cacheDumpDir.exists());
+
+ Set<File> dumpFiles = Arrays.asList(cacheDumpDir.listFiles()).stream()
+ .filter(f -> {
+ try {
+ return Files.size(f.toPath()) > 0;
+ }
+ catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .collect(Collectors.toSet());
+
+ String partDumpName = PART_FILE_PREFIX + 0 + DUMP_FILE_EXT;
+
+ assertTrue(dumpFiles.stream().anyMatch(f -> f.getName().equals(CACHE_DATA_FILENAME)));
+ assertTrue(dumpFiles.stream().anyMatch(f -> f.getName().equals(partDumpName)));
+
+ try (FileChannel fc = FileChannel.open(Paths.get(cacheDumpDir.getAbsolutePath(), partDumpName), READ, WRITE)) {
+ fc.position(HEADER_SZ); // Skip first entry header.
+
+ int bufSz = 5;
+
+ ByteBuffer buf = ByteBuffer.allocate(bufSz);
+
+ assertEquals(bufSz, fc.read(buf));
+
+ buf.position(0);
+
+ // Increment first five bytes in dumped entry.
+ for (int i = 0; i < bufSz; i++) {
+ byte b = buf.get();
+ b++;
+ buf.position(i);
+ buf.put(b);
+ }
+
+ fc.position(HEADER_SZ);
+
+ buf.rewind();
+ fc.write(buf);
+ }
+
+ assertThrows(
+ null,
+ () -> dump.iterator(nodes.get(0), CU.cacheId(DEFAULT_CACHE_NAME), 0).next(),
+ IgniteException.class,
+ "Data corrupted"
+ );
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/IgniteCacheDumpSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/IgniteCacheDumpSelfTest.java
new file mode 100644
index 00000000000..893d94f77d9
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/dump/IgniteCacheDumpSelfTest.java
@@ -0,0 +1,503 @@
+/*
+ * 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.ignite.internal.processors.cache.persistence.snapshot.dump;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.OpenOption;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.stream.IntStream;
+import javax.cache.expiry.Duration;
+import javax.cache.expiry.ExpiryPolicy;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CacheEntryProcessor;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.cache.CacheObjectContext;
+import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
+import org.apache.ignite.internal.processors.cache.persistence.file.FileIODecorator;
+import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.T2;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.platform.model.Key;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.junit.Test;
+
+import static java.lang.Boolean.FALSE;
+import static java.lang.Boolean.TRUE;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNP_RUNNING_DIR_KEY;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.CreateDumpFutureTask.DUMP_FILE_EXT;
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+
+/** */
+public class IgniteCacheDumpSelfTest extends AbstractCacheDumpTest {
+ /** */
+ public static final String EXISTS_ERR_MSG = "Create snapshot request has been rejected. " +
+ "Snapshot with given name already exists on local node";
+
+ /** */
+ public static final long TTL = 5 * 1000;
+
+ /** */
+ public static final ExpiryPolicy EXPIRY_POLICY = new ExpiryPolicy() {
+ @Override public Duration getExpiryForCreation() {
+ return new Duration(MILLISECONDS, TTL);
+ }
+
+ @Override public Duration getExpiryForAccess() {
+ return null;
+ }
+
+ @Override public Duration getExpiryForUpdate() {
+ return null;
+ }
+ };
+
+ /** */
+ private Boolean explicitTtl;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ explicitTtl = null;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ if (explicitTtl == TRUE) {
+ for (CacheConfiguration<?, ?> ccfg : cfg.getCacheConfiguration())
+ ccfg.setExpiryPolicyFactory(() -> EXPIRY_POLICY);
+ }
+
+ return cfg;
+ }
+
+ /** */
+ @Test
+ public void testCacheDump() throws Exception {
+ snpPoolSz = 4;
+
+ try {
+ IgniteEx ign = startGridAndFillCaches();
+
+ createDump(ign);
+
+ checkDump(ign);
+
+ assertThrows(null, () -> createDump(ign), IgniteException.class, EXISTS_ERR_MSG);
+
+ createDump(ign, DMP_NAME + 2);
+
+ checkDump(ign, DMP_NAME + 2);
+
+ if (persistence) {
+ assertThrows(null, () -> ign.snapshot().createSnapshot(DMP_NAME).get(), IgniteException.class, EXISTS_ERR_MSG);
+
+ ign.snapshot().createSnapshot(DMP_NAME + 3).get();
+ }
+ else {
+ assertThrows(
+ null,
+ () -> ign.snapshot().createSnapshot(DMP_NAME + 3).get(),
+ IgniteException.class,
+ "Create snapshot request has been rejected. Snapshots on an in-memory clusters are not allowed."
+ );
+ }
+ }
+ finally {
+ snpPoolSz = 1;
+ }
+ }
+
+ /** */
+ @Test
+ public void testConcurrentDumpCreationThrows() throws Exception {
+ doTestConcurrentOperations(ignite -> {
+ assertThrows(
+ null,
+ () -> createDump(ignite, "other_dump"),
+ IgniteException.class,
+ "Create snapshot request has been rejected. The previous snapshot operation was not completed."
+ );
+
+ if (persistence) {
+ assertThrows(
+ null,
+ () -> ignite.snapshot().createSnapshot("other_snapshot").get(getTestTimeout()),
+ IgniteException.class,
+ "Create snapshot request has been rejected. The previous snapshot operation was not completed."
+ );
+ }
+ });
+ }
+
+ /** */
+ @Test
+ public void testWithConcurrentInserts() throws Exception {
+ doTestConcurrentOperations(ignite -> {
+ for (int i = KEYS_CNT * 10; i < KEYS_CNT * 10 + 3; i++) {
+ assertFalse(ignite.cache(DEFAULT_CACHE_NAME).containsKey(i));
+ assertFalse(ignite.cache(CACHE_0).containsKey(i));
+ assertFalse(ignite.cache(CACHE_1).containsKey(new Key(i)));
+
+ insertOrUpdate(ignite, i);
+ }
+
+ for (int i = KEYS_CNT * 10 + 3; i < KEYS_CNT * 10 + 6; i++) {
+ assertFalse(cli.cache(DEFAULT_CACHE_NAME).containsKey(i));
+ assertFalse(cli.cache(CACHE_0).containsKey(i));
+ assertFalse(cli.cache(CACHE_1).containsKey(new Key(i)));
+
+ insertOrUpdate(cli, i);
+ }
+ });
+ }
+
+ /** */
+ @Test
+ public void testWithConcurrentUpdates() throws Exception {
+ doTestConcurrentOperations(ignite -> {
+ for (int i = 0; i < 3; i++) {
+ assertTrue(ignite.cache(DEFAULT_CACHE_NAME).containsKey(i));
+ assertTrue(ignite.cache(CACHE_0).containsKey(i));
+ assertTrue(ignite.cache(CACHE_1).containsKey(new Key(i)));
+
+ insertOrUpdate(ignite, i);
+ }
+
+ for (int i = 3; i < 6; i++) {
+ assertTrue(cli.cache(DEFAULT_CACHE_NAME).containsKey(i));
+ assertTrue(cli.cache(CACHE_0).containsKey(i));
+ assertTrue(cli.cache(CACHE_1).containsKey(new Key(i)));
+
+ insertOrUpdate(cli, i);
+ }
+ });
+ }
+
+ /** */
+ @Test
+ public void testWithConcurrentEntryProcessor() throws Exception {
+ doTestConcurrentOperations(ignite -> {
+ for (int i = KEYS_CNT * 10; i < KEYS_CNT * 10 + 3; i++) {
+ assertFalse(ignite.cache(DEFAULT_CACHE_NAME).containsKey(i));
+
+ ignite.<Integer, Integer>cache(DEFAULT_CACHE_NAME).invoke(i, new CacheEntryProcessor<Integer, Integer, Void>() {
+ @Override public Void process(MutableEntry<Integer, Integer> entry, Object... arguments) {
+ entry.setValue(entry.getKey());
+ return null;
+ }
+ });
+ }
+
+ for (int i = 0; i < 3; i++) {
+ assertTrue(ignite.cache(DEFAULT_CACHE_NAME).containsKey(i));
+
+ ignite.<Integer, Integer>cache(DEFAULT_CACHE_NAME).invoke(i, new CacheEntryProcessor<Integer, Integer, Void>() {
+ @Override public Void process(MutableEntry<Integer, Integer> entry, Object... arguments) {
+ entry.setValue(entry.getKey() + 1);
+ return null;
+ }
+ });
+ }
+ });
+ }
+
+ /** */
+ @Test
+ public void testWithConcurrentRemovals() throws Exception {
+ doTestConcurrentOperations(ignite -> {
+ for (int i = 0; i < 3; i++) {
+ assertTrue(ignite.cache(DEFAULT_CACHE_NAME).containsKey(i));
+
+ remove(ignite, i);
+ }
+
+ for (int i = 3; i < 6; i++) {
+ assertTrue(ignite.cache(DEFAULT_CACHE_NAME).containsKey(i));
+
+ remove(cli, i);
+ }
+ });
+ }
+
+ /** */
+ @Test
+ public void testDumpWithExplicitExpireTime() throws Exception {
+ explicitTtl = true;
+
+ doTestDumpWithExpiry();
+ }
+
+ /** */
+ @Test
+ public void testDumpWithImplicitExpireTime() throws Exception {
+ explicitTtl = false;
+
+ doTestDumpWithExpiry();
+ }
+
+ /** */
+ @Test
+ public void testDumpCancelOnFileCreateError() throws Exception {
+ IgniteEx ign = startGridAndFillCaches();
+
+ for (Ignite node : G.allGrids()) {
+ if (node.configuration().isClientMode() == TRUE)
+ continue;
+
+ ((IgniteEx)node).context().cache().context().snapshotMgr().ioFactory(new DumpFailingFactory((IgniteEx)node, false));
+ }
+
+ assertThrows(null, () -> ign.snapshot().createDump(DMP_NAME).get(), IgniteException.class, "Test error");
+
+ checkDumpCleared(ign);
+
+ ign.cache(DEFAULT_CACHE_NAME).put(KEYS_CNT, KEYS_CNT);
+ }
+
+ /** */
+ @Test
+ public void testDumpCancelOnIteratorWriteError() throws Exception {
+ IgniteEx ign = startGridAndFillCaches();
+
+ DumpFailingFactory ioFactory = new DumpFailingFactory(ign, true);
+
+ ign.context().cache().context().snapshotMgr().ioFactory(ioFactory);
+
+ assertThrows(null, () -> ign.snapshot().createDump(DMP_NAME).get(), IgniteException.class, "Test write error");
+
+ assertTrue(ioFactory.errorAfter.get() <= 0);
+
+ checkDumpCleared(ign);
+
+ ign.cache(DEFAULT_CACHE_NAME).put(KEYS_CNT, KEYS_CNT);
+ }
+
+ /** */
+ @Test
+ public void testDumpCancelOnListenerWriteError() throws Exception {
+ IgniteEx ign = startGridAndFillCaches();
+
+ IgniteCache<Object, Object> cache = ign.cache(DEFAULT_CACHE_NAME);
+
+ int keyToFail = findKeys(ign.localNode(), cache, 1, KEYS_CNT, 0).get(0);
+
+ byte[] valToFail = new byte[100];
+
+ AtomicBoolean keyToFailFound = new AtomicBoolean();
+
+ IntStream.range(0, 100).forEach(i -> valToFail[i] = (byte)i);
+
+ cache.put(keyToFail, valToFail);
+
+ FileIOFactory delegate = ign.context().cache().context().snapshotMgr().ioFactory();
+
+ ign.context().cache().context().snapshotMgr().ioFactory(new FileIOFactory() {
+ @Override public FileIO create(File file, OpenOption... modes) throws IOException {
+ if (file.getName().endsWith(DUMP_FILE_EXT)) {
+ return new FileIODecorator(delegate.create(file, modes)) {
+ /** {@inheritDoc} */
+ @Override public int writeFully(ByteBuffer srcBuf) throws IOException {
+ if (findValToFail(srcBuf)) {
+ keyToFailFound.set(true);
+
+ throw new IOException("Val to fail found");
+ }
+
+ return super.writeFully(srcBuf);
+ }
+
+ private boolean findValToFail(ByteBuffer srcBuf) {
+ if (srcBuf.limit() >= valToFail.length) {
+ int valIdx = 0;
+
+ for (int i = 0; i < srcBuf.limit(); i++) {
+ if (srcBuf.get(i) == valToFail[valIdx]) {
+ valIdx++;
+
+ if (valIdx == valToFail.length)
+ return true;
+ }
+ else
+ valIdx = 0;
+ }
+ }
+
+ return false;
+ }
+ };
+ }
+
+ return delegate.create(file, modes);
+ }
+ });
+
+ T2<CountDownLatch, IgniteInternalFuture<?>> latchAndFut = runDumpAsyncAndStopBeforeStart();
+
+ cache.put(keyToFail, "test string");
+
+ latchAndFut.get1().countDown();
+
+ assertThrows(null, () -> latchAndFut.get2().get(10 * 1000), IgniteCheckedException.class, "Val to fail found");
+
+ assertTrue(keyToFailFound.get());
+
+ checkDumpCleared(ign);
+
+ cache.put(keyToFail, valToFail);
+ }
+
+ /** */
+ private void checkDumpCleared(IgniteEx ign) throws IgniteCheckedException {
+ if (persistence)
+ assertNull(ign.context().cache().context().database().metaStorage().read(SNP_RUNNING_DIR_KEY));
+
+ assertFalse(
+ new File(U.resolveWorkDirectory(U.defaultWorkDirectory(), ign.configuration().getSnapshotPath(), false), DMP_NAME).exists()
+ );
+ }
+
+ /** */
+ private void doTestDumpWithExpiry() throws Exception {
+ IgniteEx ign = startGridAndFillCaches();
+
+ T2<CountDownLatch, IgniteInternalFuture<?>> latchAndFut = runDumpAsyncAndStopBeforeStart();
+
+ Thread.sleep(TTL);
+
+ assertTrue(GridTestUtils.waitForCondition(() -> {
+ for (int i = 0; i < KEYS_CNT; i++) {
+ if (ign.cache(DEFAULT_CACHE_NAME).containsKey(i))
+ return false;
+
+ if (ign.cache(CACHE_0).containsKey(i))
+ return false;
+
+ if (ign.cache(CACHE_1).containsKey(new Key(i)))
+ return false;
+ }
+
+ return true;
+ }, 2 * TTL));
+
+ latchAndFut.get1().countDown();
+
+ latchAndFut.get2().get();
+
+ checkDump(ign);
+ }
+
+ /** */
+ private void doTestConcurrentOperations(Consumer<IgniteEx> op) throws Exception {
+ IgniteEx ign = startGridAndFillCaches();
+
+ T2<CountDownLatch, IgniteInternalFuture<?>> latchAndFut = runDumpAsyncAndStopBeforeStart();
+
+ // This operations will be catched by change listeners. Old value must be stored in dump.
+ op.accept(ign);
+
+ latchAndFut.get1().countDown();
+
+ latchAndFut.get2().get(10 * 1000);
+
+ checkDump(ign);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void putData(
+ IgniteCache<Object, Object> cache,
+ IgniteCache<Object, Object> grpCache0,
+ IgniteCache<Object, Object> grpCache1
+ ) {
+ if (explicitTtl == FALSE) {
+ super.putData(
+ cache.withExpiryPolicy(EXPIRY_POLICY),
+ grpCache0.withExpiryPolicy(EXPIRY_POLICY),
+ grpCache1.withExpiryPolicy(EXPIRY_POLICY)
+ );
+ }
+ else
+ super.putData(cache, grpCache0, grpCache1);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void checkDefaultCacheEntry(DumpEntry e, CacheObjectContext coCtx) {
+ super.checkDefaultCacheEntry(e, coCtx);
+
+ if (explicitTtl != null) {
+ assertTrue("Expire time must be set", e.expireTime() != 0);
+ assertTrue("Expire time must be in past", System.currentTimeMillis() >= e.expireTime());
+ assertTrue("Expire time must be set during test run", System.currentTimeMillis() - getTestTimeout() < e.expireTime());
+ }
+ }
+
+ /** */
+ public static class DumpFailingFactory implements FileIOFactory {
+ /** */
+ private final FileIOFactory delegate;
+
+ /** */
+ private final AtomicInteger errorAfter;
+
+ /** */
+ private final boolean failOnWrite;
+
+ /** */
+ public DumpFailingFactory(IgniteEx ign, boolean failOnWrite) {
+ this.delegate = ign.context().cache().context().snapshotMgr().ioFactory();
+ this.failOnWrite = failOnWrite;
+ this.errorAfter = new AtomicInteger(KEYS_CNT / 20);
+ }
+
+ /** {@inheritDoc} */
+ @Override public FileIO create(File file, OpenOption... modes) throws IOException {
+ if (failOnWrite) {
+ return new FileIODecorator(delegate.create(file, modes)) {
+ /** {@inheritDoc} */
+ @Override public int writeFully(ByteBuffer srcBuf) throws IOException {
+ if (errorAfter.decrementAndGet() > 0)
+ return super.writeFully(srcBuf);
+
+ throw new IOException("Test write error");
+ }
+ };
+ }
+ else if (file.getName().endsWith(DUMP_FILE_EXT)) {
+ throw new IOException("Test error");
+ }
+
+ return delegate.create(file, modes);
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/platform/model/Key.java b/modules/core/src/test/java/org/apache/ignite/platform/model/Key.java
index 218d9c0b1e1..70caefc1615 100644
--- a/modules/core/src/test/java/org/apache/ignite/platform/model/Key.java
+++ b/modules/core/src/test/java/org/apache/ignite/platform/model/Key.java
@@ -48,4 +48,9 @@ public class Key {
@Override public int hashCode() {
return Objects.hash(id);
}
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "Key[id=" + id + ']';
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/platform/model/Role.java b/modules/core/src/test/java/org/apache/ignite/platform/model/Role.java
index ad8b3d3f908..a39fc2471ec 100644
--- a/modules/core/src/test/java/org/apache/ignite/platform/model/Role.java
+++ b/modules/core/src/test/java/org/apache/ignite/platform/model/Role.java
@@ -17,6 +17,8 @@
package org.apache.ignite.platform.model;
+import java.util.Objects;
+
/** Test value object. */
public class Role {
/** */
@@ -50,4 +52,17 @@ public class Role {
public void setAccessLevel(AccessLevel accessLevel) {
this.accessLevel = accessLevel;
}
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ Role role = (Role)o;
+ return Objects.equals(name, role.name) && accessLevel == role.accessLevel;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Objects.hash(name, accessLevel);
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/platform/model/User.java b/modules/core/src/test/java/org/apache/ignite/platform/model/User.java
index 88849d16d65..3377765be76 100644
--- a/modules/core/src/test/java/org/apache/ignite/platform/model/User.java
+++ b/modules/core/src/test/java/org/apache/ignite/platform/model/User.java
@@ -17,6 +17,8 @@
package org.apache.ignite.platform.model;
+import java.util.Objects;
+
/** Test value object. */
public class User {
/** */
@@ -64,4 +66,17 @@ public class User {
public void setRole(Role role) {
this.role = role;
}
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ User user = (User)o;
+ return id == user.id && acl == user.acl && Objects.equals(role, user.role);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Objects.hash(id, acl, role);
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite.java
index a8a9ed93d31..ba890dd428c 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite.java
@@ -35,6 +35,8 @@ import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSn
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotRestoreFromRemoteTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotWithMetastorageTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.PlainSnapshotTest;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.IgniteCacheDumpSelf2Test;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.IgniteCacheDumpSelfTest;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.DynamicSuite;
import org.junit.runner.RunWith;
@@ -69,5 +71,7 @@ public class IgniteSnapshotTestSuite {
GridTestUtils.addTestIfNeeded(suite, IgniteSnapshotConsistencyTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite, IgniteClusterSnapshotDeltaTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite, IncrementalSnapshotsTestSuite.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite, IgniteCacheDumpSelfTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite, IgniteCacheDumpSelf2Test.class, ignoredTests);
}
}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/SqlCdcTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/SqlCdcTest.java
index 0cd6b7964ce..8691a0a91d2 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/SqlCdcTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/cdc/SqlCdcTest.java
@@ -376,7 +376,7 @@ public class SqlCdcTest extends AbstractCdcTest {
}
/** */
- static List<List<?>> executeSql(IgniteEx node, String sqlText, Object... args) {
+ public static List<List<?>> executeSql(IgniteEx node, String sqlText, Object... args) {
return node.context().query().querySqlFields(new SqlFieldsQuery(sqlText).setArgs(args), true).getAll();
}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/dump/DumpCacheConfigTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/dump/DumpCacheConfigTest.java
new file mode 100644
index 00000000000..e111223dc7e
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/dump/DumpCacheConfigTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.ignite.internal.dump;
+
+import java.util.Collection;
+import java.util.List;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.StoredCacheData;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.Dump;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.Dump.DumpedPartitionIterator;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.DumpEntry;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+import static org.apache.ignite.internal.cdc.SqlCdcTest.executeSql;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.AbstractCacheDumpTest.DMP_NAME;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.AbstractCacheDumpTest.KEYS_CNT;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.dump.AbstractCacheDumpTest.dump;
+
+/** */
+public class DumpCacheConfigTest extends GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ cleanPersistenceDir();
+
+ super.beforeTest();
+ }
+
+ /** */
+ @Test
+ public void testSQLTableDump() throws Exception {
+ IgniteEx srv = (IgniteEx)startGridsMultiThreaded(2);
+
+ executeSql(srv, "CREATE TABLE T1(ID INT, NAME VARCHAR, PRIMARY KEY (ID)) WITH \"CACHE_NAME=T1\"");
+
+ for (int i = 0; i < KEYS_CNT; i++)
+ executeSql(srv, "INSERT INTO T1 VALUES(?, ?)", i, "Name-" + i);
+
+ checkDump(srv, DMP_NAME, true);
+
+ executeSql(srv, "ALTER TABLE T1 ADD COLUMN(ADDRESS VARCHAR)");
+
+ for (int i = KEYS_CNT; i < KEYS_CNT * 2; i++)
+ executeSql(srv, "INSERT INTO T1 VALUES(?, ?, ?)", i, "Name-" + i, "Address-" + i);
+
+ checkDump(srv, DMP_NAME + 2, false);
+ }
+
+ /** */
+ private void checkDump(IgniteEx srv, String name, boolean first) throws Exception {
+ srv.snapshot().createDump(name).get(10_000L);
+
+ int grpId = CU.cacheId("T1");
+ int cnt = 0;
+
+ Dump dump = dump(srv, name);
+
+ for (String nodeDir : dump.nodesDirectories()) {
+ for (int part : dump.partitions(nodeDir, grpId)) {
+ try (DumpedPartitionIterator iter = dump.iterator(nodeDir, grpId, part)) {
+ while (iter.hasNext()) {
+ DumpEntry e = iter.next();
+
+ assertNotNull(e);
+
+ int id = e.key().value(null, false);
+
+ BinaryObject val = (BinaryObject)e.value();
+
+ assertNotNull(val);
+ assertEquals("Name-" + id, val.field("NAME"));
+
+ cnt++;
+ }
+ }
+ }
+
+ List<StoredCacheData> ccfgs = dump.configs(nodeDir, grpId);
+
+ assertNotNull(ccfgs);
+ assertEquals(1, ccfgs.size());
+ assertTrue(ccfgs.get(0).sql());
+
+ CacheConfiguration ccfg = ccfgs.get(0).config();
+
+ assertEquals("T1", ccfg.getName());
+
+ Collection<QueryEntity> qes = ccfgs.get(0).queryEntities();
+
+ assertNotNull(qes);
+ assertEquals(1, qes.size());
+
+ QueryEntity qe = qes.iterator().next();
+
+ assertNotNull(qe);
+ assertEquals("T1", qe.getTableName());
+ assertEquals(first ? 2 : 3, qe.getFields().size());
+ assertTrue(qe.getFields().containsKey("ID"));
+ assertTrue(qe.getFields().containsKey("NAME"));
+ if (!first)
+ assertTrue(qe.getFields().containsKey("ADDRESS"));
+ }
+
+ assertEquals(first ? KEYS_CNT : (KEYS_CNT * 2), cnt);
+ }
+}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
index 4805bf92d01..d33591cf3c9 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java
@@ -22,6 +22,7 @@ import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndexKey
import org.apache.ignite.internal.cdc.CacheEventsCdcTest;
import org.apache.ignite.internal.cdc.CdcIndexRebuildTest;
import org.apache.ignite.internal.cdc.SqlCdcTest;
+import org.apache.ignite.internal.dump.DumpCacheConfigTest;
import org.apache.ignite.internal.metric.SystemViewSelfTest;
import org.apache.ignite.internal.processors.cache.BigEntryQueryTest;
import org.apache.ignite.internal.processors.cache.BinaryMetadataConcurrentUpdateWithIndexesTest;
@@ -373,7 +374,9 @@ import org.junit.runners.Suite;
// CDC tests.
SqlCdcTest.class,
CacheEventsCdcTest.class,
- CdcIndexRebuildTest.class
+ CdcIndexRebuildTest.class,
+
+ DumpCacheConfigTest.class
})
public class IgniteBinaryCacheQueryTestSuite3 {
/** Setup lazy mode default. */