Ignite-2.5.18

modules/cassandra/store/pom.xml

@@ -35,10 +35,10 @@
     <url>http://ignite.apache.org</url>
 
     <properties>
-        <commons-beanutils.version>1.9.2</commons-beanutils.version>
-        <cassandra-driver.version>3.0.0</cassandra-driver.version>
-        <cassandra-all.version>3.3</cassandra-all.version>
-        <netty.version>4.1.27.Final</netty.version>
+        <commons-beanutils.version>1.9.4</commons-beanutils.version>
+        <cassandra-driver.version>3.2.0</cassandra-driver.version>
+        <cassandra-all.version>3.11.3</cassandra-all.version>
+        <netty.version>4.1.44.Final</netty.version>
         <guava.version>25.1-jre</guava.version>
         <metrics-core.version>3.0.2</metrics-core.version>
     </properties>

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

@@ -1148,14 +1148,6 @@ public final class IgniteSystemProperties {
      */
     public static final String IGNITE_LOG_CLASSPATH_CONTENT_ON_STARTUP = "IGNITE_LOG_CLASSPATH_CONTENT_ON_STARTUP";
 
-    /**
-     * Index rebuilding parallelism level. If specified, sets the count of threads that are used for index rebuilding
-     * and can only be greater than <code>0</code>, otherwise default value will be used. Maximum count of threads
-     * can't be greater than total available processors count.
-     * Default value is minimum of <code>4</code> and processors count / 4, but always greater than <code>0</code>.
-     */
-    public static final String INDEX_REBUILDING_PARALLELISM = "INDEX_REBUILDING_PARALLELISM";
-
     /** */
     public static final String IGNITE_USE_POOL_FOR_LAZY_QUERIES = "IGNITE_USE_POOL_FOR_LAZY_QUERIES";
 
@@ -1225,6 +1217,18 @@ public final class IgniteSystemProperties {
      */
     public static final String IGNITE_ENABLE_EXTRA_INDEX_REBUILD_LOGGING = "IGNITE_ENABLE_EXTRA_INDEX_REBUILD_LOGGING";
 
+    /**
+     * Choose the index cost function. May be used to compatibility with old version
+     * .
+     * The possible values:
+     *         - "LAST",
+     *         - "COMPATIBLE_8_5_17",
+     *         - COMPATIBLE_8_5_13
+     *
+     * The last cost function is used by default.
+     */
+    public static final String IGNITE_INDEX_COST_FUNCTION = "IGNITE_INDEX_COST_FUNCTION";
+
     /**
      * Enforces singleton.
      */

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

@@ -1169,14 +1169,15 @@ public CacheRebalanceMode getRebalanceMode() {
      * Gets cache rebalance order. Rebalance order can be set to non-zero value for caches with
      * {@link CacheRebalanceMode#SYNC SYNC} or {@link CacheRebalanceMode#ASYNC ASYNC} rebalance modes only.
      * <p/>
-     * If cache rebalance order is positive, rebalancing for this cache will be started only when rebalancing for
+     * The rebalance order guarantees that rebalancing for this cache will start only when rebalancing for
      * all caches with smaller rebalance order will be completed.
      * <p/>
-     * Note that cache with order {@code 0} does not participate in ordering. This means that cache with
-     * rebalance order {@code 0} will never wait for any other caches. All caches with order {@code 0} will
-     * be rebalanced right away concurrently with each other and ordered rebalance processes.
+     * Note that a cache with {@link CacheRebalanceMode#SYNC SYNC} rebalancing mode always takes precedence
+     * over caches with {@link CacheRebalanceMode#ASYNC ASYNC} rebalancing mode
+     * in the group of caches with the same rebalance order. This means caches with {@link CacheRebalanceMode#SYNC SYNC}
+     * rebalancing mode will be rebalanced in the first place.
      * <p/>
-     * If not set, cache order is 0, i.e. rebalancing is not ordered.
+     * If not set, cache order is 0.
      *
      * @return Cache rebalance order.
      */

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

@@ -79,6 +79,8 @@
 import org.apache.ignite.ssl.SslContextFactory;
 import org.jetbrains.annotations.Nullable;
 
+import static java.lang.Math.max;
+import static java.lang.Math.min;
 import static org.apache.ignite.plugin.segmentation.SegmentationPolicy.STOP;
 
 /**
@@ -147,7 +149,7 @@ public class IgniteConfiguration {
     public static final int AVAILABLE_PROC_CNT = Runtime.getRuntime().availableProcessors();
 
     /** Default core size of public thread pool. */
-    public static final int DFLT_PUBLIC_THREAD_CNT = Math.max(8, AVAILABLE_PROC_CNT);
+    public static final int DFLT_PUBLIC_THREAD_CNT = max(8, AVAILABLE_PROC_CNT);
 
     /** Default size of data streamer thread pool. */
     public static final int DFLT_DATA_STREAMER_POOL_SIZE = DFLT_PUBLIC_THREAD_CNT;
@@ -173,6 +175,9 @@ public class IgniteConfiguration {
     /** Default size of query thread pool. */
     public static final int DFLT_QUERY_THREAD_POOL_SIZE = DFLT_PUBLIC_THREAD_CNT;
 
+    /** Default size of index create/rebuild thread pool. */
+    public static final int DFLT_BUILD_IDX_THREAD_POOL_SIZE = min(4, max(1, AVAILABLE_PROC_CNT / 4));
+
     /** Default Ignite thread keep alive time. */
     public static final long DFLT_THREAD_KEEP_ALIVE_TIME = 60_000L;
 
@@ -281,6 +286,9 @@ public class IgniteConfiguration {
     /** Query pool size. */
     private int qryPoolSize = DFLT_QUERY_THREAD_POOL_SIZE;
 
+    /** Index create/rebuild pool size. */
+    private int buildIdxPoolSize = DFLT_BUILD_IDX_THREAD_POOL_SIZE;
+
     /** Ignite installation folder. */
     private String igniteHome;
 
@@ -626,6 +634,7 @@ public IgniteConfiguration(IgniteConfiguration cfg) {
         pluginCfgs = cfg.getPluginConfigurations();
         pubPoolSize = cfg.getPublicThreadPoolSize();
         qryPoolSize = cfg.getQueryThreadPoolSize();
+        buildIdxPoolSize = cfg.getBuildIndexThreadPoolSize();
         rebalanceThreadPoolSize = cfg.getRebalanceThreadPoolSize();
         rebalanceTimeout = cfg.getRebalanceTimeout();
         rebalanceBatchesPrefetchCnt = cfg.getRebalanceBatchesPrefetchCount();
@@ -1022,6 +1031,31 @@ public int getQueryThreadPoolSize() {
         return qryPoolSize;
     }
 
+    /**
+     * Size of thread pool for create/rebuild index.
+     * <p>
+     * If not provided, executor service will have size
+     * {@link #DFLT_BUILD_IDX_THREAD_POOL_SIZE}.
+     *
+     * @return Thread pool size for create/rebuild index.
+     */
+    public int getBuildIndexThreadPoolSize() {
+        return buildIdxPoolSize;
+    }
+
+    /**
+     * Sets index create/rebuild thread pool size to use within grid.
+     *
+     * @param poolSize Thread pool size to use within grid.
+     * @return {@code this} for chaining.
+     * @see IgniteConfiguration#getBuildIndexThreadPoolSize()
+     */
+    public IgniteConfiguration setBuildIndexThreadPoolSize(int poolSize) {
+        buildIdxPoolSize = poolSize;
+
+        return this;
+    }
+
     /**
      * Sets thread pool size to use within grid.
      *

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

@@ -340,30 +340,30 @@ private void initFilter(IgnitePredicate<Event> filter, GridKernalContext ctx) th
 
                 ClassLoader ldr = null;
 
-                if (cache != null) {
-                    GridCacheDeploymentManager depMgr = cache.context().deploy();
-
-                    GridDeploymentInfo depInfo = wrapper.depInfo;
-
-                    if (depInfo != null) {
-                        depMgr.p2pContext(
-                            nodeId,
-                            depInfo.classLoaderId(),
-                            depInfo.userVersion(),
-                            depInfo.deployMode(),
-                            depInfo.participants()
-                        );
-                    }
+                try {
+                    if (cache != null) {
+                        GridCacheDeploymentManager depMgr = cache.context().deploy();
+
+                        GridDeploymentInfo depInfo = wrapper.depInfo;
+
+                        if (depInfo != null) {
+                            depMgr.p2pContext(
+                                nodeId,
+                                depInfo.classLoaderId(),
+                                depInfo.userVersion(),
+                                depInfo.deployMode(),
+                                depInfo.participants()
+                            );
+                        }
 
-                    ldr = depMgr.globalLoader();
-                }
-                else {
-                    U.warn(ctx.log(getClass()), "Received cache event for cache that is not configured locally " +
-                        "when peer class loading is enabled: " + wrapper.cacheName + ". Will try to unmarshal " +
-                        "with default class loader.");
-                }
+                        ldr = depMgr.globalLoader();
+                    }
+                    else {
+                        U.warn(ctx.log(getClass()), "Received cache event for cache that is not configured locally " +
+                            "when peer class loading is enabled: " + wrapper.cacheName + ". Will try to unmarshal " +
+                            "with default class loader.");
+                    }
 
-                try {
                     wrapper.p2pUnmarshal(ctx.config().getMarshaller(), U.resolveClassLoader(ldr, ctx.config()));
                 }
                 catch (IgniteCheckedException e) {

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

@@ -716,4 +716,11 @@ public interface GridKernalContext extends Iterable<GridComponent> {
      * @return Local continuous tasks processor.
      */
     public DurableBackgroundTasksProcessor durableBackgroundTasksProcessor();
+
+    /**
+     * Return Thread pool for create/rebuild indexes.
+     *
+     * @return Thread pool for create/rebuild indexes.
+     */
+    public ExecutorService buildIndexExecutorService();
 }

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

@@ -348,6 +348,10 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
     @GridToStringExclude
     protected ExecutorService idxExecSvc;
 
+    /** Thread pool for create/rebuild indexes. */
+    @GridToStringExclude
+    private ExecutorService buildIdxExecSvc;
+
     /** */
     @GridToStringExclude
     protected IgniteStripedThreadPoolExecutor callbackExecSvc;
@@ -451,6 +455,7 @@ public GridKernalContextImpl() {
      * @param restExecSvc REST executor service.
      * @param affExecSvc Affinity executor service.
      * @param idxExecSvc Indexing executor service.
+     * @param buildIdxExecSvc Create/rebuild indexes executor service.
      * @param callbackExecSvc Callback executor service.
      * @param qryExecSvc Query executor service.
      * @param schemaExecSvc Schema executor service.
@@ -479,6 +484,7 @@ protected GridKernalContextImpl(
         ExecutorService restExecSvc,
         ExecutorService affExecSvc,
         @Nullable ExecutorService idxExecSvc,
+        @Nullable ExecutorService buildIdxExecSvc,
         IgniteStripedThreadPoolExecutor callbackExecSvc,
         ExecutorService qryExecSvc,
         ExecutorService schemaExecSvc,
@@ -509,6 +515,7 @@ protected GridKernalContextImpl(
         this.restExecSvc = restExecSvc;
         this.affExecSvc = affExecSvc;
         this.idxExecSvc = idxExecSvc;
+        this.buildIdxExecSvc = buildIdxExecSvc;
         this.callbackExecSvc = callbackExecSvc;
         this.qryExecSvc = qryExecSvc;
         this.schemaExecSvc = schemaExecSvc;
@@ -1219,6 +1226,11 @@ public void recoveryMode(boolean recoveryMode) {
         return durableBackgroundTasksProcessor;
     }
 
+    /** {@inheritDoc} */
+    @Override public ExecutorService buildIndexExecutorService() {
+        return buildIdxExecSvc;
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(GridKernalContextImpl.class, this);

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

@@ -18,9 +18,13 @@
 package org.apache.ignite.internal;
 
 import java.util.BitSet;
+import java.util.Collection;
 import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi;
 import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
 import org.apache.ignite.spi.communication.tcp.messages.HandshakeWaitMessage;
+import org.apache.ignite.spi.discovery.DiscoverySpi;
 
 import static org.apache.ignite.IgniteSystemProperties.getBoolean;
 import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_FEATURES;
@@ -83,7 +87,13 @@ public enum IgniteFeatures {
     DISTRIBUTED_CHANGE_LONG_OPERATIONS_DUMP_TIMEOUT(30),
 
     /** Cluster has task to get value from cache by key value. */
-    WC_GET_CACHE_VALUE(31);
+    WC_GET_CACHE_VALUE(31),
+
+    /** */
+    VOLATILE_DATA_STRUCTURES_REGION(33),
+
+    /** Partition reconciliation utility. */
+    PARTITION_RECONCILIATION(34);
 
     /**
      * Unique feature identifier.
@@ -160,6 +170,24 @@ public static boolean allNodesSupports(Iterable<ClusterNode> nodes, IgniteFeatur
         return true;
     }
 
+    /**
+     * @param cfg Configuration.
+     * @param feature Feature to check.
+     *
+     * @return {@code True} if all nodes in the cluster support given feature.
+     */
+    public static boolean allNodesSupport(IgniteConfiguration cfg, IgniteFeatures feature) {
+        DiscoverySpi discoSpi = cfg.getDiscoverySpi();
+
+        if (discoSpi instanceof IgniteDiscoverySpi)
+            return ((IgniteDiscoverySpi)discoSpi).allNodesSupport(feature);
+        else {
+            Collection<ClusterNode> nodes = discoSpi.getRemoteNodes();
+
+            return allNodesSupports(nodes, feature);
+        }
+    }
+
     /**
      * Features supported by the current node.
      *