You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by rp...@apache.org on 2025/04/21 12:46:44 UTC
(ignite-3) branch main updated: IGNITE-25149 Do not trigger FailureProcessor on node stop (#5653)
This is an automated email from the ASF dual-hosted git repository.
rpuch pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 11fd3ce226a IGNITE-25149 Do not trigger FailureProcessor on node stop (#5653)
11fd3ce226a is described below
commit 11fd3ce226a97cd2f04a38e8737a0f3f99210f6e
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Mon Apr 21 16:46:39 2025 +0400
IGNITE-25149 Do not trigger FailureProcessor on node stop (#5653)
1. Wrap CancellationException and RejectedExecutionException thrown due to RaftGroupService being stopped in either NodeStoppingException (for CMG and MG) or ComponentStoppingException (for partition groups)
2. Avoid using CancellationException and RejectedExecutionException to detect whether a failure should be ignored, instead look at NodeStoppingException and ComponentStoppingException
3. Do not trigger failure processor/log when a transient replica start problem occurs
4. Use message passed to failure processor to construct a stack trace capturing exception
5. Mute failure processor logging in some tests
---
.../internal/catalog/CatalogManagerImpl.java | 3 +-
.../management/ClusterManagementGroupManager.java | 3 +-
.../ignite/internal/failure/FailureContext.java | 8 ++-
...eption.java => ComponentStoppingException.java} | 36 ++++++------
.../internal/lang/NodeStoppingException.java | 2 +-
.../testframework/log4j2/LogInspector.java | 4 ++
.../distributionzones/DataNodesManager.java | 3 +-
.../distributionzones/DistributionZoneManager.java | 3 +-
.../RebalanceRaftGroupEventsListener.java | 3 +-
.../ZoneRebalanceRaftGroupEventsListener.java | 3 +-
.../ignite/internal/failure/FailureManager.java | 12 +---
.../failure/StackTraceCapturingException.java | 31 ++++++++++
.../internal/index/IndexBuildController.java | 3 +-
.../impl/ItMetaStorageManagerImplTest.java | 3 +-
.../impl/MetaStorageCompactionTrigger.java | 3 +-
.../impl/MetaStorageLeaderElectionListener.java | 3 +-
.../impl/MetaStorageLearnerManager.java | 3 +-
.../metastorage/impl/MetaStorageManagerImpl.java | 3 +-
.../metastorage/server/WatchProcessor.java | 27 +++++----
.../metastorage/server/time/ClusterTimeImpl.java | 8 +--
.../PartitionReplicaLifecycleManager.java | 4 +-
.../internal/placementdriver/LeaseUpdater.java | 3 +-
.../placementdriver/PlacementDriverManager.java | 4 +-
.../internal/placementdriver/ActiveActorTest.java | 6 +-
.../ignite/internal/raft/ExceptionFactory.java | 40 +++++++++++++
.../apache/ignite/internal/raft/RaftManager.java | 6 +-
.../ignite/internal/raft/RaftServiceFactory.java | 4 +-
.../raft/ComponentStoppingExceptionFactory.java | 35 ++++++++++++
.../java/org/apache/ignite/internal/raft/Loza.java | 56 ++++++++++++++----
.../raft/NodeStoppingExceptionFactory.java | 36 ++++++++++++
.../ignite/internal/raft/RaftGroupServiceImpl.java | 66 ++++++++++++++++++++--
.../internal/raft/StoppingExceptionFactories.java | 45 +++++++++++++++
.../raft/client/TopologyAwareRaftGroupService.java | 16 +++++-
.../TopologyAwareRaftGroupServiceFactory.java | 7 ++-
.../PlacementDriverMessageProcessor.java | 8 +--
.../ignite/internal/replicator/ReplicaManager.java | 6 +-
.../internal/replicator/ReplicaStateManager.java | 3 +-
.../replicator/TransientReplicaStartException.java | 34 +++++++++++
.../AbstractTopologyAwareGroupServiceTest.java | 4 +-
.../metastorage/server/WatchListenerInhibitor.java | 2 +-
.../PartitionReplicatorNodeRecovery.java | 6 +-
.../internal/table/distributed/TableManager.java | 4 +-
.../distributed/TableManagerRecoveryTest.java | 2 +-
.../table/distributed/TableManagerTest.java | 4 +-
.../ItDisasterRecoveryReconfigurationTest.java | 9 +++
.../internal/worker/CriticalWorkerWatchdog.java | 7 +--
46 files changed, 455 insertions(+), 126 deletions(-)
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
index a57a8a8fca0..56239010c22 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java
@@ -37,7 +37,6 @@ import java.util.BitSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.NavigableMap;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.function.LongSupplier;
@@ -506,7 +505,7 @@ public class CatalogManagerImpl extends AbstractEventProducer<CatalogEvent, Cata
// we guarantee recovery for a zones' catalog actions only if that actions were completed.
return allOf(eventFutures.toArray(CompletableFuture[]::new))
.whenComplete((ignore, err) -> {
- if (err != null && !hasCause(err, NodeStoppingException.class, CancellationException.class)) {
+ if (err != null && !hasCause(err, NodeStoppingException.class)) {
failureProcessor.process(new FailureContext(err, "Failed to apply catalog update."));
}
diff --git a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java
index 100133d62b4..c028ec2de58 100644
--- a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java
+++ b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java
@@ -34,7 +34,6 @@ import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executors;
@@ -591,7 +590,7 @@ public class ClusterManagementGroupManager extends AbstractEventProducer<Cluster
}))
.whenComplete((v, e) -> {
if (e != null) {
- if (hasCause(e, NodeStoppingException.class, CancellationException.class)) {
+ if (hasCause(e, NodeStoppingException.class)) {
LOG.info("Unable to execute onLeaderElected callback, because the node is stopping", e);
} else {
failureProcessor.process(new FailureContext(
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/failure/FailureContext.java b/modules/core/src/main/java/org/apache/ignite/internal/failure/FailureContext.java
index bf9891a1d7d..4b6be7a3f9b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/failure/FailureContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/failure/FailureContext.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.failure;
import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.Nullable;
/**
* Failure context contains information about failure type and exception if applicable.
@@ -28,7 +29,7 @@ public class FailureContext {
private final FailureType type;
/** Error. */
- private final Throwable err;
+ private final @Nullable Throwable err;
/** Message describing the failure. */
private final String message;
@@ -66,10 +67,10 @@ public class FailureContext {
* Creates instance of {@link FailureContext}.
*
* @param type Failure type.
- * @param err Exception.
+ * @param err Exception. Might be {@code null} if no exception is available.
* @param message Message describing the failure.
*/
- public FailureContext(FailureType type, Throwable err, String message) {
+ public FailureContext(FailureType type, @Nullable Throwable err, String message) {
this.type = type;
this.err = err;
this.message = message;
@@ -89,6 +90,7 @@ public class FailureContext {
*
* @return Exception or {@code null}.
*/
+ @Nullable
public Throwable error() {
return err;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/lang/NodeStoppingException.java b/modules/core/src/main/java/org/apache/ignite/internal/lang/ComponentStoppingException.java
similarity index 52%
copy from modules/core/src/main/java/org/apache/ignite/internal/lang/NodeStoppingException.java
copy to modules/core/src/main/java/org/apache/ignite/internal/lang/ComponentStoppingException.java
index f3334648cee..02e7759ae49 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/lang/NodeStoppingException.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/lang/ComponentStoppingException.java
@@ -17,23 +17,25 @@
package org.apache.ignite.internal.lang;
-import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import java.util.UUID;
import org.jetbrains.annotations.Nullable;
/**
- * This exception is used in order to indicate that Ignite node is stopping (already stopped) for some reason.
+ * This exception is used to indicate that a component (like replication group client) is stopping (already stopped) for some reason.
+ * This is different from {@link NodeStoppingException} as {@link ComponentStoppingException} might mean that just the component is stopped,
+ * not the whole node.
*/
-public class NodeStoppingException extends IgniteInternalCheckedException {
+public class ComponentStoppingException extends IgniteInternalCheckedException {
/** Serial version UID. */
private static final long serialVersionUID = 0L;
/**
- * Creates an empty node stopping exception.
+ * Creates an empty component stopping exception.
*/
- public NodeStoppingException() {
- super(NODE_STOPPING_ERR, "Operation has been cancelled (node is stopping).");
+ public ComponentStoppingException() {
+ super(INTERNAL_ERR, "Operation has been cancelled (component is stopping).");
}
/**
@@ -41,37 +43,37 @@ public class NodeStoppingException extends IgniteInternalCheckedException {
*
* @param msg Error message.
*/
- public NodeStoppingException(String msg) {
- super(NODE_STOPPING_ERR, msg);
+ public ComponentStoppingException(String msg) {
+ super(INTERNAL_ERR, msg);
}
/**
- * Creates a new node stopping exception with the given throwable as a cause and source of error message.
+ * Creates a new component stopping exception with the given throwable as a cause and source of error message.
*
* @param cause Non-null throwable cause.
*/
- public NodeStoppingException(Throwable cause) {
- super(NODE_STOPPING_ERR, cause);
+ public ComponentStoppingException(Throwable cause) {
+ super(INTERNAL_ERR, cause);
}
/**
- * Creates a new node stopping exception with the given error message and optional nested exception.
+ * Creates a new component stopping exception with the given error message and optional nested exception.
*
* @param msg Error message.
* @param cause Optional nested exception (can be {@code null}).
* @param writableStackTrace Whether or not the stack trace should be writable.
*/
- public NodeStoppingException(String msg, @Nullable Throwable cause, boolean writableStackTrace) {
- super(UUID.randomUUID(), NODE_STOPPING_ERR, msg, cause, writableStackTrace);
+ public ComponentStoppingException(String msg, @Nullable Throwable cause, boolean writableStackTrace) {
+ super(UUID.randomUUID(), INTERNAL_ERR, msg, cause, writableStackTrace);
}
/**
- * Creates a new node stopping exception with the given error message and optional nested exception.
+ * Creates a new component stopping exception with the given error message and optional nested exception.
*
* @param msg Error message.
* @param cause Optional nested exception (can be {@code null}).
*/
- public NodeStoppingException(String msg, @Nullable Throwable cause) {
- super(NODE_STOPPING_ERR, msg, cause);
+ public ComponentStoppingException(String msg, @Nullable Throwable cause) {
+ super(INTERNAL_ERR, msg, cause);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/lang/NodeStoppingException.java b/modules/core/src/main/java/org/apache/ignite/internal/lang/NodeStoppingException.java
index f3334648cee..8983ab6e92a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/lang/NodeStoppingException.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/lang/NodeStoppingException.java
@@ -23,7 +23,7 @@ import java.util.UUID;
import org.jetbrains.annotations.Nullable;
/**
- * This exception is used in order to indicate that Ignite node is stopping (already stopped) for some reason.
+ * This exception is used to indicate that Ignite node is stopping (already stopped) for some reason.
*/
public class NodeStoppingException extends IgniteInternalCheckedException {
/** Serial version UID. */
diff --git a/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/log4j2/LogInspector.java b/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/log4j2/LogInspector.java
index fae0b21f819..6d208fafefc 100755
--- a/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/log4j2/LogInspector.java
+++ b/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/log4j2/LogInspector.java
@@ -97,6 +97,10 @@ public class LogInspector {
/** Logger configuration. */
private Configuration config;
+ static {
+ Log4jUtils.waitTillConfigured();
+ }
+
/**
* Creates a new instance of {@link LogInspector} for the given {@code loggerName}.
*
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DataNodesManager.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DataNodesManager.java
index e59da50e8a8..827dc321cf7 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DataNodesManager.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DataNodesManager.java
@@ -72,7 +72,6 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
@@ -1100,7 +1099,7 @@ public class DataNodesManager {
}
private static boolean relatesToNodeStopping(Throwable e) {
- return hasCause(e, NodeStoppingException.class, CancellationException.class);
+ return hasCause(e, NodeStoppingException.class);
}
/**
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
index 5a66184b629..f97eca5e0ef 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
@@ -63,7 +63,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListMap;
@@ -464,7 +463,7 @@ public class DistributionZoneManager extends
}
private static boolean relatesToNodeStopping(Throwable e) {
- return hasCause(e, NodeStoppingException.class, CancellationException.class);
+ return hasCause(e, NodeStoppingException.class);
}
/**
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java
index 913daea1c0f..ad31e580b8c 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/RebalanceRaftGroupEventsListener.java
@@ -37,7 +37,6 @@ import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
@@ -504,7 +503,7 @@ public class RebalanceRaftGroupEventsListener implements RaftGroupEventsListener
} catch (InterruptedException | ExecutionException e) {
// TODO: IGNITE-14693
- if (!hasCause(e, NodeStoppingException.class, CancellationException.class)) {
+ if (!hasCause(e, NodeStoppingException.class)) {
String errorMessage = String.format("Unable to commit partition configuration to metastore: %s", tablePartitionId);
failureProcessor.process(new FailureContext(e, errorMessage));
}
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceRaftGroupEventsListener.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceRaftGroupEventsListener.java
index fb67764b2c7..69d9ea59b0f 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceRaftGroupEventsListener.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/rebalance/ZoneRebalanceRaftGroupEventsListener.java
@@ -42,7 +42,6 @@ import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
@@ -541,7 +540,7 @@ public class ZoneRebalanceRaftGroupEventsListener implements RaftGroupEventsList
} catch (InterruptedException | ExecutionException e) {
// TODO: IGNITE-14693
- if (!hasCause(e, NodeStoppingException.class, CancellationException.class)) {
+ if (!hasCause(e, NodeStoppingException.class)) {
String errorMessage = String.format("Unable to commit partition configuration to metastore: %s", zonePartitionId);
failureProcessor.process(new FailureContext(e, errorMessage));
}
diff --git a/modules/failure-handler/src/main/java/org/apache/ignite/internal/failure/FailureManager.java b/modules/failure-handler/src/main/java/org/apache/ignite/internal/failure/FailureManager.java
index 2e469683a29..c6268220cd2 100644
--- a/modules/failure-handler/src/main/java/org/apache/ignite/internal/failure/FailureManager.java
+++ b/modules/failure-handler/src/main/java/org/apache/ignite/internal/failure/FailureManager.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.failure;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.ExceptionUtils.hasCauseOrSuppressed;
import static org.apache.ignite.lang.ErrorGroups.Common.COMPONENT_NOT_STARTED_ERR;
-import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import java.util.EnumMap;
import java.util.EnumSet;
@@ -38,7 +37,6 @@ import org.apache.ignite.internal.failure.handlers.configuration.NoOpFailureHand
import org.apache.ignite.internal.failure.handlers.configuration.StopNodeFailureHandlerConfigurationSchema;
import org.apache.ignite.internal.failure.handlers.configuration.StopNodeOrHaltFailureHandlerConfigurationSchema;
import org.apache.ignite.internal.failure.handlers.configuration.StopNodeOrHaltFailureHandlerView;
-import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.manager.ComponentContext;
@@ -160,14 +158,14 @@ public class FailureManager implements FailureProcessor, IgniteComponent {
return false;
}
- var exceptionForLogging = new StackTraceCapturingException(failureCtx.error());
+ var exceptionForLogging = new StackTraceCapturingException(failureCtx.message(), failureCtx.error());
if (handler.ignoredFailureTypes().contains(failureCtx.type())) {
LOG.warn(IGNORED_FAILURE_LOG_MSG, exceptionForLogging, handler, failureCtx.type());
} else {
LOG.error(FAILURE_LOG_MSG, exceptionForLogging, handler, failureCtx.type());
}
- if (reserveBuf != null && hasCauseOrSuppressed(failureCtx.error(), OutOfMemoryError.class)) {
+ if (reserveBuf != null && failureCtx.error() != null && hasCauseOrSuppressed(failureCtx.error(), OutOfMemoryError.class)) {
reserveBuf = null;
}
@@ -302,10 +300,4 @@ public class FailureManager implements FailureProcessor, IgniteComponent {
return throttle;
}
-
- private static class StackTraceCapturingException extends IgniteInternalException {
- private StackTraceCapturingException(Throwable cause) {
- super(INTERNAL_ERR, cause);
- }
- }
}
diff --git a/modules/failure-handler/src/main/java/org/apache/ignite/internal/failure/StackTraceCapturingException.java b/modules/failure-handler/src/main/java/org/apache/ignite/internal/failure/StackTraceCapturingException.java
new file mode 100644
index 00000000000..b27b4e2dfe3
--- /dev/null
+++ b/modules/failure-handler/src/main/java/org/apache/ignite/internal/failure/StackTraceCapturingException.java
@@ -0,0 +1,31 @@
+/*
+ * 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.failure;
+
+import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
+
+import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.jetbrains.annotations.Nullable;
+
+class StackTraceCapturingException extends IgniteInternalException {
+ private static final long serialVersionUID = 0L;
+
+ StackTraceCapturingException(String message, @Nullable Throwable cause) {
+ super(INTERNAL_ERR, message, cause);
+ }
+}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java
index 98ebea8693f..499084adba9 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java
@@ -35,7 +35,6 @@ import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
import java.util.ArrayList;
import java.util.Set;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -336,7 +335,7 @@ class IndexBuildController implements ManuallyCloseable {
}
}).whenComplete((res, ex) -> {
if (ex != null) {
- if (!hasCause(ex, NodeStoppingException.class, CancellationException.class)) {
+ if (!hasCause(ex, NodeStoppingException.class)) {
failureProcessor.process(new FailureContext(FailureType.CRITICAL_ERROR, ex));
}
}
diff --git a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageManagerImplTest.java b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageManagerImplTest.java
index 6332fde7cf0..9e8b1e2b68b 100644
--- a/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageManagerImplTest.java
+++ b/modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/impl/ItMetaStorageManagerImplTest.java
@@ -48,7 +48,6 @@ import static org.mockito.Mockito.when;
import java.util.List;
import java.util.Set;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ScheduledExecutorService;
@@ -271,7 +270,7 @@ public class ItMetaStorageManagerImplTest extends IgniteAbstractTest {
CompletableFuture<Entry> fut = svc.get(FOO_KEY);
- assertThat(fut, willThrowFast(CancellationException.class));
+ assertThat(fut, willThrowFast(NodeStoppingException.class));
}
@Test
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageCompactionTrigger.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageCompactionTrigger.java
index 964033b7677..e3fa73cfaaa 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageCompactionTrigger.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageCompactionTrigger.java
@@ -27,7 +27,6 @@ import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockSafe;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -207,7 +206,7 @@ public class MetaStorageCompactionTrigger implements IgniteComponent {
metaStorageManager.sendCompactionCommand(newCompactionRevision)
.whenComplete((unused, throwable) -> {
if (throwable != null) {
- if (!hasCause(throwable, NodeStoppingException.class, CancellationException.class)) {
+ if (!hasCause(throwable, NodeStoppingException.class)) {
String errorMessage = String.format(
"Unknown error occurred while sending the metastorage compaction command: "
+ "[newCompactionRevision=%s]",
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageLeaderElectionListener.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageLeaderElectionListener.java
index 4234cd98ceb..2f80f6f4ad3 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageLeaderElectionListener.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageLeaderElectionListener.java
@@ -21,7 +21,6 @@ import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFu
import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
import java.util.List;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.function.BooleanSupplier;
import java.util.function.Function;
@@ -181,7 +180,7 @@ public class MetaStorageLeaderElectionListener implements LeaderElectionListener
})
.whenComplete((v, e) -> {
if (e != null) {
- if (!hasCause(e, NodeStoppingException.class, CancellationException.class)) {
+ if (!hasCause(e, NodeStoppingException.class)) {
failureProcessor.process(new FailureContext(e, "Unable to start Idle Safe Time scheduler"));
}
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageLearnerManager.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageLearnerManager.java
index cd36b0187ba..d1e5ed2bc20 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageLearnerManager.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageLearnerManager.java
@@ -23,7 +23,6 @@ import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
import java.util.List;
import java.util.Set;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
@@ -130,7 +129,7 @@ class MetaStorageLearnerManager {
return action.get()
.whenComplete((v, e) -> {
if (e != null) {
- if (!hasCause(e, NodeStoppingException.class, CancellationException.class)) {
+ if (!hasCause(e, NodeStoppingException.class)) {
failureProcessor.process(new FailureContext(e, "Unable to change peers on topology update"));
}
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java
index 54ca2a9e9b0..e5930391d20 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java
@@ -34,7 +34,6 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -666,7 +665,7 @@ public class MetaStorageManagerImpl implements MetaStorageManager, MetastorageGr
if (unwrapped instanceof RaftException && ((RaftException) unwrapped).raftError() == RaftError.ECATCHUP) {
// Some node has left, it's not a reason to fail our node; just log the error.
LOG.error("Error while changing voting set to {}", ex, currentState.targetPeers);
- } else if (!hasCause(ex, NodeStoppingException.class, CancellationException.class)) {
+ } else if (!hasCause(ex, NodeStoppingException.class)) {
String errorMessage = IgniteStringFormatter.format(
"Error while changing voting set to {}",
currentState.targetPeers
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchProcessor.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchProcessor.java
index f386253fb2c..d48ab17b474 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchProcessor.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/WatchProcessor.java
@@ -17,11 +17,13 @@
package org.apache.ignite.internal.metastorage.server;
+import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.emptyList;
import static java.util.concurrent.CompletableFuture.allOf;
import static java.util.concurrent.CompletableFuture.failedFuture;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.stream.Collectors.joining;
import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.failure.FailureType.CRITICAL_ERROR;
import static org.apache.ignite.internal.metastorage.server.raft.MetaStorageWriteHandler.IDEMPOTENT_COMMAND_PREFIX_BYTES;
@@ -30,7 +32,6 @@ import static org.apache.ignite.internal.util.CompletableFutures.copyStateTo;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
-import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -46,7 +47,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Supplier;
-import java.util.stream.Collectors;
import org.apache.ignite.internal.close.ManuallyCloseable;
import org.apache.ignite.internal.failure.FailureContext;
import org.apache.ignite.internal.failure.FailureProcessor;
@@ -218,10 +218,11 @@ public class WatchProcessor implements ManuallyCloseable {
* Composes passed action with {@link #notificationFuture} and handles any exceptions that might have occurred.
*
* @param asyncAction Action to compose.
+ * @param additionalInfoSupplier Supplier of additional information that will be used for logging and/or invoking the FailureProcessor.
* @return Updated value of {@link #notificationFuture}.
*/
@VisibleForTesting
- CompletableFuture<Void> enqueue(Supplier<CompletableFuture<Void>> asyncAction) {
+ CompletableFuture<Void> enqueue(Supplier<CompletableFuture<Void>> asyncAction, Supplier<String> additionalInfoSupplier) {
while (true) {
CompletableFuture<Void> chainingFuture = new CompletableFuture<>();
@@ -229,7 +230,7 @@ public class WatchProcessor implements ManuallyCloseable {
.thenComposeAsync(v -> inBusyLockAsync(asyncAction), watchExecutor)
.whenComplete((unused, e) -> {
if (e != null) {
- notifyFailureHandlerOnFirstFailureInNotificationChain(e);
+ notifyFailureHandlerOnFirstFailureInNotificationChain(e, additionalInfoSupplier);
}
});
@@ -266,7 +267,13 @@ public class WatchProcessor implements ManuallyCloseable {
newNotificationFuture.whenComplete((unused, e) -> maybeLogLongProcessing(filteredUpdatedEntries, startTimeNanos));
return newNotificationFuture;
- });
+ }, updatedEntriesKeysInfo(updatedEntries));
+ }
+
+ private static Supplier<String> updatedEntriesKeysInfo(List<Entry> updatedEntries) {
+ return () -> updatedEntries.stream()
+ .map(entry -> new String(entry.key(), UTF_8))
+ .collect(joining(",", "Keys of updated entries: ", ""));
}
private static CompletableFuture<Void> performWatchesNotifications(
@@ -309,8 +316,8 @@ public class WatchProcessor implements ManuallyCloseable {
if (durationMillis > WATCH_EVENT_PROCESSING_LOG_THRESHOLD_MILLIS) {
String keysHead = updatedEntries.stream()
.limit(WATCH_EVENT_PROCESSING_LOG_KEYS)
- .map(entry -> new String(entry.key(), StandardCharsets.UTF_8))
- .collect(Collectors.joining(", "));
+ .map(entry -> new String(entry.key(), UTF_8))
+ .collect(joining(", "));
String keysTail = updatedEntries.size() > WATCH_EVENT_PROCESSING_LOG_KEYS ? ", ..." : "";
@@ -384,16 +391,16 @@ public class WatchProcessor implements ManuallyCloseable {
watchEventHandlingCallback.onSafeTimeAdvanced(time);
return nullCompletedFuture();
- });
+ }, () -> "<nothing>");
}
- private void notifyFailureHandlerOnFirstFailureInNotificationChain(Throwable e) {
+ private void notifyFailureHandlerOnFirstFailureInNotificationChain(Throwable e, Supplier<String> additionalInfoSupplier) {
if (firedFailureOnChain.compareAndSet(false, true)) {
boolean nodeStopping = hasCause(e, NodeStoppingException.class);
if (!nodeStopping) {
LOG.error("Notification chain encountered an error, so no notifications will be ever fired for subsequent revisions "
- + "until a restart. Notifying the FailureManager");
+ + "until a restart. Notifying the FailureManager. Additional info: '{}'", additionalInfoSupplier.get(), e);
failureProcessor.process(new FailureContext(CRITICAL_ERROR, e));
} else {
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/time/ClusterTimeImpl.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/time/ClusterTimeImpl.java
index 34c8399cbf6..210fb4bdf58 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/time/ClusterTimeImpl.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/time/ClusterTimeImpl.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.metastorage.server.time;
import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
@@ -258,12 +257,7 @@ public class ClusterTimeImpl implements ClusterTime, MetaStorageMetrics, Manuall
syncTimeAction.syncTime(clock.now())
.whenComplete((v, e) -> {
if (e != null) {
- if (!hasCause(
- e,
- NodeStoppingException.class,
- CancellationException.class,
- RejectedExecutionException.class
- )) {
+ if (!hasCause(e, NodeStoppingException.class)) {
failureProcessor.process(new FailureContext(e, "Unable to perform idle time sync"));
}
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java
index cc95f89dc6c..da271addb01 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java
@@ -71,7 +71,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -101,6 +100,7 @@ import org.apache.ignite.internal.failure.FailureProcessor;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.ByteArray;
+import org.apache.ignite.internal.lang.ComponentStoppingException;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.lang.IgniteStringFormatter;
import org.apache.ignite.internal.lang.NodeStoppingException;
@@ -1321,7 +1321,7 @@ public class PartitionReplicaLifecycleManager extends
return LeaderWithTerm.NO_LEADER;
}
- if (hasCause(throwable, RejectedExecutionException.class)) {
+ if (hasCause(throwable, ComponentStoppingException.class)) {
LOG.info("Replica is being stopped so the changing peers is skipped [grp={}].", replicaGrpId);
return LeaderWithTerm.NO_LEADER;
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
index 394c65c6145..dd14fb37a3e 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
@@ -43,7 +43,6 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
@@ -588,7 +587,7 @@ public class LeaseUpdater {
noop()
).whenComplete((success, e) -> {
if (e != null) {
- if (!hasCause(e, NodeStoppingException.class, CancellationException.class)) {
+ if (!hasCause(e, NodeStoppingException.class)) {
failureProcessor.process(new FailureContext(e, "Lease update invocation failed"));
}
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java
index 2435cee1e0d..12556b1de53 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java
@@ -49,6 +49,7 @@ import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParam
import org.apache.ignite.internal.placementdriver.leases.LeaseTracker;
import org.apache.ignite.internal.raft.PeersAndLearners;
import org.apache.ignite.internal.raft.RaftManager;
+import org.apache.ignite.internal.raft.StoppingExceptionFactories;
import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupService;
import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupServiceFactory;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
@@ -180,7 +181,8 @@ public class PlacementDriverManager implements IgniteComponent {
replicationGroupId,
PeersAndLearners.fromConsistentIds(placementDriverNodes),
topologyAwareRaftGroupServiceFactory,
- null // Use default commands marshaller.
+ null, // Use default commands marshaller.
+ StoppingExceptionFactories.indicateNodeStop()
);
return raftClient.subscribeLeader(this::onLeaderChange).thenApply(v -> raftClient);
diff --git a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/ActiveActorTest.java b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/ActiveActorTest.java
index e5634d6779a..d508a52b8c5 100644
--- a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/ActiveActorTest.java
+++ b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/ActiveActorTest.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.raft.Peer;
import org.apache.ignite.internal.raft.PeersAndLearners;
+import org.apache.ignite.internal.raft.StoppingExceptionFactories;
import org.apache.ignite.internal.raft.client.AbstractTopologyAwareGroupServiceTest;
import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupServiceFactory;
import org.apache.ignite.internal.replicator.configuration.ReplicationConfiguration;
@@ -108,13 +109,14 @@ public class ActiveActorTest extends AbstractTopologyAwareGroupServiceTest {
var mockRaftMgr = mock(Loza.class);
try {
- when(mockRaftMgr.startRaftGroupService(any(), any(), any(), any())).then(invocation ->
+ when(mockRaftMgr.startRaftGroupService(any(), any(), any(), any(), any())).then(invocation ->
raftGroupServiceFactory.startRaftGroupService(
GROUP_ID,
peersAndLearners,
raftConfiguration,
executor,
- null
+ null,
+ StoppingExceptionFactories.indicateComponentStop()
)
);
} catch (NodeStoppingException e) {
diff --git a/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/ExceptionFactory.java b/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/ExceptionFactory.java
new file mode 100644
index 00000000000..988293551fc
--- /dev/null
+++ b/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/ExceptionFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.raft;
+
+/**
+ * Factory for instances of exceptions.
+ */
+public interface ExceptionFactory {
+ /**
+ * Creates an exception with the given message.
+ *
+ * @param message Exception message.
+ * @return New exception instance.
+ */
+ Exception create(String message);
+
+ /**
+ * Wraps an exception.
+ *
+ * @param message New exception message.
+ * @param cause Exception to wrap
+ * @return New exception instance.
+ */
+ Exception wrap(String message, Throwable cause);
+}
diff --git a/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java b/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java
index 92b9064aeb8..9f7b2243f92 100644
--- a/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java
+++ b/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java
@@ -63,6 +63,8 @@ public interface RaftManager extends IgniteComponent {
*
* <p>Synchronously waits for the Raft log to be applied.
*
+ * <p>The started RaftGroupService will indicate that it is being stopped (when it's stopped) with {@link NodeStoppingException}s.
+ *
* @param nodeId Raft node ID.
* @param configuration Peers and Learners of the Raft group.
* @param lsnr Raft group listener.
@@ -121,6 +123,7 @@ public interface RaftManager extends IgniteComponent {
* @param factory Factory that should be used to create raft service.
* @param commandsMarshaller Marshaller that should be used to serialize commands. {@code null} if default marshaller should be
* used.
+ * @param stoppingExceptionFactory Exception factory used to create exceptions thrown to indicate that the object is being stopped.
* @return Raft group service.
* @throws NodeStoppingException If node stopping intention was detected.
*/
@@ -128,7 +131,8 @@ public interface RaftManager extends IgniteComponent {
ReplicationGroupId groupId,
PeersAndLearners configuration,
RaftServiceFactory<T> factory,
- @Nullable Marshaller commandsMarshaller
+ @Nullable Marshaller commandsMarshaller,
+ ExceptionFactory stoppingExceptionFactory
) throws NodeStoppingException;
/**
diff --git a/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftServiceFactory.java b/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftServiceFactory.java
index abaa546e8fc..c11ed3520c1 100644
--- a/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftServiceFactory.java
+++ b/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftServiceFactory.java
@@ -34,6 +34,7 @@ public interface RaftServiceFactory<T extends RaftGroupService> {
* @param raftConfiguration Raft configuration.
* @param raftClientExecutor Client executor.
* @param commandsMarshaller Marshaller that should be used to serialize commands.
+ * @param stoppingExceptionFactory Exception factory used to create exceptions thrown to indicate that the object is being stopped.
* @return New Raft client.
*/
T startRaftGroupService(
@@ -41,6 +42,7 @@ public interface RaftServiceFactory<T extends RaftGroupService> {
PeersAndLearners peersAndLearners,
RaftConfiguration raftConfiguration,
ScheduledExecutorService raftClientExecutor,
- Marshaller commandsMarshaller
+ Marshaller commandsMarshaller,
+ ExceptionFactory stoppingExceptionFactory
);
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/ComponentStoppingExceptionFactory.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/ComponentStoppingExceptionFactory.java
new file mode 100644
index 00000000000..3c3a77e0b08
--- /dev/null
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/ComponentStoppingExceptionFactory.java
@@ -0,0 +1,35 @@
+/*
+ * 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.raft;
+
+import org.apache.ignite.internal.lang.ComponentStoppingException;
+
+/**
+ * {@link ExceptionFactory} that produces instances of {@link ComponentStoppingException}.
+ */
+public class ComponentStoppingExceptionFactory implements ExceptionFactory {
+ @Override
+ public Exception create(String message) {
+ return new ComponentStoppingException(message);
+ }
+
+ @Override
+ public Exception wrap(String message, Throwable cause) {
+ return new ComponentStoppingException(message, cause);
+ }
+}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
index a2636de58d9..fe88191e541 100644
--- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
@@ -293,7 +293,15 @@ public class Loza implements RaftManager {
}
try {
- return startRaftGroupNodeInternal(nodeId, configuration, lsnr, eventsLsnr, groupOptions, raftServiceFactory);
+ return startRaftGroupNodeInternal(
+ nodeId,
+ configuration,
+ lsnr,
+ eventsLsnr,
+ groupOptions,
+ raftServiceFactory,
+ StoppingExceptionFactories.indicateComponentStop()
+ );
} finally {
busyLock.leaveBusy();
}
@@ -325,7 +333,8 @@ public class Loza implements RaftManager {
eventsLsnr,
// Use default marshaller here, because this particular method is used in very specific circumstances.
raftGroupOptions,
- factory
+ factory,
+ StoppingExceptionFactories.indicateNodeStop()
);
} finally {
busyLock.leaveBusy();
@@ -341,7 +350,12 @@ public class Loza implements RaftManager {
try {
// Use default command marshaller here.
- return startRaftGroupServiceInternal(groupId, configuration, opts.getCommandsMarshaller());
+ return startRaftGroupServiceInternal(
+ groupId,
+ configuration,
+ opts.getCommandsMarshaller(),
+ StoppingExceptionFactories.indicateComponentStop()
+ );
} finally {
busyLock.leaveBusy();
}
@@ -352,7 +366,8 @@ public class Loza implements RaftManager {
ReplicationGroupId groupId,
PeersAndLearners configuration,
RaftServiceFactory<T> factory,
- @Nullable Marshaller commandsMarshaller
+ @Nullable Marshaller commandsMarshaller,
+ ExceptionFactory stoppingExceptionFactory
) throws NodeStoppingException {
if (!busyLock.enterBusy()) {
throw new NodeStoppingException();
@@ -363,7 +378,14 @@ public class Loza implements RaftManager {
commandsMarshaller = opts.getCommandsMarshaller();
}
- return factory.startRaftGroupService(groupId, configuration, raftConfiguration, executor, commandsMarshaller);
+ return factory.startRaftGroupService(
+ groupId,
+ configuration,
+ raftConfiguration,
+ executor,
+ commandsMarshaller,
+ stoppingExceptionFactory
+ );
} finally {
busyLock.leaveBusy();
}
@@ -424,15 +446,25 @@ public class Loza implements RaftManager {
RaftGroupListener lsnr,
RaftGroupEventsListener raftGrpEvtsLsnr,
RaftGroupOptions groupOptions,
- @Nullable RaftServiceFactory<T> raftServiceFactory
+ @Nullable RaftServiceFactory<T> raftServiceFactory,
+ ExceptionFactory stoppingExceptionFactory
) {
startRaftGroupNodeInternalWithoutService(nodeId, configuration, lsnr, raftGrpEvtsLsnr, groupOptions);
Marshaller cmdMarshaller = requireNonNullElse(groupOptions.commandsMarshaller(), opts.getCommandsMarshaller());
- return raftServiceFactory == null
- ? (T) startRaftGroupServiceInternal(nodeId.groupId(), configuration, cmdMarshaller)
- : raftServiceFactory.startRaftGroupService(nodeId.groupId(), configuration, raftConfiguration, executor, cmdMarshaller);
+ if (raftServiceFactory == null) {
+ return (T) startRaftGroupServiceInternal(nodeId.groupId(), configuration, cmdMarshaller, stoppingExceptionFactory);
+ } else {
+ return raftServiceFactory.startRaftGroupService(
+ nodeId.groupId(),
+ configuration,
+ raftConfiguration,
+ executor,
+ cmdMarshaller,
+ stoppingExceptionFactory
+ );
+ }
}
private void startRaftGroupNodeInternalWithoutService(
@@ -459,7 +491,8 @@ public class Loza implements RaftManager {
private RaftGroupService startRaftGroupServiceInternal(
ReplicationGroupId grpId,
PeersAndLearners membersConfiguration,
- Marshaller commandsMarshaller
+ Marshaller commandsMarshaller,
+ ExceptionFactory stoppingExceptionFactory
) {
return RaftGroupServiceImpl.start(
grpId,
@@ -468,7 +501,8 @@ public class Loza implements RaftManager {
raftConfiguration,
membersConfiguration,
executor,
- commandsMarshaller
+ commandsMarshaller,
+ stoppingExceptionFactory
);
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/NodeStoppingExceptionFactory.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/NodeStoppingExceptionFactory.java
new file mode 100644
index 00000000000..ccfce49a31e
--- /dev/null
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/NodeStoppingExceptionFactory.java
@@ -0,0 +1,36 @@
+/*
+ * 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.raft;
+
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.raft.service.RaftGroupService;
+
+/**
+ * Factory used to produce {@link NodeStoppingException}s to indicate that a {@link RaftGroupService} is stopping.
+ */
+public class NodeStoppingExceptionFactory implements ExceptionFactory {
+ @Override
+ public Exception create(String message) {
+ return new NodeStoppingException(message);
+ }
+
+ @Override
+ public Exception wrap(String message, Throwable cause) {
+ return new NodeStoppingException(message, cause);
+ }
+}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java
index 7e63938f6cb..5eb2f9a5c6d 100644
--- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java
@@ -23,6 +23,7 @@ import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.tostring.IgniteToStringBuilder.includeSensitive;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import static org.apache.ignite.raft.jraft.rpc.CliRequests.AddLearnersRequest;
@@ -48,6 +49,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
@@ -114,6 +116,8 @@ public class RaftGroupServiceImpl implements RaftGroupService {
private final Marshaller commandsMarshaller;
+ private final ExceptionFactory stoppingExceptionFactory;
+
/** Busy lock. */
private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
@@ -139,7 +143,8 @@ public class RaftGroupServiceImpl implements RaftGroupService {
PeersAndLearners membersConfiguration,
@Nullable Peer leader,
ScheduledExecutorService executor,
- Marshaller commandsMarshaller
+ Marshaller commandsMarshaller,
+ ExceptionFactory stoppingExceptionFactory
) {
this.cluster = cluster;
this.configuration = configuration;
@@ -151,6 +156,7 @@ public class RaftGroupServiceImpl implements RaftGroupService {
this.leader = leader;
this.executor = executor;
this.commandsMarshaller = commandsMarshaller;
+ this.stoppingExceptionFactory = stoppingExceptionFactory;
}
/**
@@ -172,6 +178,40 @@ public class RaftGroupServiceImpl implements RaftGroupService {
PeersAndLearners membersConfiguration,
ScheduledExecutorService executor,
Marshaller commandsMarshaller
+ ) {
+ return start(
+ groupId,
+ cluster,
+ factory,
+ configuration,
+ membersConfiguration,
+ executor,
+ commandsMarshaller,
+ StoppingExceptionFactories.indicateComponentStop()
+ );
+ }
+
+ /**
+ * Starts raft group service.
+ *
+ * @param groupId Raft group id.
+ * @param cluster Cluster service.
+ * @param factory Message factory.
+ * @param configuration Raft configuration.
+ * @param membersConfiguration Raft members configuration.
+ * @param executor Executor for retrying requests
+ * @param stoppingExceptionFactory Exception factory used to create exceptions thrown to indicate that the object is being stopped.
+ * @return A new Raft group service.
+ */
+ public static RaftGroupService start(
+ ReplicationGroupId groupId,
+ ClusterService cluster,
+ RaftMessagesFactory factory,
+ RaftConfiguration configuration,
+ PeersAndLearners membersConfiguration,
+ ScheduledExecutorService executor,
+ Marshaller commandsMarshaller,
+ ExceptionFactory stoppingExceptionFactory
) {
boolean inBenchmark = IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_SKIP_REPLICATION_IN_BENCHMARK);
@@ -185,7 +225,8 @@ public class RaftGroupServiceImpl implements RaftGroupService {
membersConfiguration,
null,
executor,
- commandsMarshaller
+ commandsMarshaller,
+ stoppingExceptionFactory
) {
@Override
public <R> CompletableFuture<R> run(Command cmd) {
@@ -201,7 +242,8 @@ public class RaftGroupServiceImpl implements RaftGroupService {
membersConfiguration,
null,
executor,
- commandsMarshaller
+ commandsMarshaller,
+ stoppingExceptionFactory
);
}
@@ -566,7 +608,7 @@ public class RaftGroupServiceImpl implements RaftGroupService {
*/
private <R extends NetworkMessage> void sendWithRetry(CompletableFuture<R> fut, RetryContext retryContext) {
if (!busyLock.enterBusy()) {
- fut.cancel(true);
+ fut.completeExceptionally(stoppingExceptionFactory.create("Raft client is stopping [" + groupId + "]."));
return;
}
@@ -611,7 +653,11 @@ public class RaftGroupServiceImpl implements RaftGroupService {
fut.complete((R) resp);
}
} catch (Throwable e) {
- fut.completeExceptionally(e);
+ if (hasCause(e, RejectedExecutionException.class)) {
+ fut.completeExceptionally(wrapInComponentStoppingException(e));
+ } else {
+ fut.completeExceptionally(e);
+ }
}
});
} finally {
@@ -619,11 +665,19 @@ public class RaftGroupServiceImpl implements RaftGroupService {
}
}
+ private Exception wrapInComponentStoppingException(Throwable e) {
+ return stoppingExceptionFactory.wrap("Raft client is stopping [" + groupId + "].", e);
+ }
+
private void handleThrowable(CompletableFuture<? extends NetworkMessage> fut, Throwable err, RetryContext retryContext) {
err = unwrapCause(err);
if (!recoverable(err)) {
- fut.completeExceptionally(err);
+ if (hasCause(err, RejectedExecutionException.class)) {
+ fut.completeExceptionally(wrapInComponentStoppingException(err));
+ } else {
+ fut.completeExceptionally(err);
+ }
return;
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/StoppingExceptionFactories.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/StoppingExceptionFactories.java
new file mode 100644
index 00000000000..98cdfb041c6
--- /dev/null
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/StoppingExceptionFactories.java
@@ -0,0 +1,45 @@
+/*
+ * 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.raft;
+
+import org.apache.ignite.internal.lang.ComponentStoppingException;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+
+/**
+ * Shortcut methods used to create {@link ExceptionFactory} instances to indicate that component cannot service requests anymore
+ * as it (or the node) is being stopped.
+ */
+public class StoppingExceptionFactories {
+ private static final ExceptionFactory INDICATE_COMPONENT_STOP = new ComponentStoppingExceptionFactory();
+
+ private static final ExceptionFactory INDICATE_NODE_STOP = new NodeStoppingExceptionFactory();
+
+ /**
+ * Returns a factory indicating component stop (using {@link ComponentStoppingException}).
+ */
+ public static ExceptionFactory indicateComponentStop() {
+ return INDICATE_COMPONENT_STOP;
+ }
+
+ /**
+ * Returns a factory indicating whole node stop (using {@link NodeStoppingException}).
+ */
+ public static ExceptionFactory indicateNodeStop() {
+ return INDICATE_NODE_STOP;
+ }
+}
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupService.java b/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupService.java
index d9328bb28e9..759561d326f 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupService.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupService.java
@@ -41,6 +41,7 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.raft.Command;
+import org.apache.ignite.internal.raft.ExceptionFactory;
import org.apache.ignite.internal.raft.LeaderElectionListener;
import org.apache.ignite.internal.raft.Marshaller;
import org.apache.ignite.internal.raft.Peer;
@@ -189,6 +190,7 @@ public class TopologyAwareRaftGroupService implements RaftGroupService {
* @param notifyOnSubscription Whether to notify callback after subscription to pass the current leader and term into it, even
* if the leader did not change in that moment (see {@link #subscribeLeader}).
* @param cmdMarshaller Marshaller that should be used to serialize/deserialize commands.
+ * @param stoppingExceptionFactory Exception factory used to create exceptions thrown to indicate that the object is being stopped.
* @return New Raft client.
*/
public static TopologyAwareRaftGroupService start(
@@ -201,14 +203,24 @@ public class TopologyAwareRaftGroupService implements RaftGroupService {
LogicalTopologyService logicalTopologyService,
RaftGroupEventsClientListener eventsClientListener,
boolean notifyOnSubscription,
- Marshaller cmdMarshaller
+ Marshaller cmdMarshaller,
+ ExceptionFactory stoppingExceptionFactory
) {
return new TopologyAwareRaftGroupService(
cluster,
factory,
executor,
raftConfiguration,
- RaftGroupServiceImpl.start(groupId, cluster, factory, raftConfiguration, configuration, executor, cmdMarshaller),
+ RaftGroupServiceImpl.start(
+ groupId,
+ cluster,
+ factory,
+ raftConfiguration,
+ configuration,
+ executor,
+ cmdMarshaller,
+ stoppingExceptionFactory
+ ),
logicalTopologyService,
eventsClientListener,
notifyOnSubscription
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupServiceFactory.java b/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupServiceFactory.java
index 785a76811c1..64cb663ffa9 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupServiceFactory.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/raft/client/TopologyAwareRaftGroupServiceFactory.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.raft.client;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.raft.ExceptionFactory;
import org.apache.ignite.internal.raft.Marshaller;
import org.apache.ignite.internal.raft.PeersAndLearners;
import org.apache.ignite.internal.raft.RaftServiceFactory;
@@ -67,7 +68,8 @@ public class TopologyAwareRaftGroupServiceFactory implements RaftServiceFactory<
PeersAndLearners peersAndLearners,
RaftConfiguration raftConfiguration,
ScheduledExecutorService raftClientExecutor,
- Marshaller commandsMarshaller
+ Marshaller commandsMarshaller,
+ ExceptionFactory stoppingExceptionFactory
) {
return TopologyAwareRaftGroupService.start(
groupId,
@@ -79,7 +81,8 @@ public class TopologyAwareRaftGroupServiceFactory implements RaftServiceFactory<
logicalTopologyService,
eventsClientListener,
true,
- commandsMarshaller
+ commandsMarshaller,
+ stoppingExceptionFactory
);
}
}
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java
index 9e689fd7d96..808fdc38b49 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java
@@ -24,10 +24,8 @@ import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
import static org.apache.ignite.internal.util.IgniteUtils.retryOperationUntilSuccess;
import java.util.UUID;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiFunction;
@@ -35,6 +33,7 @@ import org.apache.ignite.internal.failure.FailureContext;
import org.apache.ignite.internal.failure.FailureProcessor;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.ComponentStoppingException;
import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
@@ -145,9 +144,10 @@ public class PlacementDriverMessageProcessor {
if (!hasCause(
e,
NodeStoppingException.class,
- CancellationException.class,
+ ComponentStoppingException.class,
TrackerClosedException.class,
- RejectedExecutionException.class
+ // TODO: IGNITE-25206 - is it safe to ignore TimeoutException here?
+ TimeoutException.class
)) {
String errorMessage = String.format("Failed to process the lease granted message [msg=%s].", msg);
failureProcessor.process(new FailureContext(e, errorMessage));
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
index 16747b3e966..936db5a5ffa 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
@@ -46,7 +46,6 @@ import java.util.List;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
@@ -55,7 +54,6 @@ import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -70,6 +68,7 @@ import org.apache.ignite.internal.failure.FailureContext;
import org.apache.ignite.internal.failure.FailureProcessor;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.ComponentStoppingException;
import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.IgniteThrottledLogger;
@@ -1116,8 +1115,7 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
if (!hasCause(
ex,
NodeStoppingException.class,
- CancellationException.class,
- RejectedExecutionException.class,
+ ComponentStoppingException.class,
// Not a problem, there will be a retry.
TimeoutException.class
)) {
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java
index 11022130b5a..950ec0a0d09 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java
@@ -23,6 +23,7 @@ import static org.apache.ignite.internal.raft.PeersAndLearners.fromAssignments;
import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
import java.util.Map;
import java.util.UUID;
@@ -238,7 +239,7 @@ class ReplicaStateManager {
return partitionStarted;
}))
.whenComplete((res, ex) -> {
- if (ex != null) {
+ if (ex != null && !hasCause(ex, TransientReplicaStartException.class)) {
failureProcessor.process(new FailureContext(ex, String.format("Replica start failed [groupId=%s]", groupId)));
}
});
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/TransientReplicaStartException.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/TransientReplicaStartException.java
new file mode 100644
index 00000000000..5bac0ecef58
--- /dev/null
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/TransientReplicaStartException.java
@@ -0,0 +1,34 @@
+/*
+ * 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.replicator;
+
+import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
+
+import org.apache.ignite.internal.lang.IgniteInternalException;
+
+/**
+ * Thrown when a replica could not be started for some transient reason. 'Transient' here means that a partition might be restarted
+ * successfully in the future.
+ */
+public class TransientReplicaStartException extends IgniteInternalException {
+ private static final long serialVersionUID = 0L;
+
+ public TransientReplicaStartException(String msg) {
+ super(INTERNAL_ERR, msg);
+ }
+}
diff --git a/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/raft/client/AbstractTopologyAwareGroupServiceTest.java b/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/raft/client/AbstractTopologyAwareGroupServiceTest.java
index 11d7d33bd3e..c3ba909b984 100644
--- a/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/raft/client/AbstractTopologyAwareGroupServiceTest.java
+++ b/modules/replicator/src/testFixtures/java/org/apache/ignite/internal/raft/client/AbstractTopologyAwareGroupServiceTest.java
@@ -51,6 +51,7 @@ import org.apache.ignite.internal.raft.LeaderElectionListener;
import org.apache.ignite.internal.raft.Peer;
import org.apache.ignite.internal.raft.PeersAndLearners;
import org.apache.ignite.internal.raft.RaftNodeId;
+import org.apache.ignite.internal.raft.StoppingExceptionFactories;
import org.apache.ignite.internal.raft.TestRaftGroupListener;
import org.apache.ignite.internal.raft.configuration.RaftConfiguration;
import org.apache.ignite.internal.raft.server.RaftGroupOptions;
@@ -529,7 +530,8 @@ public abstract class AbstractTopologyAwareGroupServiceTest extends IgniteAbstra
logicalTopologyService,
eventsClientListener,
notifyOnSubscription,
- commandsMarshaller
+ commandsMarshaller,
+ StoppingExceptionFactories.indicateComponentStop()
);
}
diff --git a/modules/runner/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/WatchListenerInhibitor.java b/modules/runner/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/WatchListenerInhibitor.java
index af85f488805..b6876ab1e37 100644
--- a/modules/runner/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/WatchListenerInhibitor.java
+++ b/modules/runner/src/testFixtures/java/org/apache/ignite/internal/metastorage/server/WatchListenerInhibitor.java
@@ -72,7 +72,7 @@ public class WatchListenerInhibitor {
* Starts inhibiting events.
*/
public void startInhibit() {
- watchProcessor.enqueue(() -> inhibitFuture);
+ watchProcessor.enqueue(() -> inhibitFuture, () -> "<inhibiting>");
}
/**
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/PartitionReplicatorNodeRecovery.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/PartitionReplicatorNodeRecovery.java
index 1f0590548e8..544e849d2d0 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/PartitionReplicatorNodeRecovery.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/PartitionReplicatorNodeRecovery.java
@@ -37,7 +37,6 @@ import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.IntFunction;
-import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.network.TopologyEventHandler;
@@ -51,6 +50,7 @@ import org.apache.ignite.internal.partitiondistribution.Assignment;
import org.apache.ignite.internal.raft.Peer;
import org.apache.ignite.internal.raft.PeersAndLearners;
import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.TransientReplicaStartException;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.StorageClosedException;
@@ -216,9 +216,9 @@ class PartitionReplicatorNodeRecovery {
} else {
// No majority and not a full partition restart - need to restart nodes
// with current partition.
- String msg = "Unable to start partition " + partId + ". Majority not available.";
+ String msg = "Unable to start partition " + tablePartitionId + ". Majority not available.";
- throw new IgniteInternalException(msg);
+ throw new TransientReplicaStartException(msg);
}
});
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index cfcc654de84..ac50a77afc6 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -56,6 +56,7 @@ import static org.apache.ignite.internal.util.CompletableFutures.emptyListComple
import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
import static org.apache.ignite.internal.util.IgniteUtils.closeAllManually;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
@@ -177,6 +178,7 @@ import org.apache.ignite.internal.replicator.ReplicaManager.WeakReplicaStopReaso
import org.apache.ignite.internal.replicator.ReplicaService;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.TransientReplicaStartException;
import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.replicator.configuration.ReplicationConfiguration;
import org.apache.ignite.internal.replicator.listener.ReplicaListener;
@@ -1334,7 +1336,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
startReplicaSupplier,
forcedAssignments
).handle((res, ex) -> {
- if (ex != null) {
+ if (ex != null && !(hasCause(ex, TransientReplicaStartException.class))) {
String errorMessage = String.format(
"Unable to update raft groups on the node [tableId=%s, partitionId=%s]",
tableId,
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerRecoveryTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerRecoveryTest.java
index 286089c4ff4..b04b8ea0422 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerRecoveryTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerRecoveryTest.java
@@ -325,7 +325,7 @@ public class TableManagerRecoveryTest extends IgniteAbstractTest {
RaftGroupService raftGrpSrvcMock = mock(TopologyAwareRaftGroupService.class);
when(raftGrpSrvcMock.leader()).thenReturn(new Peer("node0"));
- when(rm.startRaftGroupService(any(), any(), any(), any())).thenAnswer(mock -> raftGrpSrvcMock);
+ when(rm.startRaftGroupService(any(), any(), any(), any(), any())).thenAnswer(mock -> raftGrpSrvcMock);
when(clusterService.messagingService()).thenReturn(mock(MessagingService.class));
when(clusterService.topologyService()).thenReturn(topologyService);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
index d08e6db2fd2..e3d0c8611bc 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
@@ -368,7 +368,7 @@ public class TableManagerTest extends IgniteAbstractTest {
*/
@Test
public void testPreconfiguredTable() throws Exception {
- when(rm.startRaftGroupService(any(), any(), any(), any()))
+ when(rm.startRaftGroupService(any(), any(), any(), any(), any()))
.thenAnswer(mock -> mock(TopologyAwareRaftGroupService.class));
TableManager tableManager = createTableManager(tblManagerFut);
@@ -700,7 +700,7 @@ public class TableManagerTest extends IgniteAbstractTest {
* partition storage is emulated instead.
*/
private void testStoragesGetClearedInMiddleOfFailedRebalance(boolean isTxStorageUnderRebalance) throws NodeStoppingException {
- when(rm.startRaftGroupService(any(), any(), any(), any()))
+ when(rm.startRaftGroupService(any(), any(), any(), any(), any()))
.thenAnswer(mock -> mock(TopologyAwareRaftGroupService.class));
when(distributionZoneManager.dataNodes(any(), anyInt(), anyInt()))
.thenReturn(completedFuture(Set.of(NODE_NAME)));
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java
index 2c80ef282ae..cbeb707357b 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryReconfigurationTest.java
@@ -114,6 +114,8 @@ import org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStat
import org.apache.ignite.internal.table.distributed.disaster.GlobalTablePartitionState;
import org.apache.ignite.internal.table.distributed.disaster.LocalTablePartitionStateByNode;
import org.apache.ignite.internal.testframework.WithSystemProperty;
+import org.apache.ignite.internal.testframework.failure.FailureManagerExtension;
+import org.apache.ignite.internal.testframework.failure.MuteFailureManagerLogging;
import org.apache.ignite.internal.util.ExceptionUtils;
import org.apache.ignite.lang.ErrorGroups.Replicator;
import org.apache.ignite.lang.IgniteException;
@@ -134,6 +136,7 @@ import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
/**
* Tests for scenarios where majority of peers is not available. In this class we frequently assert partition distributions, this means that
@@ -143,6 +146,7 @@ import org.junit.jupiter.api.Timeout;
@Timeout(120)
// TODO https://issues.apache.org/jira/browse/IGNITE-24332
@WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "false")
+@ExtendWith(FailureManagerExtension.class)
public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegrationTest {
/** Scale-down timeout. */
private static final int SCALE_DOWN_TIMEOUT_SECONDS = 2;
@@ -393,6 +397,7 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
@Test
@ZoneParams(nodes = 5, replicas = 3, partitions = 1)
+ @MuteFailureManagerLogging
void testManualRebalanceRecovery() throws Exception {
int partId = 0;
// Disable scale down to avoid unwanted rebalance.
@@ -473,6 +478,7 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
@Test
@ZoneParams(nodes = 4, replicas = 3, partitions = 1)
+ @MuteFailureManagerLogging
void testManualRebalanceRecoveryNoPending() throws Exception {
int partId = 0;
@@ -645,6 +651,7 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
@Test
@ZoneParams(nodes = 5, replicas = 3, partitions = 1)
+ @MuteFailureManagerLogging
public void testNewResetOverwritesFlags() throws Exception {
int partId = 0;
@@ -879,6 +886,7 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
*/
@Test
@ZoneParams(nodes = 6, replicas = 3, partitions = 1)
+ @MuteFailureManagerLogging
public void testIncompleteRebalanceBeforeAutomaticResetPartitions() throws Exception {
int partId = 0;
@@ -1217,6 +1225,7 @@ public class ItDisasterRecoveryReconfigurationTest extends ClusterPerTestIntegra
@Test
@ZoneParams(nodes = 6, replicas = 3, partitions = 1)
+ @MuteFailureManagerLogging
void testTwoPhaseResetOnEmptyNodes() throws Exception {
int partId = 0;
diff --git a/modules/workers/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java b/modules/workers/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java
index 0ddea48244d..32a927b5144 100644
--- a/modules/workers/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java
+++ b/modules/workers/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.worker;
import static org.apache.ignite.internal.failure.FailureType.CRITICAL_ERROR;
import static org.apache.ignite.internal.failure.FailureType.SYSTEM_WORKER_BLOCKED;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
-import static org.apache.ignite.lang.ErrorGroups.CriticalWorkers.SYSTEM_WORKER_BLOCKED_ERR;
import it.unimi.dsi.fastutil.longs.Long2LongMap;
import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap;
@@ -42,7 +41,6 @@ import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.manager.ComponentContext;
import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.worker.configuration.CriticalWorkersConfiguration;
-import org.apache.ignite.lang.IgniteException;
import org.jetbrains.annotations.Nullable;
/**
@@ -148,10 +146,7 @@ public class CriticalWorkerWatchdog implements CriticalWorkerRegistry, IgniteCom
appendThreadInfo(message, threadInfo);
- failureManager.process(
- new FailureContext(
- SYSTEM_WORKER_BLOCKED,
- new IgniteException(SYSTEM_WORKER_BLOCKED_ERR, message.toString(), false)));
+ failureManager.process(new FailureContext(SYSTEM_WORKER_BLOCKED, null, message.toString()));
}
}
}