You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sk...@apache.org on 2020/02/28 16:20:00 UTC
[ignite] branch master updated: IGNITE-12656 Cleanup
GridCacheProcessor from functionality not related to it.
This is an automated email from the ASF dual-hosted git repository.
sk0x50 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 0afd1f6 IGNITE-12656 Cleanup GridCacheProcessor from functionality not related to it.
0afd1f6 is described below
commit 0afd1f6f664a23a5dd688a282efcc61d75748476
Author: Slava Koptilin <sl...@gmail.com>
AuthorDate: Fri Feb 28 19:19:21 2020 +0300
IGNITE-12656 Cleanup GridCacheProcessor from functionality not related to it.
Co-authored-by: Pavel Kovalenko <jo...@gmail.com>
---
.../org/apache/ignite/internal/IgniteKernal.java | 8 +-
.../ignite/internal/TransactionsMXBeanImpl.java | 10 +-
.../ignite/internal/cluster/IgniteClusterImpl.java | 14 +-
.../processors/affinity/LocalAffinityFunction.java | 78 +++
.../cache/CacheAffinitySharedManager.java | 2 +-
.../processors/cache/ClusterCachesInfo.java | 10 +-
.../cache/GridCachePartitionExchangeManager.java | 8 +-
.../processors/cache/GridCachePreloader.java | 2 +-
.../processors/cache/GridCacheProcessor.java | 646 ++++-----------------
.../processors/cache/GridCacheSharedContext.java | 10 +-
.../internal/processors/cache/GridCacheUtils.java | 7 +-
.../cache/ValidationOnNodeJoinUtils.java | 5 +-
.../internal/processors/cache/WalStateManager.java | 17 +-
.../dht/topology/GridDhtLocalPartition.java | 2 +-
.../PartitionDefferedDeleteQueueCleanupTask.java | 107 ++++
.../GridCacheDatabaseSharedManager.java | 17 +
.../cache/persistence/metastorage/MetaStorage.java | 11 +-
.../cache/transactions/IgniteTxManager.java | 184 ++++++
.../processors/query/GridQueryProcessor.java | 14 +
.../ignite/internal/util/IgniteCollectors.java | 55 ++
.../internal/util/future/GridCompoundFuture.java | 8 +-
.../affinity/local/LocalAffinityFunctionTest.java | 5 +-
.../cache/GridCacheProcessorActiveTxTest.java | 42 +-
.../IgniteStartCacheInTransactionSelfTest.java | 2 +-
24 files changed, 654 insertions(+), 610 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 96c04b2..6db4c91 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -17,9 +17,6 @@
package org.apache.ignite.internal;
-import java.text.DecimalFormatSymbols;
-import javax.cache.CacheException;
-import javax.management.JMException;
import java.io.Externalizable;
import java.io.File;
import java.io.IOException;
@@ -40,6 +37,7 @@ import java.nio.file.Path;
import java.nio.file.Paths;
import java.text.DateFormat;
import java.text.DecimalFormat;
+import java.text.DecimalFormatSymbols;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -56,6 +54,8 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import javax.cache.CacheException;
+import javax.management.JMException;
import org.apache.ignite.DataRegionMetrics;
import org.apache.ignite.DataRegionMetricsAdapter;
import org.apache.ignite.DataStorageMetrics;
@@ -3990,7 +3990,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
guard();
try {
- ctx.cache().resetCacheState(cacheNames).get();
+ ctx.cache().resetCacheState(cacheNames.isEmpty() ? ctx.cache().cacheNames() : cacheNames).get();
}
catch (IgniteCheckedException e) {
throw U.convertException(e);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/TransactionsMXBeanImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/TransactionsMXBeanImpl.java
index 4420a29..e9dd008 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/TransactionsMXBeanImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/TransactionsMXBeanImpl.java
@@ -124,7 +124,7 @@ public class TransactionsMXBeanImpl implements TransactionsMXBean {
/** {@inheritDoc} */
@Override public void setTxTimeoutOnPartitionMapExchange(long timeout) {
try {
- ctx.grid().context().cache().setTxTimeoutOnPartitionMapExchange(timeout);
+ ctx.cache().context().tm().setTxTimeoutOnPartitionMapExchange(timeout);
}
catch (Exception e) {
throw new RuntimeException(e.getMessage());
@@ -138,7 +138,7 @@ public class TransactionsMXBeanImpl implements TransactionsMXBean {
/** {@inheritDoc} */
@Override public void setTxOwnerDumpRequestsAllowed(boolean allowed) {
- ctx.cache().setTxOwnerDumpRequestsAllowed(allowed);
+ ctx.cache().context().tm().setTxOwnerDumpRequestsAllowedDistributed(allowed);
}
/** {@inheritDoc} */
@@ -148,7 +148,7 @@ public class TransactionsMXBeanImpl implements TransactionsMXBean {
/** {@inheritDoc} */
@Override public void setLongTransactionTimeDumpThreshold(long threshold) {
- ctx.cache().longTransactionTimeDumpThreshold(threshold);
+ ctx.cache().context().tm().longTransactionTimeDumpThresholdDistributed(threshold);
}
/** {@inheritDoc} */
@@ -158,7 +158,7 @@ public class TransactionsMXBeanImpl implements TransactionsMXBean {
/** {@inheritDoc} */
@Override public void setTransactionTimeDumpSamplesCoefficient(double coefficient) {
- ctx.cache().transactionTimeDumpSamplesCoefficient(coefficient);
+ ctx.cache().context().tm().transactionTimeDumpSamplesCoefficientDistributed(coefficient);
}
/** {@inheritDoc} */
@@ -168,7 +168,7 @@ public class TransactionsMXBeanImpl implements TransactionsMXBean {
/** {@inheritDoc} */
@Override public void setTransactionTimeDumpSamplesPerSecondLimit(int limit) {
- ctx.cache().longTransactionTimeDumpSamplesPerSecondLimit(limit);
+ ctx.cache().context().tm().longTransactionTimeDumpSamplesPerSecondLimit(limit);
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterImpl.java
index 7fc666a..070af2b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/IgniteClusterImpl.java
@@ -52,6 +52,7 @@ import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteComponentType;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.discovery.DiscoCache;
+import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
import org.apache.ignite.internal.processors.cluster.BaselineTopology;
import org.apache.ignite.internal.processors.cluster.baseline.autoadjust.BaselineAutoAdjustStatus;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
@@ -571,7 +572,7 @@ public class IgniteClusterImpl extends ClusterGroupAdapter implements IgniteClus
guard();
try {
- ctx.cache().setTxTimeoutOnPartitionMapExchange(timeout);
+ ctx.cache().context().tm().setTxTimeoutOnPartitionMapExchange(timeout);
}
catch (IgniteCheckedException e) {
throw U.convertException(e);
@@ -609,7 +610,7 @@ public class IgniteClusterImpl extends ClusterGroupAdapter implements IgniteClus
guard();
try {
- return ctx.cache().changeWalMode(Collections.singleton(cacheName), enabled).get();
+ return ctx.cache().context().walState().changeWalMode(Collections.singleton(cacheName), enabled).get();
}
catch (IgniteCheckedException e) {
throw U.convertException(e);
@@ -620,11 +621,16 @@ public class IgniteClusterImpl extends ClusterGroupAdapter implements IgniteClus
}
/** {@inheritDoc} */
- @Override public boolean isWalEnabled(String cacheName) {
+ @Override public boolean isWalEnabled(String cacheName) throws IgniteException {
guard();
try {
- return ctx.cache().walEnabled(cacheName);
+ DynamicCacheDescriptor desc = ctx.cache().cacheDescriptor(cacheName);
+
+ if (desc == null)
+ throw new IgniteException("Cache not found: " + cacheName);
+
+ return desc.groupDescriptor().walEnabled();
}
finally {
unguard();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/LocalAffinityFunction.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/LocalAffinityFunction.java
new file mode 100644
index 0000000..963de8c
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/LocalAffinityFunction.java
@@ -0,0 +1,78 @@
+/*
+ * 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.affinity;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.affinity.AffinityFunction;
+import org.apache.ignite.cache.affinity.AffinityFunctionContext;
+import org.apache.ignite.cluster.ClusterNode;
+
+/**
+ * Affinity function for {@link org.apache.ignite.cache.CacheMode#LOCAL} caches.
+ */
+public class LocalAffinityFunction implements AffinityFunction {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** {@inheritDoc} */
+ @Override public List<List<ClusterNode>> assignPartitions(AffinityFunctionContext affCtx) {
+ ClusterNode locNode = null;
+
+ for (ClusterNode n : affCtx.currentTopologySnapshot()) {
+ if (n.isLocal()) {
+ locNode = n;
+
+ break;
+ }
+ }
+
+ if (locNode == null)
+ throw new IgniteException("Local node is not included into affinity nodes for 'LOCAL' cache");
+
+ List<List<ClusterNode>> res = new ArrayList<>(partitions());
+
+ for (int part = 0; part < partitions(); part++)
+ res.add(Collections.singletonList(locNode));
+
+ return Collections.unmodifiableList(res);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void reset() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public int partitions() {
+ return 1;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int partition(Object key) {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void removeNode(UUID nodeId) {
+ // No-op.
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
index 46ca65f..6fffd16 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
@@ -966,7 +966,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
);
}
else
- cctx.kernalContext().cache().initQueryStructuresForNotStartedCache(cacheDesc);
+ cctx.kernalContext().query().initQueryStructuresForNotStartedCache(cacheDesc);
}
Map<StartCacheInfo, IgniteCheckedException> failedCaches = cctx.cache().prepareStartCachesIfPossible(startCacheInfos.keySet());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java
index 5359cc1..f40fd74 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java
@@ -34,6 +34,7 @@ import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
@@ -157,6 +158,9 @@ public class ClusterCachesInfo {
/** {@code True} if joined cluster while cluster state change was in progress. */
private boolean joinOnTransition;
+ /** Flag that caches were already filtered out. */
+ private final AtomicBoolean alreadyFiltered = new AtomicBoolean();
+
/**
* @param ctx Context.
*/
@@ -186,6 +190,9 @@ public class ClusterCachesInfo {
if (ctx.isDaemon())
return;
+ if (!alreadyFiltered.compareAndSet(false, true))
+ return;
+
filterRegisteredCachesAndCacheGroups(localCachesOnStart);
List<T2<DynamicCacheDescriptor, NearCacheConfiguration>> locJoinStartCaches = locJoinCachesCtx.caches();
@@ -200,7 +207,8 @@ public class ClusterCachesInfo {
locJoinStartCaches,
initCaches,
registeredCacheGrps,
- registeredCaches);
+ registeredCaches
+ );
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index cc65919..89d4f3b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -756,9 +756,11 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
/**
* @param active Cluster state.
* @param reconnect Reconnect flag.
+ * @return Topology version of local join exchange if cluster is active.
+ * Topology version NONE if cluster is not active or reconnect.
* @throws IgniteCheckedException If failed.
*/
- public void onKernalStart(boolean active, boolean reconnect) throws IgniteCheckedException {
+ public AffinityTopologyVersion onKernalStart(boolean active, boolean reconnect) throws IgniteCheckedException {
for (ClusterNode n : cctx.discovery().remoteNodes())
cctx.versions().onReceived(n.id(), n.metrics().getLastDataVersion());
@@ -852,7 +854,11 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
if (log.isDebugEnabled())
log.debug("Finished waiting for initial exchange: " + fut.exchangeId());
+
+ return fut.initialVersion();
}
+
+ return NONE;
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java
index a5212a3..1b6aca1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePreloader.java
@@ -101,7 +101,7 @@ public interface GridCachePreloader {
public IgniteInternalFuture<Object> startFuture();
/**
- * @return Future which will complete when preloading is finished.
+ * @return Future which will complete when initial preloading is finished.
*/
public IgniteInternalFuture<?> syncFuture();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 1cc78d8..b5415a4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -34,28 +34,22 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
-import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import javax.management.MBeanServer;
import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteCompute;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteInterruptedException;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.CacheExistsException;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.QueryEntity;
-import org.apache.ignite.cache.affinity.AffinityFunction;
-import org.apache.ignite.cache.affinity.AffinityFunctionContext;
import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cache.store.CacheStoreSessionListener;
-import org.apache.ignite.cluster.ClusterGroup;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
@@ -63,7 +57,6 @@ import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.DeploymentMode;
import org.apache.ignite.configuration.FileSystemConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
-import org.apache.ignite.configuration.TransactionConfiguration;
import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
@@ -92,8 +85,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheA
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtColocatedCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.StopCachesOnClientReconnectExchangeTask;
-import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
-import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.PartitionDefferedDeleteQueueCleanupTask;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.PartitionsEvictManager;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearAtomicCache;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTransactionalCache;
@@ -113,7 +105,6 @@ import org.apache.ignite.internal.processors.cache.persistence.RowStore;
import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList;
import org.apache.ignite.internal.processors.cache.persistence.freelist.PagesList;
-import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageLifecycleListener;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadOnlyMetastorage;
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
@@ -142,10 +133,11 @@ import org.apache.ignite.internal.processors.query.schema.SchemaExchangeWorkerTa
import org.apache.ignite.internal.processors.query.schema.SchemaNodeLeaveExchangeWorkerTask;
import org.apache.ignite.internal.processors.query.schema.message.SchemaAbstractDiscoveryMessage;
import org.apache.ignite.internal.processors.query.schema.message.SchemaProposeDiscoveryMessage;
+import org.apache.ignite.internal.processors.security.IgniteSecurity;
import org.apache.ignite.internal.processors.service.GridServiceProcessor;
-import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
import org.apache.ignite.internal.util.F0;
+import org.apache.ignite.internal.util.IgniteCollectors;
import org.apache.ignite.internal.util.InitializationProtector;
import org.apache.ignite.internal.util.StripedExecutor;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
@@ -155,7 +147,6 @@ import org.apache.ignite.internal.util.lang.GridPlainClosure2;
import org.apache.ignite.internal.util.lang.IgniteOutClosureX;
import org.apache.ignite.internal.util.lang.IgniteThrowableFunction;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
-import org.apache.ignite.internal.util.typedef.CIX1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.X;
@@ -168,7 +159,6 @@ import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
-import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.lifecycle.LifecycleAware;
import org.apache.ignite.marshaller.Marshaller;
@@ -199,23 +189,20 @@ import static org.apache.ignite.configuration.DeploymentMode.CONTINUOUS;
import static org.apache.ignite.configuration.DeploymentMode.SHARED;
import static org.apache.ignite.internal.GridComponent.DiscoveryDataExchangeType.CACHE_PROC;
import static org.apache.ignite.internal.IgniteComponentType.JTA;
-import static org.apache.ignite.internal.IgniteFeatures.LRT_SYSTEM_USER_TIME_DUMP_SETTINGS;
-import static org.apache.ignite.internal.IgniteFeatures.TRANSACTION_OWNER_THREAD_DUMP_PROVIDING;
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.isNearEnabled;
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.isPersistentCache;
import static org.apache.ignite.internal.processors.cache.ValidationOnNodeJoinUtils.validateHashIdResolvers;
import static org.apache.ignite.internal.util.IgniteUtils.doInParallel;
/**
- * Cache processor.
+ * The main responsibility of this processor is orchestrating of starting, stopping caches and everything related
+ * to that activity like validation of cache configuration, restoring configuration from persistence, etc.
*/
@SuppressWarnings({"unchecked", "TypeMayBeWeakened", "deprecation"})
public class GridCacheProcessor extends GridProcessorAdapter {
/** */
- private static final String CACHE_NAME_AND_OPERATION_FORMAT = "[cacheName=%s, operation=%s]";
-
- /** */
- private static final String CACHE_NAMES_AND_OPERATION_FORMAT = "[cacheNames=%s, operation=%s]";
+ private static final String CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT =
+ "Cannot start/stop cache within lock or transaction [cacheNames=%s, operation=%s]";
/** System view name for page lists. */
public static final String CACHE_GRP_PAGE_LIST_VIEW = "cacheGroupPageLists";
@@ -231,15 +218,15 @@ public class GridCacheProcessor extends GridProcessorAdapter {
private final boolean keepStaticCacheConfiguration = IgniteSystemProperties.getBoolean(
IgniteSystemProperties.IGNITE_KEEP_STATIC_CACHE_CONFIGURATION);
+ /** MBean group for cache group metrics */
+ private static final String CACHE_GRP_METRICS_MBEAN_GRP = "Cache groups";
+
/** Shared cache context. */
private GridCacheSharedContext<?, ?> sharedCtx;
/** */
private final ConcurrentMap<Integer, CacheGroupContext> cacheGrps = new ConcurrentHashMap<>();
- /** Flag that caches were already filtered out. */
- private final AtomicBoolean alreadyFiltered = new AtomicBoolean();
-
/** */
private final Map<String, GridCacheAdapter<?, ?>> caches;
@@ -261,10 +248,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
/** Enable/disable cache statistics futures. */
private ConcurrentMap<UUID, EnableStatisticsFuture> manageStatisticsFuts = new ConcurrentHashMap<>();
- /** The futures for changing transaction timeout on partition map exchange. */
- private ConcurrentMap<UUID, TxTimeoutOnPartitionMapExchangeChangeFuture> txTimeoutOnPartitionMapExchangeFuts =
- new ConcurrentHashMap<>();
-
/** */
private ClusterCachesInfo cachesInfo;
@@ -283,18 +266,12 @@ public class GridCacheProcessor extends GridProcessorAdapter {
/** Internal cache names. */
private final Set<String> internalCaches;
- /** MBean group for cache group metrics */
- private final String CACHE_GRP_METRICS_MBEAN_GRP = "Cache groups";
-
/** Protector of initialization of specific value. */
private final InitializationProtector initializationProtector = new InitializationProtector();
/** Cache recovery lifecycle state and actions. */
private final CacheRecoveryLifecycle recovery = new CacheRecoveryLifecycle();
- /** Tmp storage for meta migration. */
- private MetaStorage.TmpStorage tmpStorage;
-
/** Cache configuration splitter. */
private CacheConfigurationSplitter splitter;
@@ -351,14 +328,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * Start cache rebalance.
- */
- public void enableRebalance() {
- for (IgniteCacheProxy c : publicCaches())
- c.rebalance();
- }
-
- /**
* Create exchange worker task for custom discovery message.
*
* @param msg Custom discovery message.
@@ -426,7 +395,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
else if (task instanceof TxTimeoutOnPartitionMapExchangeChangeTask) {
TxTimeoutOnPartitionMapExchangeChangeTask task0 = (TxTimeoutOnPartitionMapExchangeChangeTask)task;
- processTxTimeoutOnPartitionMapExchangeChange(task0.message());
+ sharedCtx.tm().processTxTimeoutOnPartitionMapExchangeChange(task0.message());
}
else if (task instanceof StopCachesOnClientReconnectExchangeTask) {
StopCachesOnClientReconnectExchangeTask task0 = (StopCachesOnClientReconnectExchangeTask)task;
@@ -667,6 +636,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (ctx.isDaemon())
return;
+ AffinityTopologyVersion joinVer;
+
try {
boolean checkConsistency = !getBoolean(IGNITE_SKIP_CONFIGURATION_CONSISTENCY_CHECK);
@@ -679,14 +650,15 @@ public class GridCacheProcessor extends GridProcessorAdapter {
ctx.query().onCacheKernalStart();
- sharedCtx.exchange().onKernalStart(active, false);
+ joinVer = sharedCtx.exchange().onKernalStart(active, false);
}
finally {
cacheStartedLatch.countDown();
}
if (!ctx.clientNode())
- addRemovedItemsCleanupTask(Long.getLong(IGNITE_CACHE_REMOVED_ENTRIES_TTL, 10_000));
+ sharedCtx.time().addTimeoutObject(new PartitionDefferedDeleteQueueCleanupTask(
+ sharedCtx, Long.getLong(IGNITE_CACHE_REMOVED_ENTRIES_TTL, 10_000)));
// Notify shared managers.
for (GridCacheSharedManager mgr : sharedCtx.managers())
@@ -699,35 +671,34 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (ctx.service() instanceof GridServiceProcessor)
((GridServiceProcessor)ctx.service()).onUtilityCacheStarted();
- final AffinityTopologyVersion startTopVer = ctx.discovery().localJoin().joinTopologyVersion();
-
- final List<IgniteInternalFuture> syncFuts = new ArrayList<>(caches.size());
-
- sharedCtx.forAllCaches(new CIX1<GridCacheContext>() {
- @Override public void applyx(GridCacheContext cctx) {
- CacheConfiguration cfg = cctx.config();
-
- if (cctx.affinityNode() &&
- cfg.getRebalanceMode() == SYNC &&
- startTopVer.equals(cctx.startTopologyVersion())) {
- CacheMode cacheMode = cfg.getCacheMode();
-
- if (cacheMode == REPLICATED || (cacheMode == PARTITIONED && cfg.getRebalanceDelay() >= 0))
- // Need to wait outside to avoid a deadlock
- syncFuts.add(cctx.preloader().syncFuture());
- }
- }
- });
+ awaitRebalance(joinVer).get();
+ }
- for (int i = 0, size = syncFuts.size(); i < size; i++)
- syncFuts.get(i).get();
+ /**
+ * Await for local join finish and caches start.
+ *
+ * @throws IgniteCheckedException If failed to wait.
+ */
+ public void awaitStarted() throws IgniteCheckedException {
+ U.await(cacheStartedLatch);
}
/**
- * @param timeout Cleanup timeout.
+ * Await rebalance for caches with SYNC rebalance mode started on local join exchange.
+ *
+ * @param joinVer Topology version of local join.
+ * @return Future indicates that rebalance for SYNC rebalance mode caches has completed.
*/
- private void addRemovedItemsCleanupTask(long timeout) {
- ctx.timeout().addTimeoutObject(new RemovedItemsCleanupTask(timeout));
+ private GridCompoundFuture<?, ?> awaitRebalance(AffinityTopologyVersion joinVer) {
+ return internalCaches().stream()
+ .map(GridCacheAdapter::context)
+ .filter(GridCacheContext::affinityNode) // Only affinity caches.
+ .filter(ctx -> ctx.config().getRebalanceMode() == SYNC) // Only caches with SYNC mode.
+ .filter(ctx -> ctx.startTopologyVersion().equals(joinVer)) // Only caches started on local join.
+ .filter(ctx -> ctx.config().getCacheMode() == REPLICATED // Caches without manual rebalance.
+ || ctx.config().getCacheMode() == PARTITIONED && ctx.config().getRebalanceDelay() >= 0)
+ .map(ctx -> ctx.preloader().syncFuture())
+ .collect(IgniteCollectors.toCompoundFuture());
}
/** {@inheritDoc} */
@@ -879,9 +850,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
for (EnableStatisticsFuture fut : manageStatisticsFuts.values())
fut.onDone(err);
- for (TxTimeoutOnPartitionMapExchangeChangeFuture fut : txTimeoutOnPartitionMapExchangeFuts.values())
- fut.onDone(err);
-
for (CacheGroupContext grp : cacheGrps.values())
grp.onDisconnected(reconnectFut);
@@ -982,20 +950,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * Initialize query infrastructure for not started cache.
- *
- * @param cacheDesc Cache descriptor.
- * @throws IgniteCheckedException If failed.
- */
- public void initQueryStructuresForNotStartedCache(DynamicCacheDescriptor cacheDesc) throws IgniteCheckedException {
- QuerySchema schema = cacheDesc.schema() != null ? cacheDesc.schema() : new QuerySchema();
-
- GridCacheContextInfo cacheInfo = new GridCacheContextInfo(cacheDesc);
-
- ctx.query().onCacheStart(cacheInfo, schema, cacheDesc.sql());
- }
-
- /**
* @param cache Cache to stop.
* @param cancel Cancel flag.
* @param destroy Destroy data flag. Setting to <code>true</code> will remove all cache data.
@@ -1109,13 +1063,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * @throws IgniteCheckedException If failed to wait.
- */
- public void awaitStarted() throws IgniteCheckedException {
- U.await(cacheStartedLatch);
- }
-
- /**
* @param cache Cache.
* @throws IgniteCheckedException If failed.
*/
@@ -1274,7 +1221,9 @@ public class GridCacheProcessor extends GridProcessorAdapter {
CacheEvictionManager evictMgr = (nearEnabled || cfg.isOnheapCacheEnabled())
? new GridCacheEvictionManager()
: new CacheOffheapEvictionManager();
- GridCacheQueryManager qryMgr = queryManager(cfg);
+ GridCacheQueryManager qryMgr = cfg.getCacheMode() == LOCAL
+ ? new GridCacheLocalQueryManager()
+ : new GridCacheDistributedQueryManager();
CacheContinuousQueryManager contQryMgr = new CacheContinuousQueryManager();
CacheDataStructuresManager dataStructuresMgr = new CacheDataStructuresManager();
GridCacheTtlManager ttlMgr = new GridCacheTtlManager();
@@ -1556,13 +1505,9 @@ public class GridCacheProcessor extends GridProcessorAdapter {
) {
IgniteCacheProxyImpl<?, ?> proxy = jCacheProxies.get(cache.name());
- boolean canRestart = true;
-
- DynamicCacheChangeRequest req = reqs.get(cache.name());
-
- if (req != null) {
- canRestart = !req.disabledAfterStart();
- }
+ boolean canRestart = Optional.ofNullable(reqs.get(cache.name()))
+ .map(req -> !req.disabledAfterStart())
+ .orElse(true);
if (proxy != null && proxy.isRestarting() && canRestart) {
proxy.onRestarted(cacheCtx, cache);
@@ -1614,16 +1559,14 @@ public class GridCacheProcessor extends GridProcessorAdapter {
for (Map.Entry<String, DynamicCacheDescriptor> e : cachesInfo.registeredCaches().entrySet()) {
DynamicCacheDescriptor desc = e.getValue();
- if (!desc.cacheType().userCache())
+ if (!desc.cacheType().userCache() || desc.cacheConfiguration().getCacheMode() == LOCAL)
continue;
CacheConfiguration ccfg = desc.cacheConfiguration();
- if (ccfg.getCacheMode() != LOCAL) {
- dynamicStartCache(null, ccfg.getName(), null, false, true, true).get();
+ dynamicStartCache(null, ccfg.getName(), null, false, true, true).get();
- return publicJCache(ccfg.getName());
- }
+ return publicJCache(ccfg.getName());
}
}
@@ -1688,9 +1631,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
* @return Caches to be started when this node starts.
*/
@Nullable public LocalJoinCachesContext localJoinCachesContext() {
- if (ctx.discovery().localNode().order() == 1 && alreadyFiltered.compareAndSet(false, true)) {
+ if (ctx.discovery().localNode().order() == 1)
cachesInfo.filterDynamicCacheDescriptors(locCfgMgr.localCachesOnStart());
- }
return cachesInfo.localJoinCachesContext();
}
@@ -1719,7 +1661,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
locJoinCtx.initCaches()
.forEach(cacheDesc -> {
try {
- initQueryStructuresForNotStartedCache(cacheDesc);
+ ctx.query().initQueryStructuresForNotStartedCache(cacheDesc);
}
catch (Exception e) {
log.error("Can't initialize query structures for not started cache [cacheName=" + cacheDesc.cacheName() + "]");
@@ -1793,7 +1735,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
* @throws IgniteCheckedException if failed.
*/
Map<StartCacheInfo, IgniteCheckedException> prepareStartCachesIfPossible(
- Collection<StartCacheInfo> startCacheInfos) throws IgniteCheckedException {
+ Collection<StartCacheInfo> startCacheInfos
+ ) throws IgniteCheckedException {
HashMap<StartCacheInfo, IgniteCheckedException> failedCaches = new HashMap<>();
prepareStartCaches(startCacheInfos, (data, operation) -> {
@@ -3245,43 +3188,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * Callback invoked from discovery thread when discovery custom message is received.
- *
- * @param msg Discovery message for changing transaction timeout on partition map exchange.
- */
- public void onTxTimeoutOnPartitionMapExchangeChange(TxTimeoutOnPartitionMapExchangeChangeMessage msg) {
- assert msg != null;
-
- if (msg.isInit()) {
- TransactionConfiguration cfg = ctx.config().getTransactionConfiguration();
-
- if (cfg.getTxTimeoutOnPartitionMapExchange() != msg.getTimeout())
- cfg.setTxTimeoutOnPartitionMapExchange(msg.getTimeout());
- }
- else {
- TxTimeoutOnPartitionMapExchangeChangeFuture fut = txTimeoutOnPartitionMapExchangeFuts.get(
- msg.getRequestId());
-
- if (fut != null)
- fut.onDone();
- }
- }
-
- /**
- * The task for changing transaction timeout on partition map exchange processed by exchange worker.
- *
- * @param msg Message.
- */
- public void processTxTimeoutOnPartitionMapExchangeChange(TxTimeoutOnPartitionMapExchangeChangeMessage msg) {
- assert msg != null;
-
- long timeout = ctx.config().getTransactionConfiguration().getTxTimeoutOnPartitionMapExchange();
-
- if (timeout != msg.getTimeout())
- ctx.config().getTransactionConfiguration().setTxTimeoutOnPartitionMapExchange(msg.getTimeout());
- }
-
- /**
* @param stoppedCaches Stopped caches.
*/
private void stopCachesOnClientReconnect(Collection<GridCacheAdapter> stoppedCaches) {
@@ -3525,8 +3431,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
assert cacheName != null;
if (checkThreadTx) {
- checkEmptyTransactionsEx(() -> String.format(CACHE_NAME_AND_OPERATION_FORMAT, cacheName,
- "dynamicStartCache"));
+ sharedCtx.tm().checkEmptyTransactions(
+ () -> String.format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheName, "dynamicStartCache"));
}
GridPlainClosure2<Collection<byte[]>, byte[], IgniteInternalFuture<Boolean>> startCacheClsr =
@@ -3680,13 +3586,13 @@ public class GridCacheProcessor extends GridProcessorAdapter {
IgniteUuid restartId
) {
if (checkThreadTx) {
- checkEmptyTransactionsEx(() -> {
+ sharedCtx.tm().checkEmptyTransactions(() -> {
List<String> cacheNames = storedCacheDataList.stream()
.map(StoredCacheData::config)
.map(CacheConfiguration::getName)
.collect(Collectors.toList());
- return String.format(CACHE_NAMES_AND_OPERATION_FORMAT, cacheNames, "dynamicStartCachesByStoredConf");
+ return String.format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheNames, "dynamicStartCachesByStoredConf");
});
}
@@ -3793,8 +3699,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
assert cacheName != null;
if (checkThreadTx) {
- checkEmptyTransactionsEx(() -> String.format(CACHE_NAME_AND_OPERATION_FORMAT, cacheName,
- "dynamicDestroyCache"));
+ sharedCtx.tm().checkEmptyTransactions(
+ () -> String.format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheName, "dynamicDestroyCache"));
}
DynamicCacheChangeRequest req = DynamicCacheChangeRequest.stopRequest(ctx, cacheName, sql, true);
@@ -3828,17 +3734,14 @@ public class GridCacheProcessor extends GridProcessorAdapter {
boolean destroy
) {
if (checkThreadTx) {
- checkEmptyTransactionsEx(() -> String.format(CACHE_NAMES_AND_OPERATION_FORMAT, cacheNames,
- "dynamicDestroyCaches"));
+ sharedCtx.tm().checkEmptyTransactions(
+ () -> String.format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheNames, "dynamicDestroyCaches"));
}
- List<DynamicCacheChangeRequest> reqs = new ArrayList<>(cacheNames.size());
-
- for (String cacheName : cacheNames) {
- reqs.add(createStopRequest(cacheName, false, null, destroy));
- }
-
- return dynamicChangeCaches(reqs);
+ return dynamicChangeCaches(
+ cacheNames.stream().map(cacheName -> createStopRequest(cacheName, false, null, destroy))
+ .collect(Collectors.toList())
+ );
}
/**
@@ -3867,41 +3770,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
* @param reqs cache stop requests.
* @return compound future.
*/
- public @NotNull IgniteInternalFuture<?> dynamicChangeCaches(List<DynamicCacheChangeRequest> reqs) {
- GridCompoundFuture<?, ?> compoundFut = new GridCompoundFuture<>();
-
- for (DynamicCacheStartFuture fut : initiateCacheChanges(reqs))
- compoundFut.add((IgniteInternalFuture)fut);
-
- compoundFut.markInitialized();
-
- return compoundFut;
- }
-
- /**
- * Change WAL mode.
- *
- * @param cacheNames Cache names.
- * @param enabled Enabled flag.
- * @return Future completed when operation finished.
- */
- public IgniteInternalFuture<Boolean> changeWalMode(Collection<String> cacheNames, boolean enabled) {
- if (transactions().tx() != null || sharedCtx.lockedTopologyVersion(null) != null)
- throw new IgniteException("Cache WAL mode cannot be changed within lock or transaction.");
-
- return sharedCtx.walState().init(cacheNames, enabled);
- }
-
- /**
- * @param cacheName Cache name.
- */
- public boolean walEnabled(String cacheName) {
- DynamicCacheDescriptor desc = ctx.cache().cacheDescriptor(cacheName);
-
- if (desc == null)
- throw new IgniteException("Cache not found: " + cacheName);
-
- return desc.groupDescriptor().walEnabled();
+ @NotNull public IgniteInternalFuture<?> dynamicChangeCaches(List<DynamicCacheChangeRequest> reqs) {
+ return initiateCacheChanges(reqs).stream().collect(IgniteCollectors.toCompoundFuture());
}
/**
@@ -3916,7 +3786,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
if (proxy == null || proxy.isProxyClosed())
return new GridFinishedFuture<>(); // No-op.
- checkEmptyTransactionsEx(() -> String.format(CACHE_NAME_AND_OPERATION_FORMAT, cacheName, "dynamicCloseCache"));
+ sharedCtx.tm().checkEmptyTransactions(
+ () -> String.format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheName, "dynamicCloseCache"));
if (proxy.context().isLocal())
return dynamicDestroyCache(cacheName, false, true, false, null);
@@ -3931,13 +3802,8 @@ public class GridCacheProcessor extends GridProcessorAdapter {
* @return Future that will be completed when state is changed for all caches.
*/
public IgniteInternalFuture<?> resetCacheState(Collection<String> cacheNames) {
- if (F.isEmpty(cacheNames))
- cacheNames = cachesInfo.registeredCaches().keySet();
-
- Collection<String> forCheckCacheNames = cacheNames;
-
- checkEmptyTransactionsEx(() -> String.format(CACHE_NAME_AND_OPERATION_FORMAT, forCheckCacheNames,
- "resetCacheState"));
+ sharedCtx.tm().checkEmptyTransactions(
+ () -> String.format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheNames, "resetCacheState"));
Collection<DynamicCacheChangeRequest> reqs = new ArrayList<>(cacheNames.size());
@@ -3955,14 +3821,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
reqs.add(req);
}
- GridCompoundFuture fut = new GridCompoundFuture();
-
- for (DynamicCacheStartFuture f : initiateCacheChanges(reqs))
- fut.add(f);
-
- fut.markInitialized();
-
- return fut;
+ return initiateCacheChanges(reqs).stream().collect(IgniteCollectors.toCompoundFuture());
}
/**
@@ -4028,7 +3887,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
Collection<DynamicCacheChangeRequest> sndReqs = new ArrayList<>(reqs.size());
for (DynamicCacheChangeRequest req : reqs) {
- authorizeCacheChange(req);
+ authorizeCacheChange(ctx.security(), req);
DynamicCacheStartFuture fut = new DynamicCacheStartFuture(req.requestId());
@@ -4095,33 +3954,45 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * Authorize creating cache.
+ * Authorizes cache change request.
*
- * @param cfg Cache configuration.
- * @param ctx Kernal context.
+ * @param security Security.
+ * @param req Cache change request.
*/
- static void authorizeCacheCreate(CacheConfiguration cfg, GridKernalContext ctx) {
- if(cfg != null) {
- ctx.security().authorize(cfg.getName(), SecurityPermission.CACHE_CREATE);
-
- if (cfg.isOnheapCacheEnabled() &&
- IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_DISABLE_ONHEAP_CACHE))
- throw new SecurityException("Authorization failed for enabling on-heap cache.");
+ static void authorizeCacheChange(IgniteSecurity security, DynamicCacheChangeRequest req) {
+ if (req.cacheType() == null || req.cacheType() == CacheType.USER) {
+ if (req.start())
+ authorizeCacheCreate(security, req.startCacheConfiguration());
+ else if (req.stop())
+ authorizeCacheDestroy(security, req.cacheName());
}
}
/**
- * Authorize dynamic cache management.
+ * Authorizes cache destroy.
*
- * @param req start/stop cache request.
+ * @param security Security.
+ * @param cacheName Cache name.
*/
- private void authorizeCacheChange(DynamicCacheChangeRequest req) {
- if (req.cacheType() == null || req.cacheType() == CacheType.USER) {
- if (req.stop())
- ctx.security().authorize(req.cacheName(), SecurityPermission.CACHE_DESTROY);
- else
- authorizeCacheCreate(req.startCacheConfiguration(), ctx);
- }
+ static void authorizeCacheDestroy(IgniteSecurity security, String cacheName) {
+ security.authorize(cacheName, SecurityPermission.CACHE_DESTROY);
+ }
+
+ /**
+ * Authorizes cache create.
+ *
+ * @param security Security.
+ * @param cacheCfg Cache configuration.
+ */
+ static void authorizeCacheCreate(IgniteSecurity security, @Nullable CacheConfiguration cacheCfg) {
+ if (cacheCfg == null)
+ return;
+
+ security.authorize(cacheCfg.getName(), SecurityPermission.CACHE_CREATE);
+
+ if (cacheCfg.isOnheapCacheEnabled() &&
+ IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_DISABLE_ONHEAP_CACHE))
+ throw new SecurityException("Authorization failed for enabling on-heap cache.");
}
/**
@@ -4147,11 +4018,13 @@ public class GridCacheProcessor extends GridProcessorAdapter {
* @param topVer Topology version.
* @param state Cluster state.
*/
- public void onDiscoveryEvent(int type,
+ public void onDiscoveryEvent(
+ int type,
@Nullable DiscoveryCustomMessage customMsg,
ClusterNode node,
AffinityTopologyVersion topVer,
- DiscoveryDataClusterState state) {
+ DiscoveryDataClusterState state
+ ) {
cachesInfo.onDiscoveryEvent(type, node, topVer);
sharedCtx.affinity().onDiscoveryEvent(type, customMsg, node, topVer, state);
@@ -4211,7 +4084,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
onCacheStatisticsClear((CacheStatisticsClearMessage)msg);
if (msg instanceof TxTimeoutOnPartitionMapExchangeChangeMessage)
- onTxTimeoutOnPartitionMapExchangeChange((TxTimeoutOnPartitionMapExchangeChangeMessage)msg);
+ sharedCtx.tm().onTxTimeoutOnPartitionMapExchangeChange((TxTimeoutOnPartitionMapExchangeChangeMessage)msg);
return false;
}
@@ -4250,14 +4123,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * @param cfg Cache configuration.
- * @return Query manager.
- */
- private GridCacheQueryManager queryManager(CacheConfiguration cfg) {
- return cfg.getCacheMode() == LOCAL ? new GridCacheLocalQueryManager() : new GridCacheDistributedQueryManager();
- }
-
- /**
* @return Keep static cache configuration flag. If {@code true}, static cache configuration will override
* configuration persisted on disk.
*/
@@ -4748,9 +4613,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
for (EnableStatisticsFuture fut : manageStatisticsFuts.values())
fut.onDone(err);
-
- for (TxTimeoutOnPartitionMapExchangeChangeFuture fut : txTimeoutOnPartitionMapExchangeFuts.values())
- fut.onDone(err);
}
/**
@@ -4927,33 +4789,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * Method checks that current thread does not have active transactions.
- *
- * @throws IgniteException If transaction exist.
- */
- public void checkEmptyTransactions() throws IgniteException {
- if (transactions().tx() != null || sharedCtx.lockedTopologyVersion(null) != null)
- throw new IgniteException("Cannot start/stop cache within lock or transaction.");
- }
-
- /**
- * Method invoke {@link #checkEmptyTransactions()} and add message in case exception.
- *
- * @param eMsgSupplier supplier additional text message
- * @throws IgniteException If {@link #checkEmptyTransactions()} throw {@link IgniteException}
- * */
- private void checkEmptyTransactionsEx(final Supplier<String> eMsgSupplier) throws IgniteException {
- assert eMsgSupplier != null;
-
- try {
- checkEmptyTransactions();
- }
- catch (IgniteException e) {
- throw new IgniteException(e.getMessage() + ' ' + eMsgSupplier.get(), e);
- }
- }
-
- /**
* @param val Object to check.
* @return Configuration copy.
* @throws IgniteCheckedException If validation failed.
@@ -5245,26 +5080,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * Sets transaction timeout on partition map exchange.
- *
- * @param timeout Transaction timeout on partition map exchange in milliseconds.
- */
- public void setTxTimeoutOnPartitionMapExchange(long timeout) throws IgniteCheckedException {
- UUID requestId = UUID.randomUUID();
-
- TxTimeoutOnPartitionMapExchangeChangeFuture fut = new TxTimeoutOnPartitionMapExchangeChangeFuture(requestId);
-
- txTimeoutOnPartitionMapExchangeFuts.put(requestId, fut);
-
- TxTimeoutOnPartitionMapExchangeChangeMessage msg = new TxTimeoutOnPartitionMapExchangeChangeMessage(
- requestId, timeout);
-
- ctx.grid().context().discovery().sendCustomEvent(msg);
-
- fut.get();
- }
-
- /**
* @param obj Object to clone.
* @return Object copy.
* @throws IgniteCheckedException If failed.
@@ -5278,103 +5093,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- * Get Temporary storage
- */
- public MetaStorage.TmpStorage getTmpStorage() {
- return tmpStorage;
- }
-
- /**
- * Set Temporary storage
- */
- public void setTmpStorage(MetaStorage.TmpStorage tmpStorage) {
- this.tmpStorage = tmpStorage;
- }
-
- /**
- * Sets if dump requests from local node to near node are allowed, when long running transaction
- * is found. If allowed, the compute request to near node will be made to get thread dump of transaction
- * owner thread. Also broadcasts this setting on other server nodes in cluster.
- *
- * @param allowed whether allowed
- */
- public void setTxOwnerDumpRequestsAllowed(boolean allowed) {
- ClusterGroup grp = ctx.grid()
- .cluster()
- .forServers()
- .forPredicate(node -> IgniteFeatures.nodeSupports(node, TRANSACTION_OWNER_THREAD_DUMP_PROVIDING));
-
- IgniteCompute compute = ctx.grid().compute(grp);
-
- compute.broadcast(new TxOwnerDumpRequestAllowedSettingClosure(allowed));
- }
-
- /**
- * Sets threshold timeout in milliseconds for long transactions, if transaction exceeds it,
- * it will be dumped in log with information about how much time did
- * it spent in system time (time while aquiring locks, preparing, commiting, etc.)
- * and user time (time when client node runs some code while holding transaction).
- * Can be set to 0 - no transactions will be dumped in log in this case.
- *
- * @param threshold Threshold timeout in milliseconds.
- */
- public void longTransactionTimeDumpThreshold(long threshold) {
- assert threshold >= 0 : "Threshold timeout must be greater than or equal to 0.";
-
- broadcastToNodesSupportingFeature(
- new LongRunningTxTimeDumpSettingsClosure(threshold, null, null),
- LRT_SYSTEM_USER_TIME_DUMP_SETTINGS
- );
- }
-
- /**
- * Sets the coefficient for samples of long running transactions that will be dumped in log, if
- * {@link #longTransactionTimeDumpThreshold} is set to non-zero value."
- *
- * @param coefficient Coefficient, must be value between 0.0 and 1.0 inclusively.
- */
- public void transactionTimeDumpSamplesCoefficient(double coefficient) {
- assert coefficient >= 0.0 && coefficient <= 1.0 : "Percentage value must be between 0.0 and 1.0 inclusively.";
-
- broadcastToNodesSupportingFeature(
- new LongRunningTxTimeDumpSettingsClosure(null, coefficient, null),
- LRT_SYSTEM_USER_TIME_DUMP_SETTINGS
- );
- }
-
- /**
- * Sets the limit of samples of completed transactions that will be dumped in log per second,
- * if {@link #transactionTimeDumpSamplesCoefficient} is above <code>0.0</code>.
- * Must be integer value greater than <code>0</code>.
- *
- * @param limit Limit value.
- */
- public void longTransactionTimeDumpSamplesPerSecondLimit(int limit) {
- assert limit > 0 : "Limit value must be greater than 0.";
-
- broadcastToNodesSupportingFeature(
- new LongRunningTxTimeDumpSettingsClosure(null, null, limit),
- LRT_SYSTEM_USER_TIME_DUMP_SETTINGS
- );
- }
-
- /**
- * Broadcasts given job to nodes that support ignite feature.
- *
- * @param job Ignite job.
- * @param feature Ignite feature.
- */
- private void broadcastToNodesSupportingFeature(IgniteRunnable job, IgniteFeatures feature) {
- ClusterGroup grp = ctx.grid()
- .cluster()
- .forPredicate(node -> IgniteFeatures.nodeSupports(node, feature));
-
- IgniteCompute compute = ctx.grid().compute(grp);
-
- compute.broadcast(job);
- }
-
- /**
* @param oldFormat Old format.
*/
private CacheConfigurationSplitter splitter(boolean oldFormat) {
@@ -5674,128 +5392,6 @@ public class GridCacheProcessor extends GridProcessorAdapter {
}
/**
- *
- */
- static class LocalAffinityFunction implements AffinityFunction {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** {@inheritDoc} */
- @Override public List<List<ClusterNode>> assignPartitions(AffinityFunctionContext affCtx) {
- ClusterNode locNode = null;
-
- for (ClusterNode n : affCtx.currentTopologySnapshot()) {
- if (n.isLocal()) {
- locNode = n;
-
- break;
- }
- }
-
- if (locNode == null)
- throw new IgniteException("Local node is not included into affinity nodes for 'LOCAL' cache");
-
- List<List<ClusterNode>> res = new ArrayList<>(partitions());
-
- for (int part = 0; part < partitions(); part++)
- res.add(Collections.singletonList(locNode));
-
- return Collections.unmodifiableList(res);
- }
-
- /** {@inheritDoc} */
- @Override public void reset() {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public int partitions() {
- return 1;
- }
-
- /** {@inheritDoc} */
- @Override public int partition(Object key) {
- return 0;
- }
-
- /** {@inheritDoc} */
- @Override public void removeNode(UUID nodeId) {
- // No-op.
- }
- }
-
- /**
- *
- */
- private class RemovedItemsCleanupTask implements GridTimeoutObject {
- /** */
- private final IgniteUuid id = IgniteUuid.randomUuid();
-
- /** */
- private final long endTime;
-
- /** */
- private final long timeout;
-
- /**
- * @param timeout Timeout.
- */
- RemovedItemsCleanupTask(long timeout) {
- this.timeout = timeout;
-
- endTime = U.currentTimeMillis() + timeout;
- }
-
- /** {@inheritDoc} */
- @Override public IgniteUuid timeoutId() {
- return id;
- }
-
- /** {@inheritDoc} */
- @Override public long endTime() {
- return endTime;
- }
-
- /** {@inheritDoc} */
- @Override public void onTimeout() {
- ctx.closure().runLocalSafe(new Runnable() {
- @Override public void run() {
- try {
- for (CacheGroupContext grp : sharedCtx.cache().cacheGroups()) {
- if (!grp.isLocal() && grp.affinityNode()) {
- GridDhtPartitionTopology top = null;
-
- try {
- top = grp.topology();
- }
- catch (IllegalStateException ignore) {
- // Cache stopped.
- }
-
- if (top != null) {
- for (GridDhtLocalPartition part : top.currentLocalPartitions())
- part.cleanupRemoveQueue();
- }
-
- if (ctx.isStopping())
- return;
- }
- }
- }
- catch (Exception e) {
- U.error(log, "Failed to cleanup removed cache items: " + e, e);
- }
-
- if (ctx.isStopping())
- return;
-
- addRemovedItemsCleanupTask(timeout);
- }
- }, true);
- }
- }
-
- /**
* Enable statistics future.
*/
private class EnableStatisticsFuture extends GridFutureAdapter<Void> {
@@ -5822,30 +5418,4 @@ public class GridCacheProcessor extends GridProcessorAdapter {
return S.toString(EnableStatisticsFuture.class, this);
}
}
-
- /**
- * The future for changing transaction timeout on partition map exchange.
- */
- private class TxTimeoutOnPartitionMapExchangeChangeFuture extends GridFutureAdapter<Void> {
- /** */
- private UUID id;
-
- /**
- * @param id Future ID.
- */
- private TxTimeoutOnPartitionMapExchangeChangeFuture(UUID id) {
- this.id = id;
- }
-
- /** {@inheritDoc} */
- @Override public boolean onDone(@Nullable Void res, @Nullable Throwable err) {
- txTimeoutOnPartitionMapExchangeFuts.remove(id, this);
- return super.onDone(res, err);
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(TxTimeoutOnPartitionMapExchangeChangeFuture.class, this);
- }
- }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
index ce9af34..e15cc0f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheSharedContext.java
@@ -368,7 +368,15 @@ public class GridCacheSharedContext<K, V> {
this.rebalanceEnabled = rebalanceEnabled;
if (rebalanceEnabled)
- cache().enableRebalance();
+ enableRebalance();
+ }
+
+ /**
+ * Gets all caches having cache proxy and starts on them force rebalance.
+ */
+ private void enableRebalance() {
+ for (IgniteCacheProxy c : cache().publicCaches())
+ c.rebalance();
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
index d9ff104..943a531 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
@@ -67,6 +67,7 @@ import org.apache.ignite.internal.cluster.ClusterGroupEmptyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.affinity.LocalAffinityFunction;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockCancelledException;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
@@ -1774,11 +1775,11 @@ public class GridCacheUtils {
cfg.setBackups(Integer.MAX_VALUE);
}
else
- cfg.setAffinity(new GridCacheProcessor.LocalAffinityFunction());
+ cfg.setAffinity(new LocalAffinityFunction());
}
else {
- if (cfg.getCacheMode() == LOCAL && !(cfg.getAffinity() instanceof GridCacheProcessor.LocalAffinityFunction)) {
- cfg.setAffinity(new GridCacheProcessor.LocalAffinityFunction());
+ if (cfg.getCacheMode() == LOCAL && !(cfg.getAffinity() instanceof LocalAffinityFunction)) {
+ cfg.setAffinity(new LocalAffinityFunction());
U.warn(log, "AffinityFunction configuration parameter will be ignored for local cache" +
" [cacheName=" + U.maskName(cfg.getName()) + ']');
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
index 9916b3a..78e8b3f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ValidationOnNodeJoinUtils.java
@@ -50,6 +50,7 @@ import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteFeatures;
import org.apache.ignite.internal.IgniteNodeAttributes;
import org.apache.ignite.internal.binary.BinaryMarshaller;
+import org.apache.ignite.internal.processors.affinity.LocalAffinityFunction;
import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
import org.apache.ignite.internal.processors.datastructures.DataStructuresProcessor;
import org.apache.ignite.internal.processors.query.QuerySchemaPatch;
@@ -156,7 +157,7 @@ public class ValidationOnNodeJoinUtils {
for (CacheJoinNodeDiscoveryData.CacheInfo cacheInfo : nodeData.caches().values()) {
if (secCtx != null && cacheInfo.cacheType() == CacheType.USER) {
try (OperationSecurityContext s = ctx.security().withContext(secCtx)) {
- GridCacheProcessor.authorizeCacheCreate(cacheInfo.cacheData().config(), ctx);
+ GridCacheProcessor.authorizeCacheCreate(ctx.security(), cacheInfo.cacheData().config());
}
catch (SecurityException ex) {
if (errorMsg.length() > 0)
@@ -241,7 +242,7 @@ public class ValidationOnNodeJoinUtils {
throw new IgniteCheckedException("DataRegion for client caches must be explicitly configured " +
"on client node startup. Use DataStorageConfiguration to configure DataRegion.");
- if (cc.getCacheMode() == LOCAL && !cc.getAffinity().getClass().equals(GridCacheProcessor.LocalAffinityFunction.class))
+ if (cc.getCacheMode() == LOCAL && !cc.getAffinity().getClass().equals(LocalAffinityFunction.class))
U.warn(log, "AffinityFunction configuration parameter will be ignored for local cache [cacheName=" +
U.maskName(cc.getName()) + ']');
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/WalStateManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/WalStateManager.java
index 044b702..eba9d0b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/WalStateManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/WalStateManager.java
@@ -291,13 +291,28 @@ public class WalStateManager extends GridCacheSharedManagerAdapter {
}
/**
+ * Change WAL mode.
+ *
+ * @param cacheNames Cache names.
+ * @param enabled Enabled flag.
+ * @return Future completed when operation finished.
+ */
+ public IgniteInternalFuture<Boolean> changeWalMode(Collection<String> cacheNames, boolean enabled) {
+ cctx.tm().checkEmptyTransactions(() ->
+ String.format("Cache WAL mode cannot be changed within lock or transaction " +
+ "[cacheNames=%s, walEnabled=%s]", cacheNames, enabled));
+
+ return init(cacheNames, enabled);
+ }
+
+ /**
* Initiate WAL mode change operation.
*
* @param cacheNames Cache names.
* @param enabled Enabled flag.
* @return Future completed when operation finished.
*/
- public IgniteInternalFuture<Boolean> init(Collection<String> cacheNames, boolean enabled) {
+ private IgniteInternalFuture<Boolean> init(Collection<String> cacheNames, boolean enabled) {
if (!enabled && prohibitDisabling)
return errorFuture("WAL disabling is prohibited.");
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
index d767bfb..ffca47e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
@@ -395,7 +395,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
/**
* TODO FIXME Get rid of deferred delete queue https://issues.apache.org/jira/browse/IGNITE-11704
*/
- public void cleanupRemoveQueue() {
+ void cleanupRemoveQueue() {
if (state() == MOVING) {
if (rmvQueue.sizex() >= rmvQueueMaxSize) {
LT.warn(log, "Deletion queue cleanup for moving partition was delayed until rebalance is finished. " +
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/PartitionDefferedDeleteQueueCleanupTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/PartitionDefferedDeleteQueueCleanupTask.java
new file mode 100644
index 0000000..65ecef4
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/PartitionDefferedDeleteQueueCleanupTask.java
@@ -0,0 +1,107 @@
+/*
+ * 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.distributed.dht.topology;
+
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
+import org.apache.ignite.internal.util.lang.GridPlainRunnable;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteUuid;
+
+/**
+ * Task to periodically clean partition deffered delete queue.
+ *
+ * @see GridDhtLocalPartition#cleanupRemoveQueue().
+ */
+public class PartitionDefferedDeleteQueueCleanupTask implements GridTimeoutObject {
+ /** */
+ private final IgniteUuid id = IgniteUuid.randomUuid();
+
+ /** */
+ private final long endTime;
+
+ /** */
+ private final long timeout;
+
+ /** */
+ private final GridCacheSharedContext cctx;
+
+ /** */
+ private final IgniteLogger log;
+
+ /**
+ * @param timeout Timeout.
+ */
+ public PartitionDefferedDeleteQueueCleanupTask(GridCacheSharedContext cctx, long timeout) {
+ this.timeout = timeout;
+ endTime = U.currentTimeMillis() + timeout;
+ this.cctx = cctx;
+ log = cctx.logger(getClass());
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteUuid timeoutId() {
+ return id;
+ }
+
+ /** {@inheritDoc} */
+ @Override public long endTime() {
+ return endTime;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onTimeout() {
+ cctx.kernalContext().closure().runLocalSafe(new GridPlainRunnable() {
+ @Override
+ public void run() {
+ try {
+ for (CacheGroupContext grp : cctx.cache().cacheGroups()) {
+ if (!grp.isLocal() && grp.affinityNode()) {
+ GridDhtPartitionTopology top = null;
+
+ try {
+ top = grp.topology();
+ } catch (IllegalStateException ignore) {
+ // Cache stopped.
+ }
+
+ if (top != null) {
+ for (GridDhtLocalPartition part : top.currentLocalPartitions())
+ part.cleanupRemoveQueue();
+ }
+
+ if (cctx.kernalContext().isStopping())
+ return;
+ }
+ }
+ }
+ catch (Exception e) {
+ U.error(log, "Failed to cleanup removed cache items: " + e, e);
+ }
+
+ if (cctx.kernalContext().isStopping())
+ return;
+
+ // Re-schedule task after finish.
+ cctx.time().addTimeoutObject(new PartitionDefferedDeleteQueueCleanupTask(cctx, timeout));
+ }
+ }, true);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
index 0c5c378..ea7af29 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
@@ -405,6 +405,9 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
*/
private MetaStorage metaStorage;
+ /** Temporary metastorage to migration of index partition. {@see IGNITE-8735}. */
+ private MetaStorage.TmpStorage tmpMetaStorage;
+
/** */
private List<MetastorageLifecycleListener> metastorageLifecycleLsnrs;
@@ -5354,6 +5357,20 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
return metaStorage;
}
+ /**
+ * @return Temporary metastorage to migration of index partition.
+ */
+ public MetaStorage.TmpStorage temporaryMetaStorage() {
+ return tmpMetaStorage;
+ }
+
+ /**
+ * @param tmpMetaStorage Temporary metastorage to migration of index partition.
+ */
+ public void temporaryMetaStorage(MetaStorage.TmpStorage tmpMetaStorage) {
+ this.tmpMetaStorage = tmpMetaStorage;
+ }
+
/** {@inheritDoc} */
@Override public void notifyMetaStorageSubscribersOnReadyForRead() throws IgniteCheckedException {
metastorageLifecycleLsnrs = cctx.kernalContext().internalSubscriptionProcessor().getMetastorageSubscribers();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetaStorage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetaStorage.java
index d56500a..c9e9ec4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetaStorage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/metastorage/MetaStorage.java
@@ -50,7 +50,6 @@ import org.apache.ignite.internal.pagemem.wal.WALPointer;
import org.apache.ignite.internal.pagemem.wal.record.MetastoreDataRecord;
import org.apache.ignite.internal.pagemem.wal.record.delta.MetaPageInitRecord;
import org.apache.ignite.internal.processors.cache.CacheDiagnosticManager;
-import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
import org.apache.ignite.internal.processors.cache.persistence.DataRegionMetricsImpl;
@@ -169,14 +168,12 @@ public class MetaStorage implements DbCheckpointListener, ReadWriteMetastorage {
initInternal(db);
if (!PRESERVE_LEGACY_METASTORAGE_PARTITION_ID) {
- GridCacheProcessor gcProcessor = cctx.kernalContext().cache();
-
if (partId == OLD_METASTORE_PARTITION)
- gcProcessor.setTmpStorage(copyDataToTmpStorage());
- else if (gcProcessor.getTmpStorage() != null) {
- restoreDataFromTmpStorage(gcProcessor.getTmpStorage());
+ db.temporaryMetaStorage(copyDataToTmpStorage());
+ else if (db.temporaryMetaStorage() != null) {
+ restoreDataFromTmpStorage(db.temporaryMetaStorage());
- gcProcessor.setTmpStorage(null);
+ db.temporaryMetaStorage(null);
db.addCheckpointListener(new DbCheckpointListener() {
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index f5e8d8b..2e2e9ca 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -38,6 +38,7 @@ import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.binary.BinaryObjectException;
import org.apache.ignite.cluster.ClusterGroup;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.TransactionConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.failure.FailureType;
@@ -65,6 +66,9 @@ import org.apache.ignite.internal.processors.cache.GridCacheSharedManagerAdapter
import org.apache.ignite.internal.processors.cache.GridCacheVersionedFuture;
import org.apache.ignite.internal.processors.cache.GridDeferredAckMessageSender;
import org.apache.ignite.internal.processors.cache.LongOperationsDumpSettingsClosure;
+import org.apache.ignite.internal.processors.cache.LongRunningTxTimeDumpSettingsClosure;
+import org.apache.ignite.internal.processors.cache.TxOwnerDumpRequestAllowedSettingClosure;
+import org.apache.ignite.internal.processors.cache.TxTimeoutOnPartitionMapExchangeChangeMessage;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheMappedVersion;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryFuture;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedLockCancelledException;
@@ -96,14 +100,17 @@ import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
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.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgniteReducer;
import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.spi.systemview.view.TransactionView;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
import org.apache.ignite.transactions.TransactionState;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentLinkedHashMap;
@@ -126,6 +133,8 @@ import static org.apache.ignite.events.EventType.EVT_TX_STARTED;
import static org.apache.ignite.internal.GridTopic.TOPIC_CACHE_COORDINATOR;
import static org.apache.ignite.internal.GridTopic.TOPIC_TX;
import static org.apache.ignite.internal.IgniteFeatures.LONG_OPERATIONS_DUMP_TIMEOUT;
+import static org.apache.ignite.internal.IgniteFeatures.LRT_SYSTEM_USER_TIME_DUMP_SETTINGS;
+import static org.apache.ignite.internal.IgniteFeatures.TRANSACTION_OWNER_THREAD_DUMP_PROVIDING;
import static org.apache.ignite.internal.IgniteKernal.DFLT_LONG_OPERATIONS_DUMP_TIMEOUT;
import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYSTEM_POOL;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.READ;
@@ -274,9 +283,18 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
*/
private volatile boolean suspendResumeForPessimisticSupported;
+ /** The futures for changing transaction timeout on partition map exchange. */
+ private ConcurrentMap<UUID, TxTimeoutOnPartitionMapExchangeChangeFuture> txTimeoutOnPartitionMapExchangeFuts =
+ new ConcurrentHashMap<>();
+
/** {@inheritDoc} */
@Override protected void onKernalStop0(boolean cancel) {
cctx.gridIO().removeMessageListener(TOPIC_TX);
+
+ Exception err = new IgniteCheckedException("Operation has been cancelled (node is stopping).");
+
+ for (TxTimeoutOnPartitionMapExchangeChangeFuture fut : txTimeoutOnPartitionMapExchangeFuts.values())
+ fut.onDone(err);
}
/** {@inheritDoc} */
@@ -493,6 +511,9 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
for (TxDeadlockFuture fut : deadlockDetectFuts.values())
fut.onDone(err);
+
+ for (TxTimeoutOnPartitionMapExchangeChangeFuture fut : txTimeoutOnPartitionMapExchangeFuts.values())
+ fut.onDone(err);
}
/**
@@ -2761,6 +2782,143 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
}
/**
+ * Sets transaction timeout on partition map exchange.
+ *
+ * @param timeout Transaction timeout on partition map exchange in milliseconds.
+ */
+ public void setTxTimeoutOnPartitionMapExchange(long timeout) throws IgniteCheckedException {
+ UUID reqId = UUID.randomUUID();
+
+ TxTimeoutOnPartitionMapExchangeChangeFuture fut = new TxTimeoutOnPartitionMapExchangeChangeFuture(reqId);
+
+ txTimeoutOnPartitionMapExchangeFuts.put(reqId, fut);
+
+ TxTimeoutOnPartitionMapExchangeChangeMessage msg = new TxTimeoutOnPartitionMapExchangeChangeMessage(
+ reqId, timeout);
+
+ cctx.discovery().sendCustomEvent(msg);
+
+ fut.get();
+ }
+
+ /**
+ * Callback invoked from discovery thread when discovery custom message is received.
+ *
+ * @param msg Discovery message for changing transaction timeout on partition map exchange.
+ */
+ public void onTxTimeoutOnPartitionMapExchangeChange(TxTimeoutOnPartitionMapExchangeChangeMessage msg) {
+ assert msg != null;
+
+ if (msg.isInit()) {
+ TransactionConfiguration cfg = cctx.kernalContext().config().getTransactionConfiguration();
+
+ if (cfg.getTxTimeoutOnPartitionMapExchange() != msg.getTimeout())
+ cfg.setTxTimeoutOnPartitionMapExchange(msg.getTimeout());
+ }
+ else {
+ TxTimeoutOnPartitionMapExchangeChangeFuture fut = txTimeoutOnPartitionMapExchangeFuts.get(
+ msg.getRequestId());
+
+ if (fut != null)
+ fut.onDone();
+ }
+ }
+
+ /**
+ * The task for changing transaction timeout on partition map exchange processed by exchange worker.
+ *
+ * @param msg Message.
+ */
+ public void processTxTimeoutOnPartitionMapExchangeChange(TxTimeoutOnPartitionMapExchangeChangeMessage msg) {
+ assert msg != null;
+
+ long timeout = cctx.kernalContext().config().getTransactionConfiguration().getTxTimeoutOnPartitionMapExchange();
+
+ if (timeout != msg.getTimeout())
+ cctx.kernalContext().config().getTransactionConfiguration().setTxTimeoutOnPartitionMapExchange(msg.getTimeout());
+ }
+
+ /**
+ * Method checks that current thread does not have active transactions.
+ *
+ * If transaction or topology lock is hold by current thread
+ * exception {@link IgniteException} with given {@code errMsgConstructor} message will be thrown.
+ *
+ * @param errMsgConstructor Error message constructor.
+ */
+ public void checkEmptyTransactions(@NotNull IgniteOutClosure<String> errMsgConstructor) {
+ if (userTx() != null || cctx.lockedTopologyVersion(null) != null)
+ throw new IgniteException(errMsgConstructor.apply());
+ }
+
+ /**
+ * Sets if dump requests from local node to near node are allowed, when long running transaction
+ * is found. If allowed, the compute request to near node will be made to get thread dump of transaction
+ * owner thread. Also broadcasts this setting on other server nodes in cluster.
+ *
+ * @param allowed whether allowed
+ */
+ public void setTxOwnerDumpRequestsAllowedDistributed(boolean allowed) {
+ ClusterGroup grp = cctx.kernalContext().grid()
+ .cluster()
+ .forServers()
+ .forPredicate(node -> IgniteFeatures.nodeSupports(node, TRANSACTION_OWNER_THREAD_DUMP_PROVIDING));
+
+ IgniteCompute compute = cctx.kernalContext().grid().compute(grp);
+
+ compute.broadcast(new TxOwnerDumpRequestAllowedSettingClosure(allowed));
+ }
+
+ /**
+ * Sets threshold timeout in milliseconds for long transactions, if transaction exceeds it,
+ * it will be dumped in log with information about how much time did
+ * it spent in system time (time while aquiring locks, preparing, commiting, etc.)
+ * and user time (time when client node runs some code while holding transaction).
+ * Can be set to 0 - no transactions will be dumped in log in this case.
+ *
+ * @param threshold Threshold timeout in milliseconds.
+ */
+ public void longTransactionTimeDumpThresholdDistributed(long threshold) {
+ assert threshold >= 0 : "Threshold timeout must be greater than or equal to 0.";
+
+ broadcastToNodesSupportingFeature(
+ new LongRunningTxTimeDumpSettingsClosure(threshold, null, null),
+ LRT_SYSTEM_USER_TIME_DUMP_SETTINGS
+ );
+ }
+
+ /**
+ * Sets the coefficient for samples of long running transactions that will be dumped in log, if
+ * {@link #longTransactionTimeDumpThreshold} is set to non-zero value."
+ *
+ * @param coefficient Coefficient, must be value between 0.0 and 1.0 inclusively.
+ */
+ public void transactionTimeDumpSamplesCoefficientDistributed(double coefficient) {
+ assert coefficient >= 0.0 && coefficient <= 1.0 : "Percentage value must be between 0.0 and 1.0 inclusively.";
+
+ broadcastToNodesSupportingFeature(
+ new LongRunningTxTimeDumpSettingsClosure(null, coefficient, null),
+ LRT_SYSTEM_USER_TIME_DUMP_SETTINGS
+ );
+ }
+
+ /**
+ * Sets the limit of samples of completed transactions that will be dumped in log per second,
+ * if {@link #transactionTimeDumpSamplesCoefficient} is above <code>0.0</code>.
+ * Must be integer value greater than <code>0</code>.
+ *
+ * @param limit Limit value.
+ */
+ public void longTransactionTimeDumpSamplesPerSecondLimit(int limit) {
+ assert limit > 0 : "Limit value must be greater than 0.";
+
+ broadcastToNodesSupportingFeature(
+ new LongRunningTxTimeDumpSettingsClosure(null, null, limit),
+ LRT_SYSTEM_USER_TIME_DUMP_SETTINGS
+ );
+ }
+
+ /**
* Broadcasts given job to nodes that support ignite feature.
*
* @param job Ignite job.
@@ -3201,4 +3359,30 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
skippedTxCntr.incrementAndGet();
}
}
+
+ /**
+ * The future for changing transaction timeout on partition map exchange.
+ */
+ private class TxTimeoutOnPartitionMapExchangeChangeFuture extends GridFutureAdapter<Void> {
+ /** */
+ private UUID id;
+
+ /**
+ * @param id Future ID.
+ */
+ private TxTimeoutOnPartitionMapExchangeChangeFuture(UUID id) {
+ this.id = id;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean onDone(@Nullable Void res, @Nullable Throwable err) {
+ txTimeoutOnPartitionMapExchangeFuts.remove(id, this);
+ return super.onDone(res, err);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(TxTimeoutOnPartitionMapExchangeChangeFuture.class, this);
+ }
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
index 01cc322..39fd48d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
@@ -889,6 +889,20 @@ public class GridQueryProcessor extends GridProcessorAdapter {
}
/**
+ * Initialize query infrastructure for not started cache.
+ *
+ * @param cacheDesc Cache descriptor.
+ * @throws IgniteCheckedException If failed.
+ */
+ public void initQueryStructuresForNotStartedCache(DynamicCacheDescriptor cacheDesc) throws IgniteCheckedException {
+ QuerySchema schema = cacheDesc.schema() != null ? cacheDesc.schema() : new QuerySchema();
+
+ GridCacheContextInfo cacheInfo = new GridCacheContextInfo(cacheDesc);
+
+ onCacheStart(cacheInfo, schema, cacheDesc.sql());
+ }
+
+ /**
* Handle cache start. Invoked either from GridCacheProcessor.onKernalStart() method or from exchange worker.
* When called for the first time, we initialize topology thus understanding whether current node is coordinator
* or not.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteCollectors.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteCollectors.java
new file mode 100644
index 0000000..3fd9956
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteCollectors.java
@@ -0,0 +1,55 @@
+/*
+ * 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.util;
+
+import java.util.stream.Collector;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.future.GridCompoundFuture;
+
+/**
+ * Custom collectors for {@link java.util.stream.Stream} API.
+ */
+public class IgniteCollectors {
+ /**
+ * Empty constructor.
+ */
+ private IgniteCollectors() { }
+
+ /**
+ * Collector of {@link IgniteInternalFuture} inheritors stream to {@link GridCompoundFuture}.
+ *
+ * @param <T> Result type of inheritor {@link IgniteInternalFuture}.
+ * @param <R> Result type of {@link GridCompoundFuture}.
+ * @return Compound future that contains all stream futures
+ * and initialized with {@link GridCompoundFuture#markInitialized()}.
+ */
+ public static <T, R> Collector<? super IgniteInternalFuture,
+ ? super GridCompoundFuture<T, R>, GridCompoundFuture<T, R>> toCompoundFuture() {
+ final GridCompoundFuture<T, R> res = new GridCompoundFuture<>();
+
+ return Collectors.collectingAndThen(
+ Collectors.reducing(
+ res,
+ res::add,
+ (a, b) -> a // No needs to merge compound futures.
+ ),
+ GridCompoundFuture::markInitialized
+ );
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
index 04e05d1..6eb3e4a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
@@ -220,7 +220,7 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> implements Ig
*
* @param fut Future to add.
*/
- public final void add(IgniteInternalFuture<T> fut) {
+ public final GridCompoundFuture<T, R> add(IgniteInternalFuture<T> fut) {
assert fut != null;
synchronized (this) {
@@ -248,6 +248,8 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> implements Ig
onDone(e);
}
}
+
+ return this;
}
/**
@@ -268,9 +270,11 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> implements Ig
/**
* Mark this future as initialized.
*/
- public final void markInitialized() {
+ public final GridCompoundFuture<T, R> markInitialized() {
if (FLAGS_UPD.compareAndSet(this, 0, INIT_FLAG))
checkComplete();
+
+ return this;
}
/**
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/affinity/local/LocalAffinityFunctionTest.java b/modules/core/src/test/java/org/apache/ignite/cache/affinity/local/LocalAffinityFunctionTest.java
index 2f93639..e9d9584 100755
--- a/modules/core/src/test/java/org/apache/ignite/cache/affinity/local/LocalAffinityFunctionTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/affinity/local/LocalAffinityFunctionTest.java
@@ -22,6 +22,7 @@ import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.processors.affinity.LocalAffinityFunction;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
@@ -60,8 +61,6 @@ public class LocalAffinityFunctionTest extends GridCommonAbstractTest {
CacheConfiguration ccf = node.cache(CACHE1).getConfiguration(CacheConfiguration.class);
- assertEquals("org.apache.ignite.internal.processors.cache.GridCacheProcessor$LocalAffinityFunction",
- ccf.getAffinity().getClass().getName());
+ assertEquals(LocalAffinityFunction.class, ccf.getAffinity().getClass());
}
-
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheProcessorActiveTxTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheProcessorActiveTxTest.java
index 7166f1f..3099f74 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheProcessorActiveTxTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheProcessorActiveTxTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.cache;
+import java.util.List;
import org.apache.ignite.IgniteException;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.IgniteEx;
@@ -25,8 +26,6 @@ import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.ignite.transactions.Transaction;
import org.junit.Test;
-import java.util.List;
-
import static java.lang.String.format;
import static java.util.Objects.nonNull;
import static java.util.stream.Collectors.toList;
@@ -40,26 +39,12 @@ import static org.apache.ignite.testframework.GridTestUtils.getFieldValue;
*/
public class GridCacheProcessorActiveTxTest extends GridCommonAbstractTest {
/**
- * The prefix for the exception message.
- */
- private static final String CHECK_EMPTY_TRANSACTIONS_ERROR_MSG =
- "Cannot start/stop cache within lock or transaction.";
-
- /**
- * Format for displaying the cache name and operation.
- *
- * @see GridCacheProcessor#CACHE_NAME_AND_OPERATION_FORMAT
- */
- private static final String CACHE_NAME_AND_OPERATION_FORMAT =
- getFieldValue(GridCacheProcessor.class, "CACHE_NAME_AND_OPERATION_FORMAT");
-
- /**
- * Format for displaying the cache names and operation.
+ * Format for displaying the cache name(s) and operation.
*
- * @see GridCacheProcessor#CACHE_NAMES_AND_OPERATION_FORMAT
+ * @see GridCacheProcessor#CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT
*/
- private static final String CACHE_NAMES_AND_OPERATION_FORMAT =
- getFieldValue(GridCacheProcessor.class, "CACHE_NAMES_AND_OPERATION_FORMAT");
+ private static final String CHECK_EMPTY_TRANSACTIONS_ERROR_MSG =
+ getFieldValue(GridCacheProcessor.class, "CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT");
/** Node. */
private static IgniteEx NODE;
@@ -94,7 +79,6 @@ public class GridCacheProcessorActiveTxTest extends GridCommonAbstractTest {
opInActiveTx(
() -> NODE.createCache(new CacheConfiguration<>(cacheName)),
- CACHE_NAME_AND_OPERATION_FORMAT,
cacheName,
"dynamicStartCache"
);
@@ -112,7 +96,6 @@ public class GridCacheProcessorActiveTxTest extends GridCommonAbstractTest {
opInActiveTx(
() -> NODE.createCaches(cacheCfgs),
- CACHE_NAMES_AND_OPERATION_FORMAT,
cacheNames.toString(),
"dynamicStartCachesByStoredConf"
);
@@ -129,7 +112,6 @@ public class GridCacheProcessorActiveTxTest extends GridCommonAbstractTest {
opInActiveTx(
() -> NODE.destroyCache(cacheName),
- CACHE_NAME_AND_OPERATION_FORMAT,
cacheName,
"dynamicDestroyCache"
);
@@ -149,7 +131,6 @@ public class GridCacheProcessorActiveTxTest extends GridCommonAbstractTest {
opInActiveTx(
() -> NODE.destroyCaches(cacheNames),
- CACHE_NAMES_AND_OPERATION_FORMAT,
cacheNames.toString(),
"dynamicDestroyCaches"
);
@@ -168,7 +149,6 @@ public class GridCacheProcessorActiveTxTest extends GridCommonAbstractTest {
opInActiveTx(
() -> cacheProcessor.dynamicCloseCache(cacheName),
- CACHE_NAME_AND_OPERATION_FORMAT,
cacheName,
"dynamicCloseCache"
);
@@ -185,7 +165,6 @@ public class GridCacheProcessorActiveTxTest extends GridCommonAbstractTest {
opInActiveTx(
() -> NODE.context().cache().resetCacheState(cacheNames),
- CACHE_NAME_AND_OPERATION_FORMAT,
cacheNames.toString(),
"resetCacheState"
);
@@ -215,19 +194,14 @@ public class GridCacheProcessorActiveTxTest extends GridCommonAbstractTest {
/**
* Performing an operation in an active transaction with a check that an
* exception will be thrown with a format
- * {@link #CHECK_EMPTY_TRANSACTIONS_ERROR_MSG} +
- * {@link #CACHE_NAMES_AND_OPERATION_FORMAT} of
- * {@link #CACHE_NAME_AND_OPERATION_FORMAT} message.
+ * {@link #CHECK_EMPTY_TRANSACTIONS_ERROR_MSG}.
*
* @param runnableX Operation in an active transaction.
- * @param format Format for cache(s) name and operation
- * in the exception message.
* @param cacheName Cache name for the exception message.
* @param operation Operation for the exception message.
*/
- private void opInActiveTx(RunnableX runnableX, String format, String cacheName, String operation) {
+ private void opInActiveTx(RunnableX runnableX, String cacheName, String operation) {
assert nonNull(runnableX);
- assert nonNull(format);
assert nonNull(cacheName);
assert nonNull(operation);
@@ -239,7 +213,7 @@ public class GridCacheProcessorActiveTxTest extends GridCommonAbstractTest {
return null;
},
IgniteException.class,
- format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG + ' ' + format, cacheName, operation)
+ format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG, cacheName, operation)
);
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteStartCacheInTransactionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteStartCacheInTransactionSelfTest.java
index 0514548..b5fc48a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteStartCacheInTransactionSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteStartCacheInTransactionSelfTest.java
@@ -41,7 +41,7 @@ import static org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_REA
*/
public class IgniteStartCacheInTransactionSelfTest extends GridCommonAbstractTest {
/** */
- private static final String EXPECTED_MSG = "Cannot start/stop cache within lock or transaction.";
+ private static final String EXPECTED_MSG = "Cannot start/stop cache within lock or transaction";
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {