You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "PakhomovAlexander (via GitHub)" <gi...@apache.org> on 2023/12/26 09:57:45 UTC
[PR] IGNITE-20850 Add compute job failover [ignite-3]
PakhomovAlexander opened a new pull request, #2990:
URL: https://github.com/apache/ignite-3/pull/2990
* ComputeJobFailover wraps the returned by IgniteComputeImpl
Future. Failover captures the context of the job: workerNode,
deployment units, jobClassName, etc.
* ComputeJobFailover listens to topology events and if the
worker node has left the topology, the job will be restarted
on one of the failover nodes. The future returned by
IgniteComputeImpl will be completed/canceled when the
restarted job is completed/canceled.
--
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.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20850 Add compute job failover [ignite-3]
Posted by "PakhomovAlexander (via GitHub)" <gi...@apache.org>.
PakhomovAlexander commented on code in PR #2990:
URL: https://github.com/apache/ignite-3/pull/2990#discussion_r1447165197
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeImpl.java:
##########
@@ -75,7 +98,10 @@ public <R> JobExecution<R> executeAsync(Set<ClusterNode> nodes, List<DeploymentU
throw new IllegalArgumentException("nodes must not be empty.");
}
- return new JobExecutionWrapper<>(executeOnOneNode(randomNode(nodes), units, jobClassName, args));
+ Set<ClusterNode> candidates = new HashSet<>(nodes);
+ ClusterNode targetNode = randomNode(candidates);
Review Comment:
Good catch!
--
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.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20850 Add compute job failover [ignite-3]
Posted by "valepakh (via GitHub)" <gi...@apache.org>.
valepakh commented on code in PR #2990:
URL: https://github.com/apache/ignite-3/pull/2990#discussion_r1447185568
##########
modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItWorkerShutdownTest.java:
##########
@@ -0,0 +1,563 @@
+/*
+ * 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.compute;
+
+import static java.util.stream.Collectors.toList;
+import static org.apache.ignite.compute.JobState.CANCELED;
+import static org.apache.ignite.compute.JobState.COMPLETED;
+import static org.apache.ignite.compute.JobState.EXECUTING;
+import static org.apache.ignite.compute.JobState.FAILED;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.in;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.notNullValue;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.compute.JobStatus;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.table.Tuple;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Integration tests for worker node shutdown failover.
+ *
+ * <p>The logic is that if we run the job on remote node and this node has left the topology then we should restart a job on
+ * another node. This is not true for broadcast and local jobs. They should not be restarted.
+ */
+@SuppressWarnings("resource")
+class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
+ /**
+ * ACK for {@link Signal#CONTINUE}. Returned by a job that has received the signal. Used to check that the job is alive.
+ */
+ private static final Object ack = new Object();
+
+ /**
+ * Map from node name to node index in {@link super#cluster}.
+ */
+ private static final Map<String, Integer> NODES_NAMES_TO_INDEXES = new HashMap<>();
+
+ /**
+ * Class-wide queue that is used as a communication channel between {@link GlobalInteractiveJob} and test code. You can send a signal to
+ * the job via this channel and get a response from the job via {@link #GLOBAL_CHANNEL}.
+ */
+ private static BlockingQueue<Signal> GLOBAL_SIGNALS = new LinkedBlockingQueue<>();
+
+ /**
+ * Class-wide queue that is used as a communication channel between {@link GlobalInteractiveJob} and test code. You can send a signal to
+ * the job via {@link #GLOBAL_SIGNALS} and get a response from the job via this channel.
+ */
+ private static BlockingQueue<Object> GLOBAL_CHANNEL = new LinkedBlockingQueue<>();
+
+ /**
+ * Node-specific queues that are used as a communication channel between {@link InteractiveJob} and test code. The semantics are the
+ * same as for {@link #GLOBAL_SIGNALS} except that each node has its own queue. So, test code can communicate with a
+ * {@link InteractiveJob} that is running on specific node.
+ */
+ private static Map<String, BlockingQueue<Signal>> NODE_SIGNALS = new ConcurrentHashMap<>();
+
+ /**
+ * Node-specific queues that are used as a communication channel between {@link InteractiveJob} and test code. The semantics are the
+ * same as for {@link #GLOBAL_CHANNEL} except that each node has its own queue. So, test code can communicate with a
+ * {@link InteractiveJob} that is running on specific node.
+ */
+ private static Map<String, BlockingQueue<Object>> NODE_CHANNELS = new ConcurrentHashMap<>();
+
+ /**
+ * Node-specific counters that are used to count how many times {@link InteractiveJob} has been run on specific node.
+ */
+ private static Map<String, Integer> INTERACTIVE_JOB_RUN_TIMES = new ConcurrentHashMap<>();
+
+ private static void checkAllInteractiveJobsCalledOnce() {
+ INTERACTIVE_JOB_RUN_TIMES.forEach((nodeName, runTimes) -> assertThat(runTimes, equalTo(1)));
+ }
+
+ private static void finishAllInteractiveJobs() {
+ NODE_SIGNALS.forEach((nodeName, channel) -> {
+ try {
+ channel.offer(Signal.RETURN, 10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ private static void initChannels(List<String> nodes) {
+ for (String nodeName : nodes) {
+ NODE_CHANNELS.put(nodeName, new LinkedBlockingQueue<>());
+ NODE_SIGNALS.put(nodeName, new LinkedBlockingQueue<>());
+ INTERACTIVE_JOB_RUN_TIMES.put(nodeName, 0);
+ }
+ }
+
+ private static Set<String> workerCandidates(IgniteImpl... nodes) {
+ return Arrays.stream(nodes)
+ .map(IgniteImpl::node)
+ .map(ClusterNode::name)
+ .collect(Collectors.toSet());
+ }
+
+ private static void finishGlobalJob() {
+ GLOBAL_SIGNALS.offer(Signal.RETURN);
+ }
+
+ private static void checkGlobalInteractiveJobAlive(JobExecution<?> execution)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ GLOBAL_SIGNALS.offer(Signal.CONTINUE);
+ assertThat(GLOBAL_CHANNEL.poll(10, TimeUnit.SECONDS), equalTo(ack));
+
+ assertThat(execution.resultAsync().isDone(), equalTo(false));
+ assertThat(idSync(execution), notNullValue());
+
+ // During the fob failover we might get a job that is restarted, the state will be not EXECUTING for some short time.
+ await().until(() -> execution.statusAsync().get(10, TimeUnit.SECONDS).state() == EXECUTING);
+ }
+
+ private static void checkInteractiveJobAlive(ClusterNode clusterNode, JobExecution<?> execution) {
+ NODE_SIGNALS.get(clusterNode.name()).offer(Signal.CONTINUE);
+ try {
+ assertThat(NODE_CHANNELS.get(clusterNode.name()).poll(10, TimeUnit.SECONDS), equalTo(ack));
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ assertThat(execution.resultAsync().isDone(), equalTo(false));
+ }
+
+ private static JobStatus statusSync(JobExecution<String> execution) throws InterruptedException, ExecutionException, TimeoutException {
+ return execution.statusAsync().get(10, TimeUnit.SECONDS);
+ }
+
+ @NotNull
+ private Set<ClusterNode> clusterNodesByNames(Set<String> nodes) {
+ return nodes.stream()
+ .map(NODES_NAMES_TO_INDEXES::get)
+ .map(this::node)
+ .map(IgniteImpl::node)
+ .collect(Collectors.toSet());
+ }
+
+ /**
+ * Initializes channels. Assumption: there is no any running job on the cluster.
+ */
+ @BeforeEach
+ void setUp() {
+ GLOBAL_SIGNALS.clear();
+ GLOBAL_CHANNEL.clear();
+ NODE_SIGNALS.clear();
+ NODE_CHANNELS.clear();
+ INTERACTIVE_JOB_RUN_TIMES.clear();
+
+ for (int i = 0; i < 3; i++) {
+ NODES_NAMES_TO_INDEXES.put(node(i).name(), i);
+ }
+
+ executeSql("DROP TABLE IF EXISTS PUBLIC.TEST");
+ }
+
+ @Test
+ void remoteExecutionWorkerShutdown() throws Exception {
+ // Given entry node.
+ IgniteImpl entryNode = node(0);
+ // And remote candidates to execute a job.
+ Set<String> remoteWorkerCandidates = workerCandidates(node(1), node(2));
+
+ // When execute job.
+ JobExecution<String> execution = executeGlobalInteractiveJob(entryNode, remoteWorkerCandidates);
+
+ // Then one of candidates became a worker and run the job.
+ String workerNodeName = getWorkerNodeNameFromGlobalInteractiveJob();
+ // And job is running.
+ checkGlobalInteractiveJobAlive(execution);
+
+ // And save state BEFORE worker has failed.
+ long createTimeBeforeFail = statusSync(execution).createTime().toEpochMilli();
+ long startTimeBeforeFail = statusSync(execution).startTime().toEpochMilli();
+ UUID jobIdBeforeFail = idSync(execution);
+
+ // When stop worker node.
+ stopNode(workerNodeName);
+ // And remove it from candidates.
+ remoteWorkerCandidates.remove(workerNodeName);
+
+ // Then the job is alive: it has been restarted on another candidate.
+ checkGlobalInteractiveJobAlive(execution);
+ // And remaining candidate was chosen as a failover worker.
+ String failoverWorker = getWorkerNodeNameFromGlobalInteractiveJob();
+ assertThat(remoteWorkerCandidates, hasItem(failoverWorker));
+
+ // And check create time was not changed but start time changed.
+ long createTimeAfterFail = statusSync(execution).createTime().toEpochMilli();
+ long startTimeAfterFail = statusSync(execution).startTime().toEpochMilli();
+ assertThat(createTimeAfterFail, equalTo(createTimeBeforeFail));
+ assertThat(startTimeAfterFail, greaterThan(startTimeBeforeFail));
+ // And id was not changed
+ assertThat(idSync(execution), equalTo(jobIdBeforeFail));
+
+ // When finish job.
+ finishGlobalJob();
+
+ // Then it is successfully finished.
+ assertThat(execution.resultAsync().get(10, TimeUnit.SECONDS), equalTo("Done"));
+ // And.
+ assertThat(execution.statusAsync().get().state(), is(COMPLETED));
+ // And finish time is greater then create time and start time.
+ long finishTime = execution.statusAsync().get().finishTime().toEpochMilli();
+ assertThat(finishTime, greaterThan(createTimeAfterFail));
+ assertThat(finishTime, greaterThan(startTimeAfterFail));
+ // And job id the same
+ assertThat(idSync(execution), equalTo(jobIdBeforeFail));
+ }
+
+ private static UUID idSync(JobExecution<?> execution) throws InterruptedException, ExecutionException, TimeoutException {
+ return execution.idAsync().get(10, TimeUnit.SECONDS);
+ }
+
+ @Test
+ void remoteExecutionSingleWorkerShutdown() throws Exception {
+ // Given.
+ IgniteImpl entryNode = node(0);
+ // And only one remote candidate to execute a job.
+ Set<String> remoteWorkerCandidates = workerCandidates(node(1));
+
+ // When execute job.
+ JobExecution<String> execution = executeGlobalInteractiveJob(entryNode, remoteWorkerCandidates);
+
+ // Then the job is running on worker node.
+ String workerNodeName = getWorkerNodeNameFromGlobalInteractiveJob();
+ assertThat(remoteWorkerCandidates, hasItem(workerNodeName));
+ // And.
+ checkGlobalInteractiveJobAlive(execution);
+
+ // When stop worker node.
+ stopNode(workerNodeName);
+
+ // Then the job is failed, because there is no any failover worker.
+ await().until(() -> execution.resultAsync().isCompletedExceptionally());
Review Comment:
Is it worth using `CompletableFutureExceptionMatcher.willThrow` in this cases?
--
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.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20850 Add compute job failover [ignite-3]
Posted by "PakhomovAlexander (via GitHub)" <gi...@apache.org>.
PakhomovAlexander merged PR #2990:
URL: https://github.com/apache/ignite-3/pull/2990
--
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.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20850 Add compute job failover [ignite-3]
Posted by "valepakh (via GitHub)" <gi...@apache.org>.
valepakh commented on code in PR #2990:
URL: https://github.com/apache/ignite-3/pull/2990#discussion_r1447402606
##########
modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItWorkerShutdownTest.java:
##########
@@ -149,6 +150,7 @@ private static void checkGlobalInteractiveJobAlive(JobExecution<?> execution)
assertThat(GLOBAL_CHANNEL.poll(10, TimeUnit.SECONDS), equalTo(ack));
assertThat(execution.resultAsync().isDone(), equalTo(false));
+ System.out.println("%%%%% () checking id");
Review Comment:
```suggestion
```
--
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.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20850 Add compute job failover [ignite-3]
Posted by "valepakh (via GitHub)" <gi...@apache.org>.
valepakh commented on code in PR #2990:
URL: https://github.com/apache/ignite-3/pull/2990#discussion_r1447109091
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeJobFailover.java:
##########
@@ -0,0 +1,158 @@
+/*
+ * 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.compute;
+
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import org.apache.ignite.compute.DeploymentUnit;
+import org.apache.ignite.compute.JobExecution;
+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.lang.ErrorGroups.Compute;
+import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * This is a helper class for {@link ComputeComponent} to handle job failures. You can think about this class as a "retryable compute job
+ * with captured context". Retry logic is applied ONLY if the worker node leaves the cluster. If the job itself is failing, then the
+ * exception is propagated to the caller and this class does not handle it.
+ *
+ * <p>If you want to execute a job on node1 and use node2 and node3 as failover candidates,
+ * then you should create an instance of this class with workerNode = node1, failoverCandidates = [node2, node3] as arguments and
+ * call {@link #failSafeExecute()}.
+ *
+ * @param <T> the type of the result of the job.
+ */
+class ComputeJobFailover<T> {
+ private static final IgniteLogger LOG = Loggers.forClass(ComputeJobFailover.class);
+
+ /**
+ * Compute component that is called when the {@link #runningWorkerNode} has left the cluster.
+ */
+ private final ComputeComponent computeComponent;
+
+ /**
+ * The failover listens this event source to know when the {@link #runningWorkerNode} has left the cluster.
+ */
+ private final NodeLeftEventsSource nodeLeftEventsSource;
+
+ /**
+ * Set of worker candidates in case {@link #runningWorkerNode} has left the cluster.
+ */
+ private final ConcurrentLinkedDeque<ClusterNode> failoverCandidates;
+
+ /**
+ * The node where the job is being executed at a given moment. If node leaves the cluster, then the job is restarted on one of the
+ * {@link #failoverCandidates} and the reference is CASed to the new node.
+ */
+ private final AtomicReference<ClusterNode> runningWorkerNode;
+
+ /** Context of the called job. Captures deployment units, jobClassName and arguments. */
+ private final RemoteExecutionContext<T> jobContext;
+
+ /**
+ * Creates a per-job instance.
+ *
+ * @param computeComponent compute component.
+ * @param nodeLeftEventsSource node left events source, used as dynamic topology service (allows to remove handlers).
+ * @param workerNode the node to execute the job on.
+ * @param failoverCandidates the set of nodes where the job can be restarted if the worker node leaves the cluster.
+ * @param units deployment units.
+ * @param jobClassName the name of the job class.
+ * @param args the arguments of the job.
+ */
+ ComputeJobFailover(
+ ComputeComponent computeComponent,
+ NodeLeftEventsSource nodeLeftEventsSource,
+ ClusterNode workerNode,
+ Set<ClusterNode> failoverCandidates,
+ List<DeploymentUnit> units,
+ String jobClassName,
+ Object... args
+ ) {
+ this.computeComponent = computeComponent;
+ this.nodeLeftEventsSource = nodeLeftEventsSource;
+ this.runningWorkerNode = new AtomicReference<>(workerNode);
+ this.failoverCandidates = new ConcurrentLinkedDeque<>(failoverCandidates);
+ this.jobContext = new RemoteExecutionContext<>(units, jobClassName, args);
+ }
+
+ /**
+ * Executes a job on the worker node and restarts the job on one of the failover candidates if the worker node leaves the cluster.
+ *
+ * @return JobExecution with the result of the job and the status of the job.
+ */
+ JobExecution<T> failSafeExecute() {
+ JobExecution<T> remoteJobExecution = launchJobOn(runningWorkerNode);
+ jobContext.initJobExecution(new FailSafeJobExecution<>(remoteJobExecution));
+
+ UUID handlerId = UUID.randomUUID();
+ nodeLeftEventsSource.addEventHandler(handlerId, new OnNodeLeft());
+ remoteJobExecution.resultAsync().whenComplete((r, e) -> nodeLeftEventsSource.removeEventHandler(handlerId));
+
+ return jobContext.failSafeJobExecution();
+ }
+
+ private JobExecution<T> launchJobOn(AtomicReference<ClusterNode> runningWorkerNode) {
+ return computeComponent.executeRemotely(runningWorkerNode.get(), jobContext.units(), jobContext.jobClassName(), jobContext.args());
+ }
+
+ class OnNodeLeft implements Consumer<ClusterNode> {
+ @Override
+ public void accept(ClusterNode leftNode) {
+ if (!runningWorkerNode.get().equals(leftNode)) {
Review Comment:
I wonder if it's feasible to update the candidates when a candidate leave the cluster?
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/FailSafeJobExecution.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.compute;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobStatus;
+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.lang.ErrorGroups.Compute;
+
+/**
+ * Fail-safe wrapper for the {@link JobExecution} that should be returned to the client. This wrapper holds the original
+ * job execution object this object can be updated during the lifetime of {@link FailSafeJobExecution}.
Review Comment:
```suggestion
* job execution object. This object can be updated during the lifetime of {@link FailSafeJobExecution}.
```
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeImpl.java:
##########
@@ -75,7 +98,10 @@ public <R> JobExecution<R> executeAsync(Set<ClusterNode> nodes, List<DeploymentU
throw new IllegalArgumentException("nodes must not be empty.");
}
- return new JobExecutionWrapper<>(executeOnOneNode(randomNode(nodes), units, jobClassName, args));
+ Set<ClusterNode> candidates = new HashSet<>(nodes);
+ ClusterNode targetNode = randomNode(candidates);
Review Comment:
Should we remove target node from the candidates set?
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeImpl.java:
##########
@@ -50,18 +51,40 @@ public class IgniteComputeImpl implements IgniteCompute {
private static final String DEFAULT_SCHEMA_NAME = "PUBLIC";
private final TopologyService topologyService;
+
private final IgniteTablesInternal tables;
+
private final ComputeComponent computeComponent;
private final ThreadLocalRandom random = ThreadLocalRandom.current();
+ private final NodeLeftEventsSource nodeLeftEventsSource;
+
/**
* Create new instance.
*/
public IgniteComputeImpl(TopologyService topologyService, IgniteTablesInternal tables, ComputeComponent computeComponent) {
this.topologyService = topologyService;
this.tables = tables;
this.computeComponent = computeComponent;
+ this.nodeLeftEventsSource = new NodeLeftEventsSource(topologyService);
+ }
+
+ private static ClusterNode leaderOfTablePartitionByTupleKey(TableViewInternal table, Tuple key) {
+ return requiredLeaderByPartition(table, table.partition(key));
+ }
+
+ private static <K> ClusterNode leaderOfTablePartitionByMappedKey(TableViewInternal table, K key, Mapper<K> keyMapper) {
+ return requiredLeaderByPartition(table, table.partition(key, keyMapper));
+ }
+
+ private static ClusterNode requiredLeaderByPartition(TableViewInternal table, int partitionIndex) {
+ ClusterNode leaderNode = table.leaderAssignment(partitionIndex);
+ if (leaderNode == null) {
+ throw new IgniteInternalException(Common.INTERNAL_ERR, "Leader not found for partition " + partitionIndex);
+ }
+
+ return leaderNode;
}
Review Comment:
Looks like these methods were moved accidentally.
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/FailSafeJobExecution.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.compute;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobStatus;
+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.lang.ErrorGroups.Compute;
+
+/**
+ * Fail-safe wrapper for the {@link JobExecution} that should be returned to the client. This wrapper holds the original
+ * job execution object this object can be updated during the lifetime of {@link FailSafeJobExecution}.
+ *
+ * <p>The problem that is solved by this wrapper is the following: client can join the {@link JobExecution#resultAsync()}
+ * future b1ut this original future will never be completed in case the remote worker node has left the topology. By returning
Review Comment:
```suggestion
* future but this original future will never be completed in case the remote worker node has left the topology. By returning
```
--
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.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-20850 Add compute job failover [ignite-3]
Posted by "PakhomovAlexander (via GitHub)" <gi...@apache.org>.
PakhomovAlexander commented on code in PR #2990:
URL: https://github.com/apache/ignite-3/pull/2990#discussion_r1447134000
##########
modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeJobFailover.java:
##########
@@ -0,0 +1,158 @@
+/*
+ * 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.compute;
+
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import org.apache.ignite.compute.DeploymentUnit;
+import org.apache.ignite.compute.JobExecution;
+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.lang.ErrorGroups.Compute;
+import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * This is a helper class for {@link ComputeComponent} to handle job failures. You can think about this class as a "retryable compute job
+ * with captured context". Retry logic is applied ONLY if the worker node leaves the cluster. If the job itself is failing, then the
+ * exception is propagated to the caller and this class does not handle it.
+ *
+ * <p>If you want to execute a job on node1 and use node2 and node3 as failover candidates,
+ * then you should create an instance of this class with workerNode = node1, failoverCandidates = [node2, node3] as arguments and
+ * call {@link #failSafeExecute()}.
+ *
+ * @param <T> the type of the result of the job.
+ */
+class ComputeJobFailover<T> {
+ private static final IgniteLogger LOG = Loggers.forClass(ComputeJobFailover.class);
+
+ /**
+ * Compute component that is called when the {@link #runningWorkerNode} has left the cluster.
+ */
+ private final ComputeComponent computeComponent;
+
+ /**
+ * The failover listens this event source to know when the {@link #runningWorkerNode} has left the cluster.
+ */
+ private final NodeLeftEventsSource nodeLeftEventsSource;
+
+ /**
+ * Set of worker candidates in case {@link #runningWorkerNode} has left the cluster.
+ */
+ private final ConcurrentLinkedDeque<ClusterNode> failoverCandidates;
+
+ /**
+ * The node where the job is being executed at a given moment. If node leaves the cluster, then the job is restarted on one of the
+ * {@link #failoverCandidates} and the reference is CASed to the new node.
+ */
+ private final AtomicReference<ClusterNode> runningWorkerNode;
+
+ /** Context of the called job. Captures deployment units, jobClassName and arguments. */
+ private final RemoteExecutionContext<T> jobContext;
+
+ /**
+ * Creates a per-job instance.
+ *
+ * @param computeComponent compute component.
+ * @param nodeLeftEventsSource node left events source, used as dynamic topology service (allows to remove handlers).
+ * @param workerNode the node to execute the job on.
+ * @param failoverCandidates the set of nodes where the job can be restarted if the worker node leaves the cluster.
+ * @param units deployment units.
+ * @param jobClassName the name of the job class.
+ * @param args the arguments of the job.
+ */
+ ComputeJobFailover(
+ ComputeComponent computeComponent,
+ NodeLeftEventsSource nodeLeftEventsSource,
+ ClusterNode workerNode,
+ Set<ClusterNode> failoverCandidates,
+ List<DeploymentUnit> units,
+ String jobClassName,
+ Object... args
+ ) {
+ this.computeComponent = computeComponent;
+ this.nodeLeftEventsSource = nodeLeftEventsSource;
+ this.runningWorkerNode = new AtomicReference<>(workerNode);
+ this.failoverCandidates = new ConcurrentLinkedDeque<>(failoverCandidates);
+ this.jobContext = new RemoteExecutionContext<>(units, jobClassName, args);
+ }
+
+ /**
+ * Executes a job on the worker node and restarts the job on one of the failover candidates if the worker node leaves the cluster.
+ *
+ * @return JobExecution with the result of the job and the status of the job.
+ */
+ JobExecution<T> failSafeExecute() {
+ JobExecution<T> remoteJobExecution = launchJobOn(runningWorkerNode);
+ jobContext.initJobExecution(new FailSafeJobExecution<>(remoteJobExecution));
+
+ UUID handlerId = UUID.randomUUID();
+ nodeLeftEventsSource.addEventHandler(handlerId, new OnNodeLeft());
+ remoteJobExecution.resultAsync().whenComplete((r, e) -> nodeLeftEventsSource.removeEventHandler(handlerId));
+
+ return jobContext.failSafeJobExecution();
+ }
+
+ private JobExecution<T> launchJobOn(AtomicReference<ClusterNode> runningWorkerNode) {
+ return computeComponent.executeRemotely(runningWorkerNode.get(), jobContext.units(), jobContext.jobClassName(), jobContext.args());
+ }
+
+ class OnNodeLeft implements Consumer<ClusterNode> {
+ @Override
+ public void accept(ClusterNode leftNode) {
+ if (!runningWorkerNode.get().equals(leftNode)) {
Review Comment:
Good point. I think we should always update the candidate list when the candidate comes back after leaving the topology. This feature should be tested carefully but now it would be difficult to write such a test. I'll update the ticket https://issues.apache.org/jira/browse/IGNITE-21219
--
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.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org