You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2020/07/07 18:12:50 UTC
[GitHub] [ignite] glukos commented on a change in pull request #7970: IGNITE-13191 Public-facing API for waiting for backups on shutdown
glukos commented on a change in pull request #7970:
URL: https://github.com/apache/ignite/pull/7970#discussion_r451029022
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/CheckCpHistTask.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeJobResultPolicy;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointEntry;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistory;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.resources.LoggerResource;
+
+/**
+ * Task for test last checkpoint about applicable of history to all groups and partitions of parameters.
+ * If one of group or partition does not applicable force checkpoint will trigger.
Review comment:
Grammar:
**is** not applicable
force checkpoint will **be triggered** - passive voice
My version:
Task that checks whether last checkpoint is applicable for providing history for all groups and partitions that are passed as parameters. If at least one group or partition can't be supplied due to absence of last checkpoint, the task enforces a checkpoint to ensure possibility of the historical rebalancing.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/CheckCpHistTask.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeJobResultPolicy;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointEntry;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistory;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.resources.LoggerResource;
+
+/**
+ * Task for test last checkpoint about applicable of history to all groups and partitions of parameters.
+ * If one of group or partition does not applicable force checkpoint will trigger.
+ */
+@GridInternal
+public class CheckCpHistTask extends ComputeTaskAdapter<Map<UUID, Map<Integer, Set<Integer>>>, Boolean> {
+ /** Serial version id. */
+ private static final long serialVersionUID = 0L;
+
+ /** Reason of checkpoint, which can be triggered by this task. */
+ public static final String CP_REASON = "required by other node that shutdown was gracefully";
+
+ /** {@inheritDoc} */
+ @Override public Map<CheckCpHistClosureJob, ClusterNode> map(
+ List<ClusterNode> subgrid,
+ Map<UUID, Map<Integer, Set<Integer>>> arg
+ ) throws IgniteException {
+ Map<CheckCpHistClosureJob, ClusterNode> res = new HashMap<>();
+
+ for (ClusterNode node : subgrid) {
+ if (arg.containsKey(node.id()))
+ res.put(new CheckCpHistClosureJob(arg.get(node.id())), node);
+ }
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public ComputeJobResultPolicy result(
+ ComputeJobResult res,
+ List<ComputeJobResult> rcvd
+ ) throws IgniteException {
+ if (res.getException() != null)
+ return super.result(res, rcvd);
+
+ if (!(boolean)res.getData())
+ return ComputeJobResultPolicy.REDUCE;
+
+ return ComputeJobResultPolicy.WAIT;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Boolean reduce(List<ComputeJobResult> results) throws IgniteException {
+ for (ComputeJobResult result : results) {
+ if (!(boolean)result.getData())
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ * Job of checkpoint history task.
+ */
+ private static class CheckCpHistClosureJob implements ComputeJob {
+ /** Serial version id. */
+ private static final long serialVersionUID = 0L;
+
+ /** Logger. */
+ @LoggerResource
+ private IgniteLogger log;
+
+ /** Auto-inject ignite instance. */
+ @IgniteInstanceResource
+ private Ignite ignite;
+
+ /** Cancelled job flag. */
+ private volatile boolean cancelled;
+
+ /** list of group's ids. */
+ Map<Integer, Set<Integer>> grpIds;
+
+ /**
+ * @param grpIds List of ids.
+ */
+ public CheckCpHistClosureJob(Map<Integer, Set<Integer>> grpIds) {
+ this.grpIds = grpIds;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void cancel() {
+ cancelled = true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Boolean execute() throws IgniteException {
+ IgniteEx igniteEx = (IgniteEx)ignite;
+
+ if (igniteEx.context().cache().context().database() instanceof GridCacheDatabaseSharedManager) {
+ GridCacheSharedContext cctx = igniteEx.context().cache().context();
+ GridCacheDatabaseSharedManager databaseMng = (GridCacheDatabaseSharedManager)cctx.database();
+ CheckpointHistory cpHist = databaseMng.checkpointHistory();
+
+ CheckpointEntry lastCp = cpHist.lastCheckpoint();
+
+ try {
+
+ Map<Integer, CheckpointEntry.GroupState> states = lastCp.groupState(cctx);
+
+ for (Integer grpId : grpIds.keySet()) {
+
Review comment:
Redundant empty line
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/configuration/distributed/DistributedEnumProperty.java
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.configuration.distributed;
+
+import java.io.Serializable;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteClosure;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Wrapper for any enumerator.
Review comment:
Did you mean "enumerable"?
What does "serialized by its originality mean"? Did you mean "serialized by its ordinal"?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/configuration/distributed/DistributedEnumProperty.java
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.configuration.distributed;
+
+import java.io.Serializable;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteClosure;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Wrapper for any enumerator.
+ * It will be serialized by its originality into meta store.
+ *
+ * @param <T> Type of enum.
+ */
+public class DistributedEnumProperty<T extends Enum> implements DistributedChangeableProperty<T> {
+ /** This property stored enumerator as order. */
+ SimpleDistributedProperty<Integer> internal;
+
+ /** Function reflects an integer, which getting from meta storage, to an enumeration value. */
+ IgniteClosure<Integer, T> fromOrdinalFunc;
+
+ /** Function converts an enumeration value to an integer for stored in meta storage. */
+ IgniteClosure<T, Integer> toOredinalFunc;
+
+ /**
+ * Property constructor.
+ *
+ * @param name Name of property.
+ * @param fromOrdinalFunc Function reflects an integer to an enumiration value.
+ */
+ public DistributedEnumProperty(
+ String name,
+ IgniteClosure<Integer, T> fromOrdinalFunc
+ ) {
+ this(name, fromOrdinalFunc, (T value) -> {
+ return value == null ? null : value.ordinal();
+ });
+ }
+
+ /**
+ * Property constructor.
+ *
+ * @param name Name of property.
+ * @param fromOrdinalFunc Function reflects an integer to an enumiration value.
+ * @param toOredinalFunc Function converts an enumeration value to an integer.
+ */
+ public DistributedEnumProperty(
+ String name,
+ IgniteClosure<Integer, T> fromOrdinalFunc,
+ IgniteClosure<T, Integer> toOredinalFunc
+ ) {
+ this.internal = new SimpleDistributedProperty<>(name);
+ this.fromOrdinalFunc = fromOrdinalFunc;
+ this.toOredinalFunc = toOredinalFunc;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onAttached() {
+ internal.onAttached();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onReadyForUpdate(@NotNull PropertyUpdateClosure updater) {
+ internal.onReadyForUpdate(updater);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void localUpdate(Serializable newVal) {
+ if (newVal == null)
+ internal.localUpdate(null);
+ else if (newVal instanceof Enum)
+ internal.localUpdate(ordinalOrNull((T)newVal));
+ else if (newVal instanceof Integer)
+ internal.localUpdate(newVal);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean propagate(T newVal) throws IgniteCheckedException {
+ return internal.propagate(ordinalOrNull(newVal));
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridFutureAdapter<?> propagateAsync(T newVal) throws IgniteCheckedException {
+ return internal.propagateAsync(ordinalOrNull(newVal));
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridFutureAdapter<?> propagateAsync(T expectedVal, T newVal) throws IgniteCheckedException {
+ return internal.propagateAsync(ordinalOrNull(expectedVal), ordinalOrNull(newVal));
+ }
+
+ /** {@inheritDoc} */
+ @Override public T get() {
+ return fromOrdinalOrNull(internal.get());
+ }
+
+ /** {@inheritDoc} */
+ @Override public T getOrDefault(T dfltVal) {
+ T val = get();
+
+ return val == null ? dfltVal : val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String getName() {
+ return internal.getName();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void addListener(DistributePropertyListener<T> listener) {
+ internal.addListener(new DistributePropertyListener<Integer>() {
+ @Override public void onUpdate(String name, Integer oldVal, Integer newVal) {
+ listener.onUpdate(name, fromOrdinalOrNull(oldVal), fromOrdinalOrNull(newVal));
+ }
+ });
+ }
+
+ /**
+ * Determine ordinal by enum value, or return null if enum value is {@code null}.
+ *
+ * @param val Enum value.
+ * @return Ordinal or {@code null}.
+ */
+ private Integer ordinalOrNull(T val) {
+ return val == null ? null : toOredinalFunc.apply(val);
+ }
+
+ /**
+ * Return enum value or {@code null} if ordinal is {@code null} or less zero.
Review comment:
Return**s**
##########
File path: modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
##########
@@ -1151,6 +1159,29 @@ public IgniteConfiguration setSqlQueryHistorySize(int size) {
return this;
}
+ /**
+ * Get shutdown policy.
+ * If policy was not set default policy will be return {@code IgniteCluster.DEFAULT_SHUTDOWN_POLICY}.
+ *
+ * @return Shutdown policy.
+ */
+ public ShutdownPolicy getShutdownPolicy() {
+ return shutdown;
+ }
+
+ /**
+ * Set shutdown policy.
+ * If passed null through parameter, policy will be set as default.
Review comment:
Passive voice: If <code>null</code> is passed as a parameter,
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
##########
@@ -2623,20 +2657,44 @@ private static CacheConfiguration utilitySystemCache() {
*
* @param cancel Flag indicating whether all currently running jobs
* should be cancelled.
+ * @param shutdown This is a policy of shutdown which applied forcibly.
Review comment:
Policy of shutdown which is applied forcibly.
If this property is <code>null</code>, present policy of the cluster will be used.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
##########
@@ -299,11 +318,13 @@ public static IgniteState state(@Nullable String name) {
* default grid will be cancelled by calling {@link ComputeJob#cancel()}
* method. Note that just like with {@link Thread#interrupt()}, it is
* up to the actual job to exit from execution
+ * @param shutdown If this parameter sets explicitly this policy will use for stopping node.
Review comment:
Passive voice: If this parameter is set explicitly, this policy will be used for stopping node.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
##########
@@ -2694,6 +2877,85 @@ else if (grid0.context().invalid())
}
}
+ /**
+ * @param grpCtx Cahce group.
+ * @param nodesToExclude Nodes to exclude from check.
+ * @param proposedSuppliers Map of proposed suppliers for groups.
+ * @return True if all local partition of group specified have a copy in cluster, false otherwise.
+ */
+ private boolean haveCopyLoclaPartitions(
Review comment:
This method implicitly fills proposedSuppliers map, let's describe it in the doc.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/CheckCpHistTask.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeJobResultPolicy;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointEntry;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistory;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.resources.LoggerResource;
+
+/**
+ * Task for test last checkpoint about applicable of history to all groups and partitions of parameters.
+ * If one of group or partition does not applicable force checkpoint will trigger.
+ */
+@GridInternal
+public class CheckCpHistTask extends ComputeTaskAdapter<Map<UUID, Map<Integer, Set<Integer>>>, Boolean> {
Review comment:
Let's describe in doc what parameters are and what return value means.
I can guess that it's {nodeId, grpId -> {partId}}. But it's better when there's no need to guess.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GracefulShutdownTest.java
##########
@@ -0,0 +1,300 @@
+/*
+ * 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;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.ShutdownPolicy;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.GridCacheUtils;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.junit.Test;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
+
+/**
+ * Test checks various cluster shutdown and initiated policy.
+ */
+@WithSystemProperty(key = IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN, value = "false")
+public class GracefulShutdownTest extends GridCacheDhtPreloadWaitForBackupsWithPersistenceTest {
Review comment:
Please add at least one line to doc of every test method to describe test scenario.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/CheckCpHistTask.java
##########
@@ -0,0 +1,173 @@
+/*
+ * 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;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeJobResultPolicy;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointEntry;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointHistory;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.resources.LoggerResource;
+
+/**
+ * Task for test last checkpoint about applicable of history to all groups and partitions of parameters.
+ * If one of group or partition does not applicable force checkpoint will trigger.
+ */
+@GridInternal
+public class CheckCpHistTask extends ComputeTaskAdapter<Map<UUID, Map<Integer, Set<Integer>>>, Boolean> {
+ /** Serial version id. */
+ private static final long serialVersionUID = 0L;
+
+ /** Reason of checkpoint, which can be triggered by this task. */
+ public static final String CP_REASON = "required by other node that shutdown was gracefully";
Review comment:
"gracefully" is an adverb.
Adjectives are used to describe nouns - "was graceful" is correct.
I'd rephrase it as "required by another node which is performing a graceful shutdown".
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/configuration/distributed/DistributedEnumProperty.java
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.configuration.distributed;
+
+import java.io.Serializable;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteClosure;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Wrapper for any enumerator.
+ * It will be serialized by its originality into meta store.
+ *
+ * @param <T> Type of enum.
+ */
+public class DistributedEnumProperty<T extends Enum> implements DistributedChangeableProperty<T> {
+ /** This property stored enumerator as order. */
+ SimpleDistributedProperty<Integer> internal;
+
+ /** Function reflects an integer, which getting from meta storage, to an enumeration value. */
+ IgniteClosure<Integer, T> fromOrdinalFunc;
+
+ /** Function converts an enumeration value to an integer for stored in meta storage. */
+ IgniteClosure<T, Integer> toOredinalFunc;
+
+ /**
+ * Property constructor.
+ *
+ * @param name Name of property.
+ * @param fromOrdinalFunc Function reflects an integer to an enumiration value.
+ */
+ public DistributedEnumProperty(
+ String name,
+ IgniteClosure<Integer, T> fromOrdinalFunc
+ ) {
+ this(name, fromOrdinalFunc, (T value) -> {
+ return value == null ? null : value.ordinal();
+ });
+ }
+
+ /**
+ * Property constructor.
+ *
+ * @param name Name of property.
+ * @param fromOrdinalFunc Function reflects an integer to an enumiration value.
+ * @param toOredinalFunc Function converts an enumeration value to an integer.
+ */
+ public DistributedEnumProperty(
+ String name,
+ IgniteClosure<Integer, T> fromOrdinalFunc,
+ IgniteClosure<T, Integer> toOredinalFunc
+ ) {
+ this.internal = new SimpleDistributedProperty<>(name);
+ this.fromOrdinalFunc = fromOrdinalFunc;
+ this.toOredinalFunc = toOredinalFunc;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onAttached() {
+ internal.onAttached();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onReadyForUpdate(@NotNull PropertyUpdateClosure updater) {
+ internal.onReadyForUpdate(updater);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void localUpdate(Serializable newVal) {
+ if (newVal == null)
+ internal.localUpdate(null);
+ else if (newVal instanceof Enum)
+ internal.localUpdate(ordinalOrNull((T)newVal));
+ else if (newVal instanceof Integer)
+ internal.localUpdate(newVal);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean propagate(T newVal) throws IgniteCheckedException {
+ return internal.propagate(ordinalOrNull(newVal));
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridFutureAdapter<?> propagateAsync(T newVal) throws IgniteCheckedException {
+ return internal.propagateAsync(ordinalOrNull(newVal));
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridFutureAdapter<?> propagateAsync(T expectedVal, T newVal) throws IgniteCheckedException {
+ return internal.propagateAsync(ordinalOrNull(expectedVal), ordinalOrNull(newVal));
+ }
+
+ /** {@inheritDoc} */
+ @Override public T get() {
+ return fromOrdinalOrNull(internal.get());
+ }
+
+ /** {@inheritDoc} */
+ @Override public T getOrDefault(T dfltVal) {
+ T val = get();
+
+ return val == null ? dfltVal : val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String getName() {
+ return internal.getName();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void addListener(DistributePropertyListener<T> listener) {
+ internal.addListener(new DistributePropertyListener<Integer>() {
+ @Override public void onUpdate(String name, Integer oldVal, Integer newVal) {
+ listener.onUpdate(name, fromOrdinalOrNull(oldVal), fromOrdinalOrNull(newVal));
+ }
+ });
+ }
+
+ /**
+ * Determine ordinal by enum value, or return null if enum value is {@code null}.
Review comment:
Determine**s**, return**s**
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/configuration/distributed/DistributedEnumProperty.java
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.configuration.distributed;
+
+import java.io.Serializable;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.lang.IgniteClosure;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Wrapper for any enumerator.
+ * It will be serialized by its originality into meta store.
+ *
+ * @param <T> Type of enum.
+ */
+public class DistributedEnumProperty<T extends Enum> implements DistributedChangeableProperty<T> {
+ /** This property stored enumerator as order. */
+ SimpleDistributedProperty<Integer> internal;
Review comment:
Is absence of modifiers intentional? Fields can be private final.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
##########
@@ -2623,20 +2657,44 @@ private static CacheConfiguration utilitySystemCache() {
*
* @param cancel Flag indicating whether all currently running jobs
* should be cancelled.
+ * @param shutdown This is a policy of shutdown which applied forcibly.
+ * If this property is null policy will apply policy of whole cluster.
*/
- void stop(boolean cancel) {
+ void stop(boolean cancel, ShutdownPolicy shutdown) {
// Stop cannot be called prior to start from public API,
// since it checks for STARTED state. So, we can assert here.
assert startGuard.get();
- stop0(cancel);
+ if (shutdown == null)
+ shutdown = determineShutdownPolicy();
+
+ // If waiting for backups due to earlier invocation of stop(), stop wait and proceed shutting down.
+ if (shutdown == ShutdownPolicy.IMMEDIATE)
+ delayedShutdown = false;
+
+ stop0(cancel, shutdown);
}
/**
- * @param cancel Flag indicating whether all currently running jobs
+ * Read policy from meta storage or return a default value if it didn't find.
+ *
+ * @return Shutdown policy.
+ */
+ private ShutdownPolicy determineShutdownPolicy() {
+ if (IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN))
+ return ShutdownPolicy.GRACEFUL;
+
+ return grid.cluster().shutdownPolicy();
+ }
+
+ /**
+ * Stop instance synchronously according to parameters.
Review comment:
Stop**s**
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java
##########
@@ -330,6 +333,26 @@ public void testParseAndValidateWalActions() {
assertParseArgsThrows("Unexpected action " + rnd + " for " + WAL.text(), WAL.text(), rnd);
}
+ /**
+ * Tets checks a parser of shutdown policy command.
+ */
+ @Test
+ public void tesParsShutdownPolicyParameters() {
Review comment:
Typo: parse
##########
File path: modules/core/src/main/java/org/apache/ignite/IgniteCluster.java
##########
@@ -629,4 +629,19 @@
* @return Status of baseline auto-adjust.
*/
public BaselineAutoAdjustStatus baselineAutoAdjustStatus();
+
+ /**
+ * Return a policy of shutdown or default walue {@code IgniteConfiguration.DFLT_SHUTDOWN_POLICY}
+ * if property is not set.
+ *
+ * @return Shutdown policy.
+ */
+ public ShutdownPolicy shutdownPolicy();
+
+ /**
+ * Set a shutdown policy on a cluster.
Review comment:
Let's describe specifics of the behavior here.
Policy set by this method will override policy from IgniteConfiguration, user needs to know that.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GracefulShutdownTest.java
##########
@@ -0,0 +1,300 @@
+/*
+ * 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;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.ShutdownPolicy;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.GridCacheUtils;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.junit.Test;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
+
+/**
+ * Test checks various cluster shutdown and initiated policy.
+ */
+@WithSystemProperty(key = IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN, value = "false")
+public class GracefulShutdownTest extends GridCacheDhtPreloadWaitForBackupsWithPersistenceTest {
+
+ /** Shutdown policy of static configuration. */
+ public ShutdownPolicy policy = ShutdownPolicy.GRACEFUL;
+
+ /** Listening test logger. */
+ ListeningTestLogger listeningLog;
+
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ listeningLog = new ListeningTestLogger(log);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setGridLogger(listeningLog)
+ .setCommunicationSpi(new TestRecordingCommunicationSpi())
+ .setShutdownPolicy(policy);
+ }
+
+ /**
+ * Check static configuration of shutdown policy.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testRestartWithStaticConfiguredPolicy() throws Exception {
+ Ignite ignite0 = startGrid(0);
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ ignite0.close();
+
+ policy = ShutdownPolicy.IMMEDIATE;
+
+ ignite0 = startGrid(0);
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ShutdownPolicy.IMMEDIATE);
+ }
+
+ /**
+ * Test checked exception which is thrown when configuration of nodes different.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testTwoNodesWithDifferenConfuguration() throws Exception {
+ Ignite ignite0 = startGrid(0);
+
+ ignite0.cluster().active(true);
+
+ assertSame(ignite0.configuration().getShutdownPolicy(), ShutdownPolicy.GRACEFUL);
+
+ policy = ShutdownPolicy.IMMEDIATE;
+
+ GridTestUtils.assertThrowsAnyCause(log, () -> startGrid(1), IgniteCheckedException.class,
+ "Remote node has shutdoun policy different from local local");
+ }
+
+ /**
+ * Check dynamic configuration of shutdown policy.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testRestartWithDynamicConfiguredPolicy() throws Exception {
+ Ignite ignite0 = startGrid(0);
+
+ ignite0.cluster().active(true);
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ ShutdownPolicy configuredPolicy = ignite0.cluster().shutdownPolicy();
+
+ ShutdownPolicy policyToChange = null;
+
+ for (ShutdownPolicy policy : ShutdownPolicy.values()) {
+ if (policy != ignite0.cluster().shutdownPolicy())
+ policyToChange = policy;
+ }
+
+ assertNotNull(policyToChange);
+
+ ignite0.cluster().shutdownPolicy(policyToChange);
+
+ forceCheckpoint();
+
+ info("Policy to change: " + policyToChange);
+
+ ignite0.close();
+
+ ignite0 = startGrid(0);
+
+ info("Policy after restart: " + ignite0.cluster().shutdownPolicy());
+
+ assertNotSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ assertSame(ignite0.cluster().shutdownPolicy(), policyToChange);
+ }
+
+ /**
+ * Try to stop node when not all backups are matching of ideal assignment.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testNotIdialOwners() throws Exception {
+ backups = 1;
+
+ Ignite ignite0 = startGrid(0);
+
+ ignite0.cluster().active(true);
+
+ for (int i = 1; i <= 3; i++) {
+ IgniteCache cache = ignite0.cache("cache" + i);
+
+ assertNotNull(cache);
+
+ try (IgniteDataStreamer streamer = ignite0.dataStreamer("cache" + i)) {
+ for (int j = 0; j < 100; j++)
+ streamer.addData(j, j);
+ }
+ }
+
+ TestRecordingCommunicationSpi spi = TestRecordingCommunicationSpi.spi(ignite0);
+
+ spi.blockMessages((node, msg) -> {
+ String nodeName = (String)node.attributes().get(ATTR_IGNITE_INSTANCE_NAME);
+
+ if (msg instanceof GridDhtPartitionSupplyMessage) {
+ GridDhtPartitionSupplyMessage supplyMsg = (GridDhtPartitionSupplyMessage)msg;
+
+ if (supplyMsg.groupId() != CU.cacheId(GridCacheUtils.UTILITY_CACHE_NAME) &&
+ getTestIgniteInstanceName(1).equals(nodeName))
+ return true;
+ }
+
+ return false;
+ });
+
+ startGrid(1);
+ Ignite ignite2 = startGrid(2);
+
+ resetBaselineTopology();
+
+ spi.waitForBlocked();
+
+ for (CacheGroupContext grp: ((IgniteEx)ignite2).context().cache().cacheGroups()) {
+ GridTestUtils.waitForCondition(() ->
+ !grp.topology().partitionMap(false).get(((IgniteEx)ignite2).localNode().id()).hasMovingPartitions(), 30_000);
+ }
+
+ LogListener lnsr = LogListener.matches("This node is waiting for backups of local partitions for group")
+ .build();
+
+ listeningLog.registerListener(lnsr);
+
+ IgniteInternalFuture fut = GridTestUtils.runAsync(() -> {
+ ignite2.close();
+ });
+
+ assertTrue(GridTestUtils.waitForCondition(lnsr::check, 30_000));
+
+ assertFalse(fut.isDone());
+
+ spi.stopBlock();
+
+ assertTrue(GridTestUtils.waitForCondition(fut::isDone, 30_000));
+ }
+
+ /**
+ * Stopping node and start cache which does not allow it.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void tesStartCacheWheNodeStopping() throws Exception {
Review comment:
Typo: when
##########
File path: modules/core/src/main/java/org/apache/ignite/IgniteCluster.java
##########
@@ -629,4 +629,19 @@
* @return Status of baseline auto-adjust.
*/
public BaselineAutoAdjustStatus baselineAutoAdjustStatus();
+
+ /**
+ * Return a policy of shutdown or default walue {@code IgniteConfiguration.DFLT_SHUTDOWN_POLICY}
Review comment:
Return**s**
**v**alue
**the** property (specific property; it was mentioned a few words earlier)
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GracefulShutdownTest.java
##########
@@ -0,0 +1,300 @@
+/*
+ * 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;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.ShutdownPolicy;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.GridCacheUtils;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.junit.Test;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
+
+/**
+ * Test checks various cluster shutdown and initiated policy.
+ */
+@WithSystemProperty(key = IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN, value = "false")
+public class GracefulShutdownTest extends GridCacheDhtPreloadWaitForBackupsWithPersistenceTest {
+
+ /** Shutdown policy of static configuration. */
+ public ShutdownPolicy policy = ShutdownPolicy.GRACEFUL;
+
+ /** Listening test logger. */
+ ListeningTestLogger listeningLog;
+
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ listeningLog = new ListeningTestLogger(log);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setGridLogger(listeningLog)
+ .setCommunicationSpi(new TestRecordingCommunicationSpi())
+ .setShutdownPolicy(policy);
+ }
+
+ /**
+ * Check static configuration of shutdown policy.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testRestartWithStaticConfiguredPolicy() throws Exception {
+ Ignite ignite0 = startGrid(0);
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ ignite0.close();
+
+ policy = ShutdownPolicy.IMMEDIATE;
+
+ ignite0 = startGrid(0);
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ShutdownPolicy.IMMEDIATE);
+ }
+
+ /**
+ * Test checked exception which is thrown when configuration of nodes different.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testTwoNodesWithDifferenConfuguration() throws Exception {
+ Ignite ignite0 = startGrid(0);
+
+ ignite0.cluster().active(true);
+
+ assertSame(ignite0.configuration().getShutdownPolicy(), ShutdownPolicy.GRACEFUL);
+
+ policy = ShutdownPolicy.IMMEDIATE;
+
+ GridTestUtils.assertThrowsAnyCause(log, () -> startGrid(1), IgniteCheckedException.class,
+ "Remote node has shutdoun policy different from local local");
+ }
+
+ /**
+ * Check dynamic configuration of shutdown policy.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testRestartWithDynamicConfiguredPolicy() throws Exception {
+ Ignite ignite0 = startGrid(0);
+
+ ignite0.cluster().active(true);
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ ShutdownPolicy configuredPolicy = ignite0.cluster().shutdownPolicy();
+
+ ShutdownPolicy policyToChange = null;
+
+ for (ShutdownPolicy policy : ShutdownPolicy.values()) {
+ if (policy != ignite0.cluster().shutdownPolicy())
+ policyToChange = policy;
+ }
+
+ assertNotNull(policyToChange);
+
+ ignite0.cluster().shutdownPolicy(policyToChange);
+
+ forceCheckpoint();
+
+ info("Policy to change: " + policyToChange);
+
+ ignite0.close();
+
+ ignite0 = startGrid(0);
+
+ info("Policy after restart: " + ignite0.cluster().shutdownPolicy());
+
+ assertNotSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ assertSame(ignite0.cluster().shutdownPolicy(), policyToChange);
+ }
+
+ /**
+ * Try to stop node when not all backups are matching of ideal assignment.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testNotIdialOwners() throws Exception {
Review comment:
Typo: ideal
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GracefulShutdownTest.java
##########
@@ -0,0 +1,300 @@
+/*
+ * 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;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.ShutdownPolicy;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.cache.CacheGroupContext;
+import org.apache.ignite.internal.processors.cache.GridCacheUtils;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.LogListener;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.junit.Test;
+
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
+
+/**
+ * Test checks various cluster shutdown and initiated policy.
+ */
+@WithSystemProperty(key = IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN, value = "false")
+public class GracefulShutdownTest extends GridCacheDhtPreloadWaitForBackupsWithPersistenceTest {
+
+ /** Shutdown policy of static configuration. */
+ public ShutdownPolicy policy = ShutdownPolicy.GRACEFUL;
+
+ /** Listening test logger. */
+ ListeningTestLogger listeningLog;
+
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ listeningLog = new ListeningTestLogger(log);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setGridLogger(listeningLog)
+ .setCommunicationSpi(new TestRecordingCommunicationSpi())
+ .setShutdownPolicy(policy);
+ }
+
+ /**
+ * Check static configuration of shutdown policy.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testRestartWithStaticConfiguredPolicy() throws Exception {
+ Ignite ignite0 = startGrid(0);
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ ignite0.close();
+
+ policy = ShutdownPolicy.IMMEDIATE;
+
+ ignite0 = startGrid(0);
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ignite0.configuration().getShutdownPolicy());
+
+ assertSame(ignite0.cluster().shutdownPolicy(), ShutdownPolicy.IMMEDIATE);
+ }
+
+ /**
+ * Test checked exception which is thrown when configuration of nodes different.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testTwoNodesWithDifferenConfuguration() throws Exception {
Review comment:
Typo: different
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
##########
@@ -545,4 +545,13 @@ public void onPartitionEvicted(GridDhtLocalPartition part, boolean updateSeq) {
@Override public void resume() {
busyLock.writeLock().unlock();
}
+
+ /**
+ * Return supplier.
Review comment:
Return**s**
##########
File path: modules/core/src/main/java/org/apache/ignite/IgniteCluster.java
##########
@@ -629,4 +629,19 @@
* @return Status of baseline auto-adjust.
*/
public BaselineAutoAdjustStatus baselineAutoAdjustStatus();
+
+ /**
+ * Return a policy of shutdown or default walue {@code IgniteConfiguration.DFLT_SHUTDOWN_POLICY}
+ * if property is not set.
+ *
+ * @return Shutdown policy.
+ */
+ public ShutdownPolicy shutdownPolicy();
+
+ /**
+ * Set a shutdown policy on a cluster.
Review comment:
Set**s**
##########
File path: modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java
##########
@@ -1151,6 +1159,29 @@ public IgniteConfiguration setSqlQueryHistorySize(int size) {
return this;
}
+ /**
+ * Get shutdown policy.
Review comment:
Get**s**, Set**s**
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
##########
@@ -299,11 +318,13 @@ public static IgniteState state(@Nullable String name) {
* default grid will be cancelled by calling {@link ComputeJob#cancel()}
* method. Note that just like with {@link Thread#interrupt()}, it is
* up to the actual job to exit from execution
+ * @param shutdown If this parameter sets explicitly this policy will use for stopping node.
+ * If this parameter is {@code null} common cluster policy will use.
Review comment:
If this parameter is {@code null}, common cluster policy will be used.
##########
File path: modules/core/src/main/java/org/apache/ignite/ShutdownPolicy.java
##########
@@ -0,0 +1,92 @@
+/*
+ * 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;
+
+public enum ShutdownPolicy {
+ /**
+ * Stop immediately as soon as all components are ready.
+ */
+ IMMEDIATE(0),
+
+ /**
+ * Node will stop if and only if it does not store any unique partitions, that does not have copies on cluster.
Review comment:
, that don't have another copies in the cluster.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
##########
@@ -2623,20 +2657,44 @@ private static CacheConfiguration utilitySystemCache() {
*
* @param cancel Flag indicating whether all currently running jobs
* should be cancelled.
+ * @param shutdown This is a policy of shutdown which applied forcibly.
+ * If this property is null policy will apply policy of whole cluster.
*/
- void stop(boolean cancel) {
+ void stop(boolean cancel, ShutdownPolicy shutdown) {
// Stop cannot be called prior to start from public API,
// since it checks for STARTED state. So, we can assert here.
assert startGuard.get();
- stop0(cancel);
+ if (shutdown == null)
+ shutdown = determineShutdownPolicy();
+
+ // If waiting for backups due to earlier invocation of stop(), stop wait and proceed shutting down.
+ if (shutdown == ShutdownPolicy.IMMEDIATE)
+ delayedShutdown = false;
+
+ stop0(cancel, shutdown);
}
/**
- * @param cancel Flag indicating whether all currently running jobs
+ * Read policy from meta storage or return a default value if it didn't find.
+ *
+ * @return Shutdown policy.
+ */
+ private ShutdownPolicy determineShutdownPolicy() {
+ if (IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_WAIT_FOR_BACKUPS_ON_SHUTDOWN))
+ return ShutdownPolicy.GRACEFUL;
+
+ return grid.cluster().shutdownPolicy();
+ }
+
+ /**
+ * Stop instance synchronously according to parameters.
+ *
+ * @param cancel Flag indicating whether all currently running job
* should be cancelled.
+ * @param shutdown Policy of according which shutdown will be produced.
Review comment:
Policy according to which shutdown will be performed.
##########
File path: modules/core/src/main/java/org/apache/ignite/ShutdownPolicy.java
##########
@@ -0,0 +1,92 @@
+/*
+ * 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;
+
+public enum ShutdownPolicy {
Review comment:
Let's describe shortly in class header what this property is responsible for.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
##########
@@ -2694,6 +2877,85 @@ else if (grid0.context().invalid())
}
}
+ /**
+ * @param grpCtx Cahce group.
+ * @param nodesToExclude Nodes to exclude from check.
+ * @param proposedSuppliers Map of proposed suppliers for groups.
+ * @return True if all local partition of group specified have a copy in cluster, false otherwise.
+ */
+ private boolean haveCopyLoclaPartitions(
+ CacheGroupContext grpCtx,
+ Set<UUID> nodesToExclude,
+ Map<UUID, Map<Integer, Set<Integer>>> proposedSuppliers
+ ) {
+ GridDhtPartitionFullMap fullMap = grpCtx.topology().partitionMap(false);
+
+ if (fullMap == null)
+ return false;
+
+ UUID localNodeId = grid.getLocalNodeId();
+
+ GridDhtPartitionMap localPartMap = fullMap.get(localNodeId);
+
+ int parts = grpCtx.topology().partitions();
+
+ List<List<ClusterNode>> idealAssignment = grpCtx.affinity().idealAssignmentRaw();
+
+ for (int p = 0; p < parts; p++) {
+ if (localPartMap.get(p) != GridDhtPartitionState.OWNING)
+ continue;
+
+ boolean foundCopy = false;
+
+ for (Map.Entry<UUID, GridDhtPartitionMap> entry : fullMap.entrySet()) {
+ if (localNodeId.equals(entry.getKey()) || nodesToExclude.contains(entry.getKey()))
+ continue;
+
+ //This remote node does not present in ideal assignment.
+ if (!idealAssignment.get(p).stream().filter(node -> node.id().equals(entry.getKey())).findAny().isPresent())
Review comment:
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
##########
@@ -2623,20 +2657,44 @@ private static CacheConfiguration utilitySystemCache() {
*
* @param cancel Flag indicating whether all currently running jobs
* should be cancelled.
+ * @param shutdown This is a policy of shutdown which applied forcibly.
+ * If this property is null policy will apply policy of whole cluster.
*/
- void stop(boolean cancel) {
+ void stop(boolean cancel, ShutdownPolicy shutdown) {
// Stop cannot be called prior to start from public API,
// since it checks for STARTED state. So, we can assert here.
assert startGuard.get();
- stop0(cancel);
+ if (shutdown == null)
+ shutdown = determineShutdownPolicy();
+
+ // If waiting for backups due to earlier invocation of stop(), stop wait and proceed shutting down.
+ if (shutdown == ShutdownPolicy.IMMEDIATE)
+ delayedShutdown = false;
+
+ stop0(cancel, shutdown);
}
/**
- * @param cancel Flag indicating whether all currently running jobs
+ * Read policy from meta storage or return a default value if it didn't find.
Review comment:
Reads policy from distributed meta storage or returns a default value if it isn't present.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
##########
@@ -2661,6 +2719,131 @@ private synchronized void stop0(boolean cancel) {
if (log != null && log.isDebugEnabled())
log.debug("Shutdown is in progress (ignoring): " + e.getMessage());
}
+ }
+
+ if (shutdown == ShutdownPolicy.GRACEFUL && !grid.context().clientNode() && grid.cluster().active()) {
+ delayedShutdown = true;
+
+ if (log.isInfoEnabled())
+ log.info("Ensuring that caches have sufficient backups and local rebalance completion...");
+
+ DistributedMetaStorage metaStorage = grid.context().distributedMetastorage();
+
+ while (delayedShutdown) {
+ boolean safeToStop = true;
+
+ long topVer = grid.cluster().topologyVersion();
+
+ HashSet<UUID> originalNodesToExclude;
+
+ HashSet<UUID> nodesToExclude;
+
+ try {
+ originalNodesToExclude = metaStorage.read(GRACEFUL_SHUTDOWN_METASTORE_KEY);
+
+ nodesToExclude = originalNodesToExclude != null ? new HashSet<>(originalNodesToExclude) :
+ new HashSet<>();
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Unable to read " + GRACEFUL_SHUTDOWN_METASTORE_KEY +
+ " value from metastore.", e);
+
+ continue;
+ }
+
+ Map<UUID, Map<Integer, Set<Integer>>> proposedSuppliers = new HashMap<>();
+
+ for (CacheGroupContext grpCtx : grid.context().cache().cacheGroups()) {
+ if (grpCtx.isLocal() || grpCtx.systemCache())
+ continue;
+
+ if (grpCtx.config().getCacheMode() == PARTITIONED && grpCtx.config().getBackups() == 0) {
+ LT.warn(log, "Ignoring potential data loss on cache without backups [name="
+ + grpCtx.cacheOrGroupName() + "]");
+
+ continue;
+ }
+
+ if (topVer != grpCtx.topology().readyTopologyVersion().topologyVersion()) {
+ // At the moment, there is an exchange.
+ safeToStop = false;
+
+ break;
+ }
+
+ GridDhtPartitionFullMap fullMap = grpCtx.topology().partitionMap(false);
+
+ if (fullMap == null) {
+ safeToStop = false;
+
+ break;
+ }
+
+ nodesToExclude.retainAll(fullMap.keySet());
+
+ if (!haveCopyLoclaPartitions(grpCtx, nodesToExclude, proposedSuppliers)) {
Review comment:
Typo: local
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/commandline/shutdown/ShutdownPolicyArgument.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.commandline.shutdown;
+
+import org.apache.ignite.ShutdownPolicy;
+
+/**
+ * Srgumants of shutdown policy command.
Review comment:
Arguments ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org