Ignite 11073 12069 (mmuzaf)

modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java

@@ -745,6 +745,16 @@ public final class IgniteSystemProperties {
      */
     public static final String IGNITE_PDS_WAL_REBALANCE_THRESHOLD = "IGNITE_PDS_WAL_REBALANCE_THRESHOLD";
 
+    /**
+     * Flag to enable file rebalancing.
+     */
+    public static final String IGNITE_FILE_REBALANCE_ENABLED = "IGNITE_FILE_REBALANCE_ENABLED";
+
+    /**
+     * File rebalancing is applicable only if the number of entries in the partition is greater than this threshold.
+     */
+    public static final String IGNITE_FILE_REBALANCE_THRESHOLD = "IGNITE_PDS_FILE_REBALANCE_THRESHOLD";
+
     /** Ignite page memory concurrency level. */
     public static final String IGNITE_OFFHEAP_LOCK_CONCURRENCY_LEVEL = "IGNITE_OFFHEAP_LOCK_CONCURRENCY_LEVEL";
 

modules/core/src/main/java/org/apache/ignite/configuration/DataStorageConfiguration.java

@@ -81,6 +81,9 @@ public class DataStorageConfiguration implements Serializable {
         (long)(DFLT_DATA_REGION_FRACTION * U.getTotalMemoryAvailable()),
         DFLT_DATA_REGION_INITIAL_SIZE);
 
+    /** Default working directory for a snapshot final result. */
+    public static final String DFLT_LOCAL_SNAPSHOT_DIRECTORY = "snapshots";
+
     /** Default initial size of a memory chunk for the system cache (40 MB). */
     private static final long DFLT_SYS_REG_INIT_SIZE = 40L * 1024 * 1024;
 
@@ -159,6 +162,9 @@ public class DataStorageConfiguration implements Serializable {
     /** Default wal archive directory. */
     public static final String DFLT_WAL_ARCHIVE_PATH = "db/wal/archive";
 
+    /** Default working directory for backup temporary files. */
+    public static final String DFLT_BACKUP_DIRECTORY = "db/backup";
+
     /** Default write throttling enabled. */
     public static final boolean DFLT_WRITE_THROTTLING_ENABLED = false;
 
@@ -194,6 +200,9 @@ public class DataStorageConfiguration implements Serializable {
     /** Directory where index and partition files are stored. */
     private String storagePath;
 
+    /** Directory where will be stored all results of snapshot operations. */
+    private String locSnapshotPath = DFLT_LOCAL_SNAPSHOT_DIRECTORY;
+
     /** Checkpoint frequency. */
     private long checkpointFreq = DFLT_CHECKPOINT_FREQ;
 
@@ -472,6 +481,24 @@ public DataStorageConfiguration setStoragePath(String persistenceStorePath) {
         return this;
     }
 
+    /**
+     * @return Relative path where will be stored all local snapshot results.
+     */
+    public String getLocalSnapshotPath() {
+        return locSnapshotPath;
+    }
+
+    /**
+     * @param locSnapshotPath Relative path to store all local snapshot results. By default
+     * {@code DFLT_LOCAL_SNAPSHOT_DIRECTORY} is used.
+     * @return Data storage configurations for chaining.
+     */
+    public DataStorageConfiguration setLocalSnapshotPath(String locSnapshotPath) {
+        this.locSnapshotPath = locSnapshotPath;
+
+        return this;
+    }
+
     /**
      * Gets checkpoint frequency.
      *

modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java

@@ -176,6 +176,9 @@ public class IgniteConfiguration {
     /** Default rebalance batch size in bytes (value is {@code 512Kb}). */
     public static final int DFLT_REBALANCE_BATCH_SIZE = 512 * 1024; // 512K
 
+    /** Default WAL rebalance threshold (value is 500 000 cache entries). */
+    public static final long DFLT_PDS_WAL_REBALANCE_THRESHOLD = 500_000;
+
     /** Default size of system thread pool. */
     public static final int DFLT_SYSTEM_CORE_THREAD_CNT = DFLT_PUBLIC_THREAD_CNT;
 

modules/core/src/main/java/org/apache/ignite/internal/IgniteFeatures.java

@@ -99,7 +99,10 @@ public enum IgniteFeatures {
     PERSISTENCE_CACHE_SNAPSHOT(23),
 
     /** Long operations dump timeout. */
-    LONG_OPERATIONS_DUMP_TIMEOUT(30);
+    LONG_OPERATIONS_DUMP_TIMEOUT(30),
+
+    /** */
+    CACHE_PARTITION_FILE_REBALANCE(24);
 
     /**
      * Unique feature identifier.

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/AbstractTransmission.java

@@ -57,9 +57,9 @@ protected AbstractTransmission(
         int chunkSize
     ) {
         A.notNull(meta, "Initial file meta cannot be null");
-        A.notNullOrEmpty(meta.name(), "Trasmisson name cannot be empty or null");
+        A.notNullOrEmpty(meta.name(), "Transmission name cannot be empty or null");
         A.ensure(meta.offset() >= 0, "File start position cannot be negative");
-        A.ensure(meta.count() > 0, "Total number of bytes to transfer must be greater than zero");
+        A.ensure(meta.count() >= 0, "Total number of bytes to transfer can't be less than zero");
         A.notNull(stopChecker, "Process stop checker cannot be null");
         A.ensure(chunkSize > 0, "Size of chunks to transfer data must be positive");
 

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/FileReceiver.java

@@ -82,7 +82,8 @@ public FileReceiver(
             fileIo.position(meta.offset());
         }
         catch (IOException e) {
-            throw new IgniteException("Unable to open destination file. Receiver will will be stopped", e);
+            throw new IgniteException("Unable to open destination file. Receiver will be stopped: " +
+                file.getAbsolutePath(), e);
         }
     }
 

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java

@@ -257,6 +257,9 @@ public class GridIoManager extends GridManagerAdapter<CommunicationSpi<Serializa
     /** Direct protocol version. */
     public static final byte DIRECT_PROTO_VER = 3;
 
+    /** Session timeout which processing files. */
+    private static final int CHANNEL_SESSION_TIMEOUT_MS = 60_000;
+
     /** Current IO policy. */
     private static final ThreadLocal<Byte> CUR_PLC = new ThreadLocal<>();
 
@@ -2711,9 +2714,15 @@ private void interruptRecevier(ReceiverContext rctx, Exception ex) {
             rctx.lastState = rctx.lastState == null ?
                 new TransmissionMeta(ex) : rctx.lastState.error(ex);
 
-            rctx.hnd.onException(rctx.rmtNodeId, ex);
+            if (X.hasCause(ex, TransmissionCancelledException.class)) {
+                if (log.isInfoEnabled())
+                    log.info("Transmission receiver has been cancelled [rctx=" + rctx + ']');
+            }
+            else {
+                rctx.hnd.onException(rctx.rmtNodeId, ex);
 
-            U.error(log, "Receiver has been interrupted due to an exception occurred [ctx=" + rctx + ']', ex);
+                U.error(log, "Receiver has been interrupted due to an exception occurred [rctx=" + rctx + ']', ex);
+            }
         }
     }
 
@@ -2767,7 +2776,8 @@ private void processOpenedChannel(Object topic, UUID rmtNodeId, SessionChannelMe
                     "It's not allowed to process different sessions over the same topic simultaneously. " +
                     "Channel will be closed [initMsg=" + initMsg + ", channel=" + ch + ", nodeId=" + rmtNodeId + ']');
 
-                U.error(log, err);
+                U.error(log, "Error has been sent back to remote node. Receiver holds the local topic " +
+                    "[topic=" + topic + ", rmtNodeId=" + rmtNodeId + ", ctx=" + rcvCtx + ']', err);
 
                 out.writeObject(new TransmissionMeta(err));
 
@@ -2799,8 +2809,6 @@ private void processOpenedChannel(Object topic, UUID rmtNodeId, SessionChannelMe
             }
         }
         catch (Throwable t) {
-            U.error(log, "Download session cannot be finished due to an unexpected error [ctx=" + rcvCtx + ']', t);
-
             // Do not remove receiver context here, since sender will recconect to get this error.
             interruptRecevier(rcvCtx, new IgniteCheckedException("Channel processing error [nodeId=" + rmtNodeId + ']', t));
         }
@@ -2898,7 +2906,7 @@ private void receiveFromChannel(
                 }
 
                 @Override public long endTime() {
-                    return startTs + netTimeoutMs;
+                    return startTs + CHANNEL_SESSION_TIMEOUT_MS;
                 }
 
                 @Override public void onTimeout() {
@@ -3174,6 +3182,13 @@ private TransmissionMeta connect() throws IgniteCheckedException, IOException {
             return syncMeta;
         }
 
+        /**
+         * @return {@code true} if sender's session is currently active.
+         */
+        public boolean opened() {
+            return senderStopFlags.containsKey(sesKey);
+        }
+
         /**
          * @param file Source file to send to remote.
          * @param params Additional file params.
@@ -3312,6 +3327,11 @@ public void send(
                 if (stopping)
                     throw new NodeStoppingException("Operation has been cancelled (node is stopping)");
 
+                if (senderStopFlags.get(sesKey) == null)
+                    e.printStackTrace();
+
+                assert senderStopFlags.get(sesKey) != null : "key=" + sesKey + ", flags=" + senderStopFlags.keySet();
+
                 if (senderStopFlags.get(sesKey).get())
                     throw new ClusterTopologyCheckedException("Remote node left the cluster: " + rmtId, e);
 

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java

@@ -138,6 +138,8 @@
 import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccTxSnapshotRequest;
 import org.apache.ignite.internal.processors.cache.mvcc.msg.PartitionCountersNeighborcastRequest;
 import org.apache.ignite.internal.processors.cache.mvcc.msg.PartitionCountersNeighborcastResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRequestMessage;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotResponseMessage;
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryResponse;
 import org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery;
@@ -401,6 +403,8 @@ public class GridIoMessageFactory implements MessageFactoryProvider {
         factory.register(GridIoSecurityAwareMessage.TYPE_CODE, GridIoSecurityAwareMessage::new);
         factory.register(SessionChannelMessage.TYPE_CODE, SessionChannelMessage::new);
         factory.register(SingleNodeMessage.TYPE_CODE, SingleNodeMessage::new);
+        factory.register(SnapshotRequestMessage.TYPE_CODE, SnapshotRequestMessage::new);
+        factory.register(SnapshotResponseMessage.TYPE_CODE, SnapshotResponseMessage::new);
 
         // [-3..119] [124..129] [-23..-28] [-36..-55] - this
         // [120..123] - DR

modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java

@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.pagemem.store;
 
+import java.io.File;
 import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.function.Predicate;
@@ -117,9 +118,8 @@ public void initializeForCache(CacheGroupDescriptor grpDesc, StoredCacheData cac
      * @param grpId Cache group ID.
      * @param partId Partition ID.
      * @return {@code True} if partition store exists.
-     * @throws IgniteCheckedException If failed.
      */
-    public boolean exists(int grpId, int partId) throws IgniteCheckedException;
+    public boolean exists(int grpId, int partId);
 
     /**
      * Reads a header of a page store.
@@ -266,4 +266,14 @@ public void initializeForCache(CacheGroupDescriptor grpDesc, StoredCacheData cac
      * @throws IgniteCheckedException If failed.
      */
     public boolean checkAndInitCacheWorkDir(CacheConfiguration cacheCfg) throws IgniteCheckedException;
+
+    /**
+     * Restore partition file.
+     *
+     * @param grpId Cache group ID.
+     * @param partId Partition ID.
+     * @param src Partition file.
+     * @throws IgniteCheckedException If the page store doesn't exists or the file couldn't be moved.
+     */
+    public void restore(int grpId, int partId, File src) throws IgniteCheckedException;
 }

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/IgniteWriteAheadLogManager.java

@@ -119,13 +119,15 @@ public WALPointer log(WALRecord entry, RolloverType rolloverType)
      * Invoke this method to iterate over the written log entries.
      *
      * @param start Optional WAL pointer from which to start iteration.
+     * @param limited Limit iterator to current segment.
      * @param recordDeserializeFilter Specify a filter to skip WAL records. Those records will not be explicitly deserialized.
      * @return Records iterator.
      * @throws IgniteException If failed to start iteration.
      * @throws StorageException If IO error occurred while reading WAL entries.
      */
     public WALIterator replay(
         WALPointer start,
+        boolean limited,
         @Nullable IgniteBiPredicate<WALRecord.RecordType, WALPointer> recordDeserializeFilter
     ) throws IgniteCheckedException, StorageException;