You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ap...@apache.org on 2024/01/24 12:42:31 UTC
(ignite-3) branch main updated: IGNITE-20864 Choose primary replica for colocated jobs (#3043)
This is an automated email from the ASF dual-hosted git repository.
apkhmv pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 38cbd5b542 IGNITE-20864 Choose primary replica for colocated jobs (#3043)
38cbd5b542 is described below
commit 38cbd5b5424bb427bc2d8b1a17366e792168c8de
Author: Aleksandr Pakhomov <ap...@gmail.com>
AuthorDate: Wed Jan 24 16:42:26 2024 +0400
IGNITE-20864 Choose primary replica for colocated jobs (#3043)
* Now the primary replica is chosen as a worker for
collocated execution.
* The failover is adjusted to support the selection of
the next worker node, see NextWorkerSelector.
* InteractiveJobs class is introduced. This is helpful
API for making assertions and sending commands to
interactive jobs.
* Before this change ComputeJobFailover was listening for
physical topology changes. But it is not what we really want.
Instead, we want to listen to a logical topology changes. So,
now NodeLeftEventsSource is not needed anymore because
LogicalTopologyEventListener is enough.
---
.../java/org/apache/ignite/lang/ErrorGroups.java | 11 +-
modules/compute/build.gradle | 3 +
.../internal/compute/ItWorkerShutdownTest.java | 438 ++++++---------------
.../internal/compute/utils/InteractiveJobs.java | 366 +++++++++++++++++
.../compute/utils/TestingJobExecution.java | 143 +++++++
.../internal/compute/ComputeJobFailover.java | 128 +++---
.../internal/compute/FailSafeJobExecution.java | 51 +--
.../ignite/internal/compute/IgniteComputeImpl.java | 138 +++++--
.../compute/NextColocatedWorkerSelector.java | 119 ++++++
.../internal/compute/NextWorkerSelector.java | 34 ++
.../internal/compute/NodeLeftEventsSource.java | 54 ---
.../internal/compute/IgniteComputeImplTest.java | 34 +-
.../placementdriver/PlacementDriverTest.java | 2 +-
modules/platforms/cpp/ignite/common/error_codes.h | 5 +-
modules/platforms/cpp/ignite/odbc/common_types.cpp | 3 +
.../platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs | 11 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 9 +-
17 files changed, 1051 insertions(+), 498 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
index 968125b428..c40c5e7a14 100755
--- a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
+++ b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
@@ -542,8 +542,17 @@ public class ErrorGroups {
/** Compute job failed. */
public static final int COMPUTE_JOB_FAILED_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 9);
+ /** Cannot change job priority, compute job not found error. */
+ public static final int CHANGE_JOB_PRIORITY_NO_JOB_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 10);
+
+ /** Cannot change job priority, compute job is already executing. */
+ public static final int CHANGE_JOB_PRIORITY_JOB_EXECUTING_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 11);
+
+ /** Cannot resolve primary replica for colocated execution. */
+ public static final int PRIMARY_REPLICA_RESOLVE_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 12);
+
/** Cannot change job priority. */
- public static final int CHANGE_JOB_PRIORITY_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 10);
+ public static final int CHANGE_JOB_PRIORITY_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 13);
}
/** Catalog error group. */
diff --git a/modules/compute/build.gradle b/modules/compute/build.gradle
index f2c14fec74..b7a26d3821 100644
--- a/modules/compute/build.gradle
+++ b/modules/compute/build.gradle
@@ -27,6 +27,8 @@ dependencies {
implementation project(':ignite-configuration-api')
implementation project(':ignite-core')
implementation project(':ignite-code-deployment')
+ implementation project(':ignite-cluster-management')
+ implementation project(':ignite-placement-driver-api')
implementation libs.jetbrains.annotations
implementation libs.fastutil.core
implementation libs.auto.service.annotations
@@ -51,6 +53,7 @@ dependencies {
integrationTestImplementation project(':ignite-network')
integrationTestImplementation project(':ignite-cluster-management')
integrationTestImplementation project(':ignite-table')
+ integrationTestImplementation project(':ignite-placement-driver-api')
integrationTestImplementation testFixtures(project(':ignite-core'))
integrationTestImplementation testFixtures(project(':ignite-runner'))
}
diff --git a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItWorkerShutdownTest.java b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItWorkerShutdownTest.java
index b1a2496393..c76b6c0870 100644
--- a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItWorkerShutdownTest.java
+++ b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItWorkerShutdownTest.java
@@ -17,15 +17,6 @@
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.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
-import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
-import static org.apache.ignite.internal.testframework.matchers.JobStatusMatcher.jobStatusWithState;
-import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
@@ -33,106 +24,51 @@ 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.ArrayList;
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.compute.utils.InteractiveJobs;
+import org.apache.ignite.internal.compute.utils.InteractiveJobs.AllInteractiveJobsApi;
+import org.apache.ignite.internal.compute.utils.TestingJobExecution;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.table.TableImpl;
-import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.table.Tuple;
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
+ * <p>The logic is that if we run the job on the remote node and this node has left the logical 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();
-
+public class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
/**
* 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<>();
+ private static final String TABLE_NAME = "test";
/**
- * 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.
+ * CMG == number of nodes in cluster. We wont lose the leader in tests then.
*/
- 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);
- }
+ @Override
+ protected int[] cmgMetastoreNodes() {
+ return new int[]{0, 1, 2};
}
private static Set<String> workerCandidates(IgniteImpl... nodes) {
@@ -142,37 +78,6 @@ class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
.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);
- }
-
private Set<ClusterNode> clusterNodesByNames(Set<String> nodes) {
return nodes.stream()
.map(NODES_NAMES_TO_INDEXES::get)
@@ -181,22 +86,14 @@ class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
.collect(Collectors.toSet());
}
- private static UUID idSync(JobExecution<?> execution) throws InterruptedException, ExecutionException, TimeoutException {
- return execution.idAsync().get(10, TimeUnit.SECONDS);
- }
-
/**
* 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();
- NODES_NAMES_TO_INDEXES.clear();
+ InteractiveJobs.clearState();
+ NODES_NAMES_TO_INDEXES.clear();
for (int i = 0; i < 3; i++) {
NODES_NAMES_TO_INDEXES.put(node(i).name(), i);
}
@@ -212,50 +109,49 @@ class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
Set<String> remoteWorkerCandidates = workerCandidates(node(1), node(2));
// When execute job.
- JobExecution<String> execution = executeGlobalInteractiveJob(entryNode, remoteWorkerCandidates);
+ TestingJobExecution<String> execution = executeGlobalInteractiveJob(entryNode, remoteWorkerCandidates);
// Then one of candidates became a worker and run the job.
- String workerNodeName = getWorkerNodeNameFromGlobalInteractiveJob();
+ String workerNodeName = InteractiveJobs.globalJob().currentWorkerName();
// And job is running.
- checkGlobalInteractiveJobAlive(execution);
+ InteractiveJobs.globalJob().assertAlive();
+ // And.
+ execution.assertExecuting();
// And save state BEFORE worker has failed.
- long createTimeBeforeFail = statusSync(execution).createTime().toEpochMilli();
- long startTimeBeforeFail = statusSync(execution).startTime().toEpochMilli();
- UUID jobIdBeforeFail = idSync(execution);
+ long createTimeBeforeFail = execution.createTimeMillis();
+ long startTimeBeforeFail = execution.startTimeMillis();
+ UUID jobIdBeforeFail = execution.idSync();
// When stop worker node.
- stopNodeByName(workerNodeName);
+ stopNode(workerNodeName);
// And remove it from candidates.
remoteWorkerCandidates.remove(workerNodeName);
// Then the job is alive: it has been restarted on another candidate.
- checkGlobalInteractiveJobAlive(execution);
+ InteractiveJobs.globalJob().assertAlive();
+ // And.
+ execution.assertExecuting();
// And remaining candidate was chosen as a failover worker.
- String failoverWorker = getWorkerNodeNameFromGlobalInteractiveJob();
+ String failoverWorker = InteractiveJobs.globalJob().currentWorkerName();
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));
+ assertThat(execution.createTimeMillis(), equalTo(createTimeBeforeFail));
+ assertThat(execution.startTimeMillis(), greaterThan(startTimeBeforeFail));
+ // And id was not changed.
+ assertThat(execution.idSync(), equalTo(jobIdBeforeFail));
// When finish job.
- finishGlobalJob();
+ InteractiveJobs.globalJob().finish();
// Then it is successfully finished.
- assertThat(execution.resultAsync().get(10, TimeUnit.SECONDS), equalTo("Done"));
- // And.
- assertThat(execution.statusAsync().get().state(), is(COMPLETED));
+ execution.assertCompleted();
// 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));
+ assertThat(execution.finishTimeMillis(), greaterThan(execution.createTimeMillis()));
+ assertThat(execution.finishTimeMillis(), greaterThan(execution.startTimeMillis()));
+ // And job id the same.
+ assertThat(execution.idSync(), equalTo(jobIdBeforeFail));
}
@Test
@@ -266,20 +162,20 @@ class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
Set<String> remoteWorkerCandidates = workerCandidates(node(1));
// When execute job.
- JobExecution<String> execution = executeGlobalInteractiveJob(entryNode, remoteWorkerCandidates);
+ TestingJobExecution<String> execution = executeGlobalInteractiveJob(entryNode, remoteWorkerCandidates);
// Then the job is running on worker node.
- String workerNodeName = getWorkerNodeNameFromGlobalInteractiveJob();
+ String workerNodeName = InteractiveJobs.globalJob().currentWorkerName();
assertThat(remoteWorkerCandidates, hasItem(workerNodeName));
// And.
- checkGlobalInteractiveJobAlive(execution);
+ InteractiveJobs.globalJob().assertAlive();
+ execution.assertExecuting();
// When stop worker node.
- stopNodeByName(workerNodeName);
+ stopNode(workerNodeName);
// Then the job is failed, because there is no any failover worker.
- assertThat(execution.resultAsync(), willThrow(IgniteException.class));
- assertThat(execution.statusAsync().isCompletedExceptionally(), equalTo(true));
+ execution.assertFailed();
}
@Test
@@ -288,20 +184,20 @@ class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
IgniteImpl entryNode = node(0);
// When execute job locally.
- JobExecution<String> execution = executeGlobalInteractiveJob(entryNode, Set.of(entryNode.name()));
+ TestingJobExecution<String> execution = executeGlobalInteractiveJob(entryNode, Set.of(entryNode.name()));
// Then the job is running.
- checkGlobalInteractiveJobAlive(execution);
+ InteractiveJobs.globalJob().assertAlive();
+ execution.assertExecuting();
// And it is running on entry node.
- assertThat(getWorkerNodeNameFromGlobalInteractiveJob(), equalTo(entryNode.name()));
+ assertThat(InteractiveJobs.globalJob().currentWorkerName(), equalTo(entryNode.name()));
// When stop entry node.
- stopNodeByName(entryNode.name());
+ stopNode(entryNode.name());
// Then the job is failed, because there is no any failover worker.
assertThat(execution.resultAsync().isCompletedExceptionally(), equalTo(true));
- assertThat(execution.statusAsync().get().state(), is(FAILED));
}
@Test
@@ -309,36 +205,41 @@ class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
// Given entry node.
IgniteImpl entryNode = node(0);
// And prepare communication channels.
- initChannels(allNodeNames());
+ InteractiveJobs.initChannels(allNodeNames());
// When start broadcast job.
Map<ClusterNode, JobExecution<Object>> executions = entryNode.compute().broadcastAsync(
clusterNodesByNames(workerCandidates(node(0), node(1), node(2))),
List.of(),
- InteractiveJob.class.getName()
+ InteractiveJobs.interactiveJobName()
);
// Then all three jobs are alive.
assertThat(executions.size(), is(3));
- executions.forEach(ItWorkerShutdownTest::checkInteractiveJobAlive);
+ executions.forEach((node, execution) -> {
+ InteractiveJobs.byNode(node).assertAlive();
+ new TestingJobExecution<>(execution).assertExecuting();
+ });
// When stop one of workers.
- stopNodeByName(node(1).name());
+ String stoppedNodeName = node(1).name();
+ stopNode(node(1));
// Then two jobs are alive.
executions.forEach((node, execution) -> {
- if (node.name().equals(node(1).name())) {
- assertThat(execution.resultAsync(), willThrow(IgniteException.class));
+ if (node.name().equals(stoppedNodeName)) {
+ new TestingJobExecution<>(execution).assertFailed();
} else {
- checkInteractiveJobAlive(node, execution);
+ InteractiveJobs.byNode(node).assertAlive();
+ new TestingJobExecution<>(execution).assertExecuting();
}
});
// When.
- finishAllInteractiveJobs();
+ InteractiveJobs.all().finish();
// Then every job ran once because broadcast execution does not require failover.
- checkAllInteractiveJobsCalledOnce();
+ AllInteractiveJobsApi.assertEachCalledOnce();
}
@Test
@@ -349,82 +250,98 @@ class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
Set<String> remoteWorkerCandidates = workerCandidates(node(1), node(2));
// When execute job.
- JobExecution<String> execution = executeGlobalInteractiveJob(entryNode, remoteWorkerCandidates);
+ TestingJobExecution<String> execution = executeGlobalInteractiveJob(entryNode, remoteWorkerCandidates);
// Then one of candidates became a worker and run the job.
- String workerNodeName = getWorkerNodeNameFromGlobalInteractiveJob();
+ String workerNodeName = InteractiveJobs.globalJob().currentWorkerName();
// And job is running.
- checkGlobalInteractiveJobAlive(execution);
+ InteractiveJobs.globalJob().assertAlive();
+ execution.assertExecuting();
// When stop worker node.
- stopNodeByName(workerNodeName);
+ stopNode(workerNodeName);
// And remove it from candidates.
remoteWorkerCandidates.remove(workerNodeName);
// Then the job is alive: it has been restarted on another candidate.
- checkGlobalInteractiveJobAlive(execution);
+ InteractiveJobs.globalJob().assertAlive();
+ execution.assertExecuting();
// And remaining candidate was chosen as a failover worker.
- String failoverWorker = getWorkerNodeNameFromGlobalInteractiveJob();
+ String failoverWorker = InteractiveJobs.globalJob().currentWorkerName();
assertThat(remoteWorkerCandidates, hasItem(failoverWorker));
// When cancel job.
- assertThat(execution.cancelAsync(), willBe(true));
+ execution.cancelSync();
// Then it is cancelled.
- assertThat(execution.resultAsync(), willThrow(IgniteException.class));
- // And.
- assertThat(execution.statusAsync(), willBe(jobStatusWithState(CANCELED)));
+ execution.assertCancelled();
}
@Test
- @Disabled("https://issues.apache.org/jira/browse/IGNITE-20864")
void colocatedExecutionWorkerShutdown() throws Exception {
- // Given table.
- initChannels(allNodeNames());
- createTestTableWithOneRow();
- TableImpl table = (TableImpl) node(0).tables().table("test");
+ // Given table with replicas == 3 and partitions == 1.
+ createReplicatedTestTableWithOneRow();
// And partition leader for K=1.
- ClusterNode leader = table.leaderAssignment(table.partition(Tuple.create(1).set("K", 1)));
+ ClusterNode primaryReplica = getPrimaryReplica(cluster.node(0));
- // When start colocated job on node that is not leader.
- IgniteImpl entryNode = anyNodeExcept(leader);
- JobExecution<Object> execution = entryNode.compute().executeColocatedAsync(
- "test",
+ // When start colocated job on node that is not primary replica.
+ IgniteImpl entryNode = anyNodeExcept(primaryReplica);
+ TestingJobExecution<Object> execution = new TestingJobExecution<>(entryNode.compute().executeColocatedAsync(
+ TABLE_NAME,
Tuple.create(1).set("K", 1),
List.of(),
- GlobalInteractiveJob.class.getName()
- );
+ InteractiveJobs.globalJob().name()
+ ));
// Then the job is alive.
- checkGlobalInteractiveJobAlive(execution);
+ InteractiveJobs.globalJob().assertAlive();
+ execution.assertExecuting();
- // And it is running on leader node.
- String firstWorkerNodeName = getWorkerNodeNameFromGlobalInteractiveJob();
- assertThat(firstWorkerNodeName, equalTo(leader.name()));
- // And leader node is NOT an entry node, it is remote.
- assertThat(entryNode.name(), not(equalTo(firstWorkerNodeName)));
+ // And it is running on primary replica node.
+ String firstWorkerNodeName = InteractiveJobs.globalJob().currentWorkerName();
+ assertThat(firstWorkerNodeName, equalTo(primaryReplica.name()));
// When stop worker node.
- stopNode(nodeByName(firstWorkerNodeName));
+ stopNode(primaryReplica);
// Then the job is restarted on another node.
- checkGlobalInteractiveJobAlive(execution);
+ InteractiveJobs.globalJob().assertAlive();
+ execution.assertExecuting();
// And it is running on another node.
- String failoverNodeName = getWorkerNodeNameFromGlobalInteractiveJob();
+ String failoverNodeName = InteractiveJobs.globalJob().currentWorkerName();
assertThat(failoverNodeName, in(allNodeNames()));
- // But.
+ // And this node is the primary replica for K=1.
+ primaryReplica = getPrimaryReplica(entryNode);
+ assertThat(failoverNodeName, equalTo(primaryReplica.name()));
+ // But this is not the same node as before.
assertThat(failoverNodeName, not(equalTo(firstWorkerNodeName)));
+ }
+
+ private ClusterNode getPrimaryReplica(IgniteImpl node) {
+ try {
+ var clock = node.clock();
+ TableImpl table = (TableImpl) node.tables().table(TABLE_NAME);
+ TablePartitionId tablePartitionId = new TablePartitionId(table.tableId(), table.partition(Tuple.create(1).set("K", 1)));
+
+ ReplicaMeta replicaMeta = node.placementDriver().getPrimaryReplica(tablePartitionId, clock.now()).get();
+ if (replicaMeta == null || replicaMeta.getLeaseholder() == null) {
+ throw new RuntimeException("Can not find primary replica for partition.");
+ }
+ return nodeByName(replicaMeta.getLeaseholder()).node();
+
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
}
- private void stopNode(IgniteImpl ignite) {
- stopNodeByName(ignite.name());
+ private void stopNode(ClusterNode clusterNode) {
+ stopNode(clusterNode.name());
}
- private void stopNodeByName(String name) {
- int ind = NODES_NAMES_TO_INDEXES.get(name);
- node(ind).stop();
+ private void stopNode(IgniteImpl ignite) {
+ stopNode(ignite.name());
}
private IgniteImpl anyNodeExcept(ClusterNode except) {
@@ -441,124 +358,21 @@ class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
return cluster.runningNodes().filter(node -> node.name().equals(candidateName)).findFirst().orElseThrow();
}
- private String getWorkerNodeNameFromGlobalInteractiveJob() throws InterruptedException {
- GLOBAL_SIGNALS.offer(Signal.GET_WORKER_NAME);
- return (String) GLOBAL_CHANNEL.poll(10, TimeUnit.SECONDS);
- }
-
- private JobExecution<String> executeGlobalInteractiveJob(IgniteImpl entryNode, Set<String> nodes) {
- return entryNode.compute().executeAsync(clusterNodesByNames(nodes), List.of(), GlobalInteractiveJob.class.getName());
+ private TestingJobExecution<String> executeGlobalInteractiveJob(IgniteImpl entryNode, Set<String> nodes) {
+ return new TestingJobExecution<>(
+ entryNode.compute().executeAsync(clusterNodesByNames(nodes), List.of(), InteractiveJobs.globalJob().name())
+ );
}
- private void createTestTableWithOneRow() {
- executeSql("CREATE TABLE test (k int, v int, CONSTRAINT PK PRIMARY KEY (k))");
+ private void createReplicatedTestTableWithOneRow() {
+ // Number of replicas == number of nodes and number of partitions == 1. This gives us the majority on primary replica stop.
+ // After the primary replica is stopped we still be able to select new primary replica selected.
+ executeSql("CREATE ZONE TEST_ZONE WITH REPLICAS=3, PARTITIONS=1");
+ executeSql("CREATE TABLE test (k int, v int, CONSTRAINT PK PRIMARY KEY (k)) WITH PRIMARY_ZONE='TEST_ZONE'");
executeSql("INSERT INTO test(k, v) VALUES (1, 101)");
}
private List<String> allNodeNames() {
- return IntStream.range(0, initialNodes())
- .mapToObj(this::node)
- .map(Ignite::name)
- .collect(toList());
- }
-
- /**
- * Signals that are sent by test code to the jobs.
- */
- enum Signal {
- /**
- * Signal to the job to continue running and send ACK as a response.
- */
- CONTINUE,
- /**
- * Ask job to throw an exception.
- */
- THROW,
- /**
- * Ask job to return result.
- */
- RETURN,
- /**
- * Signal to the job to continue running and send current worker name to the response channel.
- */
- GET_WORKER_NAME
- }
-
- /**
- * Interactive job that communicates via {@link #GLOBAL_CHANNEL} and {@link #GLOBAL_SIGNALS}.
- */
- static class GlobalInteractiveJob implements ComputeJob<String> {
- private static Signal listenSignal() {
- Signal recievedSignal;
- try {
- recievedSignal = GLOBAL_SIGNALS.take();
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- return recievedSignal;
- }
-
- @Override
- public String execute(JobExecutionContext context, Object... args) {
- while (true) {
- Signal recievedSignal = listenSignal();
- switch (recievedSignal) {
- case THROW:
- throw new RuntimeException();
- case CONTINUE:
- GLOBAL_CHANNEL.offer(ack);
- break;
- case RETURN:
- return "Done";
- case GET_WORKER_NAME:
- GLOBAL_CHANNEL.add(context.ignite().name());
- break;
- default:
- throw new IllegalStateException("Unexpected value: " + recievedSignal);
- }
- }
- }
- }
-
- /**
- * Interactive job that communicates via {@link #NODE_CHANNELS} and {@link #NODE_SIGNALS}. Also, keeps track of how many times it was
- * executed via {@link #INTERACTIVE_JOB_RUN_TIMES}.
- */
- static class InteractiveJob implements ComputeJob<String> {
- private static Signal listenSignal(BlockingQueue<Signal> channel) {
- Signal recievedSignal = null;
- try {
- recievedSignal = channel.take();
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- return recievedSignal;
- }
-
- @Override
- public String execute(JobExecutionContext context, Object... args) {
- String workerNodeName = context.ignite().name();
-
- INTERACTIVE_JOB_RUN_TIMES.put(workerNodeName, INTERACTIVE_JOB_RUN_TIMES.get(workerNodeName) + 1);
- BlockingQueue<Signal> channel = NODE_SIGNALS.get(workerNodeName);
-
- while (true) {
- Signal recievedSignal = listenSignal(channel);
- switch (recievedSignal) {
- case THROW:
- throw new RuntimeException();
- case CONTINUE:
- NODE_CHANNELS.get(workerNodeName).offer(ack);
- break;
- case RETURN:
- return "Done";
- case GET_WORKER_NAME:
- NODE_CHANNELS.get(workerNodeName).add(context.ignite().name());
- break;
- default:
- throw new IllegalStateException("Unexpected value: " + recievedSignal);
- }
- }
- }
+ return new ArrayList<>(NODES_NAMES_TO_INDEXES.keySet());
}
}
diff --git a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/utils/InteractiveJobs.java b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/utils/InteractiveJobs.java
new file mode 100644
index 0000000000..e55376b0c0
--- /dev/null
+++ b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/utils/InteractiveJobs.java
@@ -0,0 +1,366 @@
+/*
+ * 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.utils;
+
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.JobExecutionContext;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Tests DSL for interactive jobs. "Interactive" means that you can send messages and get responses to/from running jobs.
+ *
+ * <p>For example, you can start {@link GlobalInteractiveJob} on some node, get the name of worker node for this job,
+ * ask this job to complete successfully or throw exception. Also, this class gives useful assertions for job states.
+ *
+ * @see org.apache.ignite.internal.compute.ItWorkerShutdownTest
+ */
+public final class InteractiveJobs {
+ /**
+ * 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();
+
+ /**
+ * 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 final 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 final 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 final 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 final 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 final Map<String, Integer> INTERACTIVE_JOB_RUN_TIMES = new ConcurrentHashMap<>();
+
+ /**
+ * Counts for all running interactive jobs.
+ */
+ private static final AtomicInteger RUNNING_INTERACTIVE_JOBS_CNT = new AtomicInteger(0);
+
+ /**
+ * This counter indicated how many {@link GlobalInteractiveJob} instances running now. This counter increased each time the
+ * {@link GlobalInteractiveJob} is called and decreased when the job is done (whatever the result is). Checked in {@link #clearState}.
+ */
+ private static final AtomicInteger RUNNING_GLOBAL_JOBS_CNT = new AtomicInteger(0);
+
+ /**
+ * The timeout in seconds that defines how long should we wait for async calls. Almost all methods use this timeout.
+ */
+ private static final long WAIT_TIMEOUT_SECONDS = 15;
+
+ /**
+ * Clear global state. Must be called before each testing scenario.
+ */
+ public static void clearState() {
+ assertThat(
+ "Interactive job is running. Can not clear global state. Please, stop the job first.",
+ RUNNING_INTERACTIVE_JOBS_CNT.get(),
+ equalTo(0)
+ );
+ assertThat(
+ "Global job is running. Can not clear global state. Please, stop the job first.",
+ RUNNING_GLOBAL_JOBS_CNT.get(),
+ equalTo(0)
+ );
+
+ GLOBAL_SIGNALS.clear();
+ GLOBAL_CHANNEL.clear();
+ NODE_SIGNALS.clear();
+ NODE_CHANNELS.clear();
+ INTERACTIVE_JOB_RUN_TIMES.clear();
+ }
+
+ public static String interactiveJobName() {
+ return InteractiveJob.class.getName();
+ }
+
+ /**
+ * Signals that are sent by test code to the jobs.
+ */
+ private enum Signal {
+ /**
+ * Signal to the job to continue running and send ACK as a response.
+ */
+ CONTINUE,
+ /**
+ * Ask job to throw an exception.
+ */
+ THROW,
+ /**
+ * Ask job to return result.
+ */
+ RETURN,
+ /**
+ * Signal to the job to continue running and send current worker name to the response channel.
+ */
+ GET_WORKER_NAME
+ }
+
+ /**
+ * Interactive job that communicates via {@link #GLOBAL_CHANNEL} and {@link #GLOBAL_SIGNALS}.
+ */
+ private static class GlobalInteractiveJob implements ComputeJob<String> {
+ private static Signal listenSignal() {
+ try {
+ return GLOBAL_SIGNALS.take();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public String execute(JobExecutionContext context, Object... args) {
+ RUNNING_INTERACTIVE_JOBS_CNT.incrementAndGet();
+
+ try {
+ while (true) {
+ Signal receivedSignal = listenSignal();
+ switch (receivedSignal) {
+ case THROW:
+ throw new RuntimeException();
+ case CONTINUE:
+ GLOBAL_CHANNEL.offer(ACK);
+ break;
+ case RETURN:
+ return "Done";
+ case GET_WORKER_NAME:
+ GLOBAL_CHANNEL.add(context.ignite().name());
+ break;
+ default:
+ throw new IllegalStateException("Unexpected value: " + receivedSignal);
+ }
+ }
+ } finally {
+ RUNNING_INTERACTIVE_JOBS_CNT.decrementAndGet();
+ }
+ }
+ }
+
+ /**
+ * Interactive job that communicates via {@link #NODE_CHANNELS} and {@link #NODE_SIGNALS}. Also, keeps track of how many times it was
+ * executed via {@link #RUNNING_INTERACTIVE_JOBS_CNT}.
+ */
+ private static class InteractiveJob implements ComputeJob<String> {
+ private static Signal listenSignal(BlockingQueue<Signal> channel) {
+ try {
+ return channel.take();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public String execute(JobExecutionContext context, Object... args) {
+ RUNNING_INTERACTIVE_JOBS_CNT.incrementAndGet();
+
+ try {
+ String workerNodeName = context.ignite().name();
+
+ INTERACTIVE_JOB_RUN_TIMES.put(workerNodeName, INTERACTIVE_JOB_RUN_TIMES.get(workerNodeName) + 1);
+ BlockingQueue<Signal> channel = NODE_SIGNALS.get(workerNodeName);
+
+ while (true) {
+ Signal receivedSignal = listenSignal(channel);
+ switch (receivedSignal) {
+ case THROW:
+ throw new RuntimeException();
+ case CONTINUE:
+ NODE_CHANNELS.get(workerNodeName).offer(ACK);
+ break;
+ case RETURN:
+ return "Done";
+ case GET_WORKER_NAME:
+ NODE_CHANNELS.get(workerNodeName).add(context.ignite().name());
+ break;
+ default:
+ throw new IllegalStateException("Unexpected value: " + receivedSignal);
+ }
+ }
+ } finally {
+ RUNNING_INTERACTIVE_JOBS_CNT.decrementAndGet();
+ }
+ }
+ }
+
+ /**
+ * Initializes channels that will be used to communicate with {@link InteractiveJob}. Note: {@link GlobalInteractiveJob} does not
+ * require to call this method before communication but if you want to communicate with {@link InteractiveJob} then you must call this
+ * method first.
+ *
+ * @param nodes the list of cluster nodes.
+ */
+ public 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);
+ }
+ }
+
+ public static InteractiveJobApi byNode(ClusterNode clusterNode) {
+ return new InteractiveJobApi(clusterNode);
+ }
+
+ /**
+ * API for communication with {@link InteractiveJob}.
+ */
+ public static final class InteractiveJobApi {
+ private final ClusterNode node;
+
+ private InteractiveJobApi(ClusterNode node) {
+ this.node = node;
+ }
+
+ /**
+ * Checks that {@link InteractiveJob} is alive.
+ */
+ public void assertAlive() {
+ NODE_SIGNALS.get(node.name()).offer(Signal.CONTINUE);
+ try {
+ assertThat(NODE_CHANNELS.get(node.name()).poll(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS), equalTo(ACK));
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ /**
+ * API for the interaction with every {@link InteractiveJob} (may run on every cluster node, for example, broadcast).
+ */
+ public static AllInteractiveJobsApi all() {
+ return new AllInteractiveJobsApi();
+ }
+
+ /**
+ * API for the interaction with every {@link InteractiveJob} (may run on every cluster node, for example, broadcast).
+ */
+ public static final class AllInteractiveJobsApi {
+ private AllInteractiveJobsApi() {
+ }
+
+ /**
+ * Checks that each instance of {@link InteractiveJob} was called once.
+ */
+ public static void assertEachCalledOnce() {
+ INTERACTIVE_JOB_RUN_TIMES.forEach((nodeName, runTimes) -> assertThat(runTimes, equalTo(1)));
+ }
+
+ /**
+ * Finishes all {@link InteractiveJob}s.
+ */
+ public void finish() {
+ NODE_SIGNALS.forEach((nodeName, channel) -> {
+ try {
+ channel.offer(Signal.RETURN, WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException("Can not send finish signal to he node", e);
+ }
+ });
+
+ await().untilAsserted(() -> {
+ assertThat(
+ "Expect all jobs to be finished",
+ RUNNING_INTERACTIVE_JOBS_CNT.get(),
+ equalTo(0)
+ );
+ });
+ }
+ }
+
+ /**
+ * API for the interaction with {@link GlobalInteractiveJob}.
+ */
+ public static GlobalApi globalJob() {
+ return new GlobalApi();
+ }
+
+ /**
+ * API for the interaction with {@link GlobalInteractiveJob}.
+ */
+ public static final class GlobalApi {
+
+ private GlobalApi() {
+ }
+
+ /**
+ * Returns the name of the worker node where {@link GlobalInteractiveJob} is running.
+ */
+ public String currentWorkerName() throws InterruptedException {
+ GLOBAL_SIGNALS.offer(Signal.GET_WORKER_NAME);
+ String workerName = (String) GLOBAL_CHANNEL.poll(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ assertThat(
+ "Can not get worker name for global job.", workerName, notNullValue()
+ );
+
+ return workerName;
+ }
+
+ /**
+ * Checks that {@link GlobalInteractiveJob} is alive.
+ */
+ public void assertAlive() throws InterruptedException {
+ GLOBAL_SIGNALS.offer(Signal.CONTINUE);
+ assertThat(GLOBAL_CHANNEL.poll(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS), equalTo(ACK));
+ }
+
+ /**
+ * Finishes {@link GlobalInteractiveJob}.
+ */
+ public void finish() {
+ GLOBAL_SIGNALS.offer(Signal.RETURN);
+ }
+
+ /**
+ * Returns the class name of {@link GlobalInteractiveJob}.
+ */
+ public String name() {
+ return GlobalInteractiveJob.class.getName();
+ }
+ }
+}
diff --git a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/utils/TestingJobExecution.java b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/utils/TestingJobExecution.java
new file mode 100644
index 0000000000..e8ecabe6ad
--- /dev/null
+++ b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/utils/TestingJobExecution.java
@@ -0,0 +1,143 @@
+/*
+ * 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.utils;
+
+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.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStatusMatcher.jobStatusWithState;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobStatus;
+import org.apache.ignite.lang.IgniteException;
+
+/**
+ * Testing instance of {@link JobExecution}. Adds useful assertions on job's state and sync methods.
+ *
+ * @param <R> Job result type.
+ */
+public class TestingJobExecution<R> implements JobExecution<R> {
+ private final JobExecution<R> jobExecution;
+
+ /**
+ * Constructor.
+ *
+ * @param jobExecution job execution to wrap.
+ */
+ public TestingJobExecution(JobExecution<R> jobExecution) {
+ this.jobExecution = jobExecution;
+ }
+
+ private JobStatus statusSync() throws InterruptedException, ExecutionException, TimeoutException {
+ return jobExecution.statusAsync().get(10, TimeUnit.SECONDS);
+ }
+
+ public UUID idSync() throws InterruptedException, ExecutionException, TimeoutException {
+ return jobExecution.idAsync().get(10, TimeUnit.SECONDS);
+ }
+
+ private R resultSync() throws ExecutionException, InterruptedException, TimeoutException {
+ return jobExecution.resultAsync().get(10, TimeUnit.SECONDS);
+ }
+
+ public long createTimeMillis() throws ExecutionException, InterruptedException, TimeoutException {
+ return statusSync().createTime().toEpochMilli();
+ }
+
+ public long startTimeMillis() throws ExecutionException, InterruptedException, TimeoutException {
+ return statusSync().startTime().toEpochMilli();
+ }
+
+ public long finishTimeMillis() throws ExecutionException, InterruptedException, TimeoutException {
+ return statusSync().finishTime().toEpochMilli();
+ }
+
+ public void cancelSync() throws ExecutionException, InterruptedException, TimeoutException {
+ jobExecution.cancelAsync().get(10, TimeUnit.SECONDS);
+ }
+
+ @Override
+ public CompletableFuture<R> resultAsync() {
+ return jobExecution.resultAsync();
+ }
+
+ @Override
+ public CompletableFuture<JobStatus> statusAsync() {
+ return jobExecution.statusAsync();
+ }
+
+ @Override
+ public CompletableFuture<Boolean> cancelAsync() {
+ return jobExecution.cancelAsync();
+ }
+
+ @Override
+ public CompletableFuture<Boolean> changePriorityAsync(int newPriority) {
+ return jobExecution.changePriorityAsync(newPriority);
+ }
+
+ /**
+ * Checks that the job execution object has EXECUTING state.
+ */
+ public void assertExecuting() {
+ await().until(() -> jobExecution.statusAsync().get(10, TimeUnit.SECONDS).state() == EXECUTING);
+
+ assertThat(jobExecution.resultAsync().isDone(), equalTo(false));
+
+ assertThat(idAsync(), willBe(notNullValue()));
+ }
+
+ /**
+ * Checks that the job execution object is cancelled.
+ */
+ public void assertCancelled() {
+ assertThat(resultAsync(), willThrow(IgniteException.class));
+ assertThat(statusAsync(), willBe(jobStatusWithState(CANCELED)));
+ }
+
+ /**
+ * Checks that the job execution object is completed successfully.
+ */
+ public void assertCompleted() {
+ assertThat(resultAsync(), willBe("Done"));
+ assertThat(statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ }
+
+ /**
+ * Checks that the job execution has failed.
+ */
+ public void assertFailed() {
+ await().untilAsserted(() -> {
+ assertThat(jobExecution.resultAsync(), willThrow(IgniteException.class));
+ });
+ await().untilAsserted(() -> {
+ assertThat(jobExecution.statusAsync(), willThrow(IgniteException.class));
+ });
+ }
+}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeJobFailover.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeJobFailover.java
index f6cbea769c..9318ca3e06 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeJobFailover.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeJobFailover.java
@@ -18,19 +18,20 @@
package org.apache.ignite.internal.compute;
import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.Executor;
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.cluster.management.topology.api.LogicalNode;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyEventListener;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
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;
+import org.apache.ignite.network.TopologyService;
/**
* This is a helper class for {@link ComputeComponent} to handle job failures. You can think about this class as a "retryable compute job
@@ -38,32 +39,43 @@ import org.jetbrains.annotations.Nullable;
* 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()}.
+ * 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);
+ /**
+ * Thread to run failover logic. We can not perform time-consuming operations in the same thread where we discover topology changes (it
+ * is network id thread).
+ */
+ private final Executor executor;
+
/**
* 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.
+ * Topology service is needed to listen logical topology events. This is how we get know that the node has left the cluster.
*/
- private final NodeLeftEventsSource nodeLeftEventsSource;
+ private final LogicalTopologyService logicalTopologyService;
/**
- * Set of worker candidates in case {@link #runningWorkerNode} has left the cluster.
+ * Physical topology service that helps to figure out if we want to restart job on the local node or on the remote one.
*/
- private final ConcurrentLinkedDeque<ClusterNode> failoverCandidates;
+ private final TopologyService topologyService;
+
+ /**
+ * The selector that returns the next worker node to execute job on.
+ */
+ private final NextWorkerSelector nextWorkerSelector;
/**
* 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.
+ * worker node returned by {@link #nextWorkerSelector} and the reference is CASed to the new node.
*/
private final AtomicReference<ClusterNode> runningWorkerNode;
@@ -76,27 +88,32 @@ class ComputeJobFailover<T> {
* 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 logicalTopologyService logical topology service.
+ * @param topologyService physical topology service.
* @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 executor the thread pool where the failover should run on.
* @param units deployment units.
* @param jobClassName the name of the job class.
* @param args the arguments of the job.
*/
ComputeJobFailover(
ComputeComponent computeComponent,
- NodeLeftEventsSource nodeLeftEventsSource,
+ LogicalTopologyService logicalTopologyService,
+ TopologyService topologyService,
ClusterNode workerNode,
- Set<ClusterNode> failoverCandidates,
+ NextWorkerSelector nextWorkerSelector,
+ Executor executor,
List<DeploymentUnit> units,
String jobClassName,
Object... args
) {
this.computeComponent = computeComponent;
- this.nodeLeftEventsSource = nodeLeftEventsSource;
this.runningWorkerNode = new AtomicReference<>(workerNode);
- this.failoverCandidates = new ConcurrentLinkedDeque<>(failoverCandidates);
+ this.logicalTopologyService = logicalTopologyService;
+ this.topologyService = topologyService;
+ this.nextWorkerSelector = nextWorkerSelector;
this.jobContext = new RemoteExecutionContext<>(units, jobClassName, args);
+ this.executor = executor;
}
/**
@@ -105,55 +122,54 @@ class ComputeJobFailover<T> {
* @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));
+ JobExecution<T> jobExecution = launchJobOn(runningWorkerNode.get());
+ jobContext.initJobExecution(new FailSafeJobExecution<>(jobExecution));
- Consumer<ClusterNode> handler = new OnNodeLeft();
- nodeLeftEventsSource.addEventHandler(handler);
- remoteJobExecution.resultAsync().whenComplete((r, e) -> nodeLeftEventsSource.removeEventHandler(handler));
+ LogicalTopologyEventListener nodeLeftEventListener = new OnNodeLeft();
+ logicalTopologyService.addEventListener(nodeLeftEventListener);
+ jobExecution.resultAsync().whenComplete((r, e) -> logicalTopologyService.removeEventListener(nodeLeftEventListener));
return jobContext.failSafeJobExecution();
}
- private JobExecution<T> launchJobOn(AtomicReference<ClusterNode> runningWorkerNode) {
- return computeComponent.executeRemotely(runningWorkerNode.get(), jobContext.units(), jobContext.jobClassName(), jobContext.args());
+ private JobExecution<T> launchJobOn(ClusterNode runningWorkerNode) {
+ if (runningWorkerNode.equals(topologyService.localMember())) {
+ return computeComponent.executeLocally(jobContext.units(), jobContext.jobClassName(), jobContext.args());
+ } else {
+ return computeComponent.executeRemotely(
+ runningWorkerNode, jobContext.units(), jobContext.jobClassName(), jobContext.args()
+ );
+ }
}
- class OnNodeLeft implements Consumer<ClusterNode> {
+ class OnNodeLeft implements LogicalTopologyEventListener {
@Override
- public void accept(ClusterNode leftNode) {
- if (!runningWorkerNode.get().equals(leftNode)) {
+ public void onNodeLeft(LogicalNode leftNode, LogicalTopologySnapshot newTopology) {
+ if (!runningWorkerNode.get().id().equals(leftNode.id())) {
return;
}
- ClusterNode nextWorkerCandidate = takeCandidate();
- if (nextWorkerCandidate == null) {
- LOG.warn("No more worker nodes to restart the job. Failing the job {}.", jobContext.jobClassName());
-
- FailSafeJobExecution<?> failSafeJobExecution = jobContext.failSafeJobExecution();
- failSafeJobExecution.completeExceptionally(
- new IgniteInternalException(Compute.COMPUTE_JOB_FAILED_ERR)
- );
- return;
- }
-
- LOG.warn(
- "Worker node {} has left the cluster. Restarting the job {} on node {}.",
- leftNode, jobContext.jobClassName(), nextWorkerCandidate
- );
-
- runningWorkerNode.set(nextWorkerCandidate);
- JobExecution<T> jobExecution = launchJobOn(runningWorkerNode);
- jobContext.updateJobExecution(jobExecution);
- }
-
- @Nullable
- private ClusterNode takeCandidate() {
- try {
- return failoverCandidates.pop();
- } catch (NoSuchElementException ex) {
- return null;
- }
+ executor.execute(() -> nextWorkerSelector.next()
+ .thenAccept(nextWorker -> {
+ if (nextWorker == null) {
+ LOG.warn("No more worker nodes to restart the job. Failing the job {}.", jobContext.jobClassName());
+
+ FailSafeJobExecution<?> failSafeJobExecution = jobContext.failSafeJobExecution();
+ failSafeJobExecution.completeExceptionally(
+ new IgniteInternalException(Compute.COMPUTE_JOB_FAILED_ERR)
+ );
+ return;
+ }
+
+ LOG.warn(
+ "Worker node {} has left the cluster. Restarting the job {} on node {}.",
+ leftNode, jobContext.jobClassName(), nextWorker
+ );
+
+ runningWorkerNode.set(nextWorker);
+ JobExecution<T> jobExecution = launchJobOn(runningWorkerNode.get());
+ jobContext.updateJobExecution(jobExecution);
+ }));
}
}
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/FailSafeJobExecution.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/FailSafeJobExecution.java
index 46e6e31b1c..e2e0a2fe24 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/FailSafeJobExecution.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/FailSafeJobExecution.java
@@ -20,9 +20,7 @@ package org.apache.ignite.internal.compute;
import java.time.Instant;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.JobState;
@@ -32,13 +30,13 @@ import org.apache.ignite.internal.logger.Loggers;
import org.jetbrains.annotations.Nullable;
/**
- * 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}.
+ * 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 but this original future will never be completed in case the remote worker node has left the topology. By returning
- * {@link FailSafeJobExecution} to the client we can update the original job execution object when it is restarted on another
- * node but the client will still be able to join the original future.
+ * {@link FailSafeJobExecution} to the client we can update the original job execution object when it is restarted on another node but the
+ * client will still be able to join the original future.
*
* @param <T> the type of the job result.
*/
@@ -76,21 +74,15 @@ class FailSafeJobExecution<T> implements JobExecution<T> {
}
private void captureStatus(JobExecution<T> runningJobExecution) {
- try {
- runningJobExecution.statusAsync().whenComplete((status, e) -> {
- if (status != null) {
- this.capturedStatus.set(status);
- } else {
- this.capturedStatus.set(
- failedStatus()
- );
- }
- }).get(10, TimeUnit.SECONDS);
- } catch (TimeoutException | InterruptedException | ExecutionException e) {
- this.capturedStatus.set(
- failedStatus()
- );
- }
+ runningJobExecution.statusAsync()
+ .completeOnTimeout(failedStatus(), 10, TimeUnit.SECONDS)
+ .whenComplete((status, e) -> {
+ if (status != null) {
+ this.capturedStatus.compareAndSet(null, status);
+ } else {
+ this.capturedStatus.compareAndSet(null, failedStatus());
+ }
+ });
}
private static JobStatus failedStatus() {
@@ -118,14 +110,23 @@ class FailSafeJobExecution<T> implements JobExecution<T> {
}
/**
- * Transforms the status by modifying the fields that should be always the same regardless of the job execution attempt.
- * For example, the job creation time should be the same for all attempts.
+ * Transforms the status by modifying the fields that should be always the same regardless of the job execution attempt. For example,
+ * the job creation time should be the same for all attempts.
*
- * @param jobStatus current job status.
+ * <p>Can update {@link #capturedStatus} as a side-effect if the one is null.
*
+ * @param jobStatus current job status.
* @return transformed job status.
*/
- private JobStatus transformStatus(JobStatus jobStatus) {
+ private @Nullable JobStatus transformStatus(@Nullable JobStatus jobStatus) {
+ if (jobStatus == null) {
+ return null;
+ }
+
+ if (capturedStatus.get() == null) {
+ capturedStatus.compareAndSet(null, jobStatus);
+ }
+
return jobStatus.toBuilder()
.createTime(capturedStatus.get().createTime())
.id(capturedStatus.get().id())
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeImpl.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeImpl.java
index 19a70a5c2b..f8c42694c7 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeImpl.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeImpl.java
@@ -17,28 +17,43 @@
package org.apache.ignite.internal.compute;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.function.Function.identity;
import static java.util.stream.Collectors.toUnmodifiableMap;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.compute.ComputeException;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.IgniteCompute;
import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.JobStatus;
-import org.apache.ignite.internal.lang.IgniteInternalException;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.table.IgniteTablesInternal;
import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.thread.NamedThreadFactory;
+import org.apache.ignite.internal.util.CompletableFutures;
import org.apache.ignite.internal.util.ExceptionUtils;
-import org.apache.ignite.lang.ErrorGroups.Common;
+import org.apache.ignite.lang.ErrorGroups.Compute;
import org.apache.ignite.lang.TableNotFoundException;
import org.apache.ignite.lang.util.IgniteNameUtils;
import org.apache.ignite.network.ClusterNode;
@@ -51,6 +66,8 @@ import org.jetbrains.annotations.Nullable;
* Implementation of {@link IgniteCompute}.
*/
public class IgniteComputeImpl implements IgniteComputeInternal {
+ private static final IgniteLogger LOG = Loggers.forClass(IgniteComputeImpl.class);
+
private static final String DEFAULT_SCHEMA_NAME = "PUBLIC";
private final TopologyService topologyService;
@@ -61,16 +78,30 @@ public class IgniteComputeImpl implements IgniteComputeInternal {
private final ThreadLocalRandom random = ThreadLocalRandom.current();
- private final NodeLeftEventsSource nodeLeftEventsSource;
+ private final LogicalTopologyService logicalTopologyService;
+
+ private final PlacementDriver placementDriver;
+
+ private final HybridClock clock;
+
+ private final Executor failoverExecutor;
/**
* Create new instance.
*/
- public IgniteComputeImpl(TopologyService topologyService, IgniteTablesInternal tables, ComputeComponent computeComponent) {
+ public IgniteComputeImpl(PlacementDriver placementDriver, TopologyService topologyService,
+ LogicalTopologyService logicalTopologyService, IgniteTablesInternal tables, ComputeComponent computeComponent,
+ HybridClock clock) {
+ this.placementDriver = placementDriver;
this.topologyService = topologyService;
this.tables = tables;
this.computeComponent = computeComponent;
- this.nodeLeftEventsSource = new NodeLeftEventsSource(topologyService);
+ this.logicalTopologyService = logicalTopologyService;
+ this.clock = clock;
+ this.failoverExecutor = Executors.newFixedThreadPool(
+ 1,
+ new NamedThreadFactory("compute-job-failover", LOG)
+ );
}
/** {@inheritDoc} */
@@ -87,8 +118,16 @@ public class IgniteComputeImpl implements IgniteComputeInternal {
Set<ClusterNode> candidates = new HashSet<>(nodes);
ClusterNode targetNode = randomNode(candidates);
candidates.remove(targetNode);
-
- return new JobExecutionWrapper<>(executeOnOneNodeWithFailover(targetNode, candidates, units, jobClassName, args));
+ NextWorkerSelector selector = new DeqNextWorkerSelector(new ConcurrentLinkedDeque<>(candidates));
+
+ return new JobExecutionWrapper<>(
+ executeOnOneNodeWithFailover(
+ targetNode,
+ selector,
+ units,
+ jobClassName,
+ args
+ ));
}
/** {@inheritDoc} */
@@ -119,7 +158,7 @@ public class IgniteComputeImpl implements IgniteComputeInternal {
private <R> JobExecution<R> executeOnOneNodeWithFailover(
ClusterNode targetNode,
- Set<ClusterNode> failoverCandidates,
+ NextWorkerSelector nextWorkerSelector,
List<DeploymentUnit> units,
String jobClassName,
Object[] args
@@ -128,23 +167,27 @@ public class IgniteComputeImpl implements IgniteComputeInternal {
return computeComponent.executeLocally(units, jobClassName, args);
} else {
return new ComputeJobFailover<R>(
- computeComponent, nodeLeftEventsSource,
- targetNode, failoverCandidates, units,
+ computeComponent, logicalTopologyService, topologyService,
+ targetNode, nextWorkerSelector, failoverExecutor, units,
jobClassName, args
).failSafeExecute();
}
}
- private <R> JobExecution<R> executeOnOneNode(
- ClusterNode targetNode,
- List<DeploymentUnit> units,
- String jobClassName,
- Object[] args
- ) {
- if (isLocal(targetNode)) {
- return computeComponent.executeLocally(units, jobClassName, args);
- } else {
- return computeComponent.executeRemotely(targetNode, units, jobClassName, args);
+ private static class DeqNextWorkerSelector implements NextWorkerSelector {
+ private final ConcurrentLinkedDeque<ClusterNode> deque;
+
+ private DeqNextWorkerSelector(ConcurrentLinkedDeque<ClusterNode> deque) {
+ this.deque = deque;
+ }
+
+ @Override
+ public CompletableFuture<ClusterNode> next() {
+ try {
+ return completedFuture(deque.pop());
+ } catch (NoSuchElementException ex) {
+ return nullCompletedFuture();
+ }
}
}
@@ -156,20 +199,24 @@ public class IgniteComputeImpl implements IgniteComputeInternal {
@Override
public <R> JobExecution<R> executeColocatedAsync(
String tableName,
- Tuple key,
+ Tuple tuple,
List<DeploymentUnit> units,
String jobClassName,
Object... args
) {
Objects.requireNonNull(tableName);
- Objects.requireNonNull(key);
+ Objects.requireNonNull(tuple);
Objects.requireNonNull(units);
Objects.requireNonNull(jobClassName);
return new JobExecutionFutureWrapper<>(
requiredTable(tableName)
- .thenApply(table -> leaderOfTablePartitionByTupleKey(table, key))
- .thenApply(primaryNode -> executeOnOneNode(primaryNode, units, jobClassName, args))
+ .thenCompose(table -> primaryReplicaForPartitionByTupleKey(table, tuple)
+ .thenApply(primaryNode -> executeOnOneNodeWithFailover(
+ primaryNode,
+ new NextColocatedWorkerSelector<>(placementDriver, topologyService, clock, table, tuple),
+ units, jobClassName, args
+ )))
);
}
@@ -189,9 +236,15 @@ public class IgniteComputeImpl implements IgniteComputeInternal {
Objects.requireNonNull(units);
Objects.requireNonNull(jobClassName);
- return new JobExecutionFutureWrapper<>(requiredTable(tableName)
- .thenApply(table -> leaderOfTablePartitionByMappedKey(table, key, keyMapper))
- .thenApply(primaryNode -> executeOnOneNode(primaryNode, units, jobClassName, args)));
+ return new JobExecutionFutureWrapper<>(
+ requiredTable(tableName)
+ .thenCompose(table -> primaryReplicaForPartitionByMappedKey(table, key, keyMapper)
+ .thenApply(primaryNode -> executeOnOneNodeWithFailover(
+ primaryNode,
+ new NextColocatedWorkerSelector<>(placementDriver, topologyService, clock, table, key, keyMapper),
+ units, jobClassName, args
+ )))
+ );
}
/** {@inheritDoc} */
@@ -239,21 +292,29 @@ public class IgniteComputeImpl implements IgniteComputeInternal {
});
}
- private static ClusterNode leaderOfTablePartitionByTupleKey(TableViewInternal table, Tuple key) {
- return requiredLeaderByPartition(table, table.partition(key));
+ private CompletableFuture<ClusterNode> primaryReplicaForPartitionByTupleKey(TableViewInternal table, Tuple key) {
+ return primaryReplicaForPartition(table, table.partition(key));
}
- private static <K> ClusterNode leaderOfTablePartitionByMappedKey(TableViewInternal table, K key, Mapper<K> keyMapper) {
- return requiredLeaderByPartition(table, table.partition(key, keyMapper));
+ private <K> CompletableFuture<ClusterNode> primaryReplicaForPartitionByMappedKey(TableViewInternal table, K key,
+ Mapper<K> keyMapper) {
+ return primaryReplicaForPartition(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);
- }
+ private CompletableFuture<ClusterNode> primaryReplicaForPartition(TableViewInternal table, int partitionIndex) {
+ TablePartitionId tablePartitionId = new TablePartitionId(table.tableId(), partitionIndex);
- return leaderNode;
+ return placementDriver.awaitPrimaryReplica(tablePartitionId, clock.now(), 30, TimeUnit.SECONDS)
+ .thenApply(replicaMeta -> {
+ if (replicaMeta != null && replicaMeta.getLeaseholderId() != null) {
+ return topologyService.getById(replicaMeta.getLeaseholderId());
+ }
+
+ throw new ComputeException(
+ Compute.PRIMARY_REPLICA_RESOLVE_ERR,
+ "Can not find primary replica for [table=" + table.name() + ", partition=" + partitionIndex + "]."
+ );
+ });
}
/** {@inheritDoc} */
@@ -272,7 +333,8 @@ public class IgniteComputeImpl implements IgniteComputeInternal {
.collect(toUnmodifiableMap(identity(),
// No failover nodes for broadcast. We use failover here in order to complete futures with exceptions
// if worker node has left the cluster.
- node -> new JobExecutionWrapper<>(executeOnOneNodeWithFailover(node, Set.of(), units, jobClassName, args))));
+ node -> new JobExecutionWrapper<>(executeOnOneNodeWithFailover(node,
+ CompletableFutures::nullCompletedFuture, units, jobClassName, args))));
}
@Override
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/NextColocatedWorkerSelector.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/NextColocatedWorkerSelector.java
new file mode 100644
index 0000000000..a9055ad5f5
--- /dev/null
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/NextColocatedWorkerSelector.java
@@ -0,0 +1,119 @@
+/*
+ * 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.TimeUnit;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.TopologyService;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.table.mapper.Mapper;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Next worker selector that returns primary replica node for next worker. If there is no such node (we lost the majority, for example) the
+ * {@code CompletableFuture.completedFuture(null)} will be returned.
+ *
+ * @param <K> type of the key for the colocated table.
+ */
+public class NextColocatedWorkerSelector<K> implements NextWorkerSelector {
+ private static final int PRIMARY_REPLICA_ASK_CLOCK_ADDITION_MILLIS = 10_000;
+
+ private static final int AWAIT_FOR_PRIMARY_REPLICA_SECONDS = 15;
+
+ private final PlacementDriver placementDriver;
+
+ private final TopologyService topologyService;
+
+ private final HybridClock clock;
+
+ @Nullable
+ private final K key;
+
+ @Nullable
+ private final Mapper<K> keyMapper;
+
+ private final Tuple tuple;
+
+ private final TableViewInternal table;
+
+ NextColocatedWorkerSelector(
+ PlacementDriver placementDriver,
+ TopologyService topologyService,
+ HybridClock clock,
+ TableViewInternal table,
+ K key,
+ Mapper<K> keyMapper) {
+ this(placementDriver, topologyService, clock, table, key, keyMapper, null);
+ }
+
+ NextColocatedWorkerSelector(
+ PlacementDriver placementDriver,
+ TopologyService topologyService,
+ HybridClock clock,
+ TableViewInternal table,
+ Tuple tuple) {
+ this(placementDriver, topologyService, clock, table, null, null, tuple);
+ }
+
+ private NextColocatedWorkerSelector(
+ PlacementDriver placementDriver,
+ TopologyService topologyService,
+ HybridClock clock,
+ TableViewInternal table,
+ @Nullable K key,
+ @Nullable Mapper<K> keyMapper,
+ @Nullable Tuple tuple) {
+ this.placementDriver = placementDriver;
+ this.topologyService = topologyService;
+ this.table = table;
+ this.clock = clock;
+ this.key = key;
+ this.keyMapper = keyMapper;
+ this.tuple = tuple;
+ }
+
+ private CompletableFuture<ClusterNode> tryToFindPrimaryReplica(TablePartitionId tablePartitionId) {
+ return placementDriver.awaitPrimaryReplica(
+ tablePartitionId,
+ clock.now().addPhysicalTime(PRIMARY_REPLICA_ASK_CLOCK_ADDITION_MILLIS),
+ AWAIT_FOR_PRIMARY_REPLICA_SECONDS,
+ TimeUnit.SECONDS
+ ).thenApply(ReplicaMeta::getLeaseholderId)
+ .thenApply(topologyService::getById);
+ }
+
+ @Override
+ public CompletableFuture<ClusterNode> next() {
+ TablePartitionId tablePartitionId = tablePartitionId();
+ return tryToFindPrimaryReplica(tablePartitionId);
+ }
+
+ private TablePartitionId tablePartitionId() {
+ if (key != null && keyMapper != null) {
+ return new TablePartitionId(table.tableId(), table.partition(key, keyMapper));
+ } else {
+ return new TablePartitionId(table.tableId(), table.partition(tuple));
+ }
+ }
+}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/NextWorkerSelector.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/NextWorkerSelector.java
new file mode 100644
index 0000000000..4b061959c9
--- /dev/null
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/NextWorkerSelector.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.compute;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * The selector that returns the next worker node to execute job on.
+ */
+public interface NextWorkerSelector {
+ /**
+ * Requests the next worker node for the job.
+ *
+ * @return {@code CompletableFuture} with the next worker node or with {@code null} if there is no candidates left.
+ */
+ CompletableFuture<@Nullable ClusterNode> next();
+}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/NodeLeftEventsSource.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/NodeLeftEventsSource.java
deleted file mode 100644
index 9a3d9e680c..0000000000
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/NodeLeftEventsSource.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.concurrent.CopyOnWriteArrayList;
-import java.util.function.Consumer;
-import org.apache.ignite.network.ClusterNode;
-import org.apache.ignite.network.TopologyEventHandler;
-import org.apache.ignite.network.TopologyService;
-
-/**
- * This is a workaround for implementing add and remove event handlers for node left events.
- * todo: remove and use TopologyService after <a href="https://issues.apache.org/jira/browse/IGNITE-14519">IGNITE-14519</a>.
- */
-class NodeLeftEventsSource {
- private final List<Consumer<ClusterNode>> handlers;
-
- NodeLeftEventsSource(TopologyService delegate) {
- this.handlers = new CopyOnWriteArrayList<>();
-
- delegate.addEventHandler(new NodeLeftTopologyEventHandler());
- }
-
- void addEventHandler(Consumer<ClusterNode> onNodeLeftHandler) {
- handlers.add(onNodeLeftHandler);
- }
-
- void removeEventHandler(Consumer<ClusterNode> onNodeLeftHandler) {
- handlers.remove(onNodeLeftHandler);
- }
-
- private class NodeLeftTopologyEventHandler implements TopologyEventHandler {
- @Override
- public void onDisappeared(ClusterNode member) {
- handlers.forEach(handler -> handler.accept(member));
- }
- }
-}
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/IgniteComputeImplTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/IgniteComputeImplTest.java
index ffc4f3af71..06b1427dcf 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/IgniteComputeImplTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/IgniteComputeImplTest.java
@@ -24,8 +24,10 @@ import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFu
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -35,6 +37,10 @@ import java.util.concurrent.CompletableFuture;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.JobStatus;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.table.IgniteTablesInternal;
import org.apache.ignite.internal.table.TableViewInternal;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
@@ -65,6 +71,15 @@ class IgniteComputeImplTest extends BaseIgniteAbstractTest {
@Mock
private ComputeComponentImpl computeComponent;
+ @Mock
+ private LogicalTopologyService logicalTopologyService;
+
+ @Mock
+ private PlacementDriver placementDriver;
+
+ @Mock
+ private HybridClock clock;
+
@InjectMocks
private IgniteComputeImpl compute;
@@ -109,10 +124,7 @@ class IgniteComputeImplTest extends BaseIgniteAbstractTest {
@Test
void executesColocatedOnLeaderNodeOfPartitionCorrespondingToTupleKey() {
respondWhenExecutingSimpleJobRemotely();
-
- when(igniteTables.tableViewAsync("TEST")).thenReturn(completedFuture(table));
- doReturn(42).when(table).partition(any());
- doReturn(remoteNode).when(table).leaderAssignment(42);
+ respondWhenAskForPrimaryReplica();
assertThat(
compute.<String>executeColocatedAsync(
@@ -129,10 +141,7 @@ class IgniteComputeImplTest extends BaseIgniteAbstractTest {
@Test
void executesColocatedOnLeaderNodeOfPartitionCorrespondingToMappedKey() {
respondWhenExecutingSimpleJobRemotely();
-
- when(igniteTables.tableViewAsync("TEST")).thenReturn(completedFuture(table));
- doReturn(42).when(table).partition(any(), any());
- doReturn(remoteNode).when(table).leaderAssignment(42);
+ respondWhenAskForPrimaryReplica();
assertThat(
compute.<Integer, String>executeColocatedAsync(
@@ -147,6 +156,15 @@ class IgniteComputeImplTest extends BaseIgniteAbstractTest {
);
}
+ private void respondWhenAskForPrimaryReplica() {
+ when(igniteTables.tableViewAsync("TEST")).thenReturn(completedFuture(table));
+ ReplicaMeta replicaMeta = mock(ReplicaMeta.class);
+ doReturn("").when(replicaMeta).getLeaseholderId();
+ CompletableFuture<ReplicaMeta> toBeReturned = completedFuture(replicaMeta);
+ doReturn(toBeReturned).when(placementDriver).awaitPrimaryReplica(any(), any(), anyLong(), any());
+ doReturn(remoteNode).when(topologyService).getById(any());
+ }
+
private void respondWhenExecutingSimpleJobLocally() {
when(computeComponent.executeLocally(testDeploymentUnits, JOB_CLASS_NAME, "a", 42))
.thenReturn(completedExecution("jobResponse"));
diff --git a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/PlacementDriverTest.java b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/PlacementDriverTest.java
index d1f8107e7a..e34cdb532f 100644
--- a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/PlacementDriverTest.java
+++ b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/PlacementDriverTest.java
@@ -64,7 +64,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-/** Tests to verify {@link PlacementDriver} implemented by {@link LeaseTracker}. */
+/** Tests to verify {@link LeaseTracker} implemented by {@link PlacementDriver}. */
public class PlacementDriverTest extends BaseIgniteAbstractTest {
private static final HybridTimestamp AWAIT_TIME_10_000 = new HybridTimestamp(10_000, 0);
diff --git a/modules/platforms/cpp/ignite/common/error_codes.h b/modules/platforms/cpp/ignite/common/error_codes.h
index 674682d103..bb80bbb1a5 100644
--- a/modules/platforms/cpp/ignite/common/error_codes.h
+++ b/modules/platforms/cpp/ignite/common/error_codes.h
@@ -183,7 +183,10 @@ enum class code : underlying_t {
RESULT_NOT_FOUND = 0x100007,
FAIL_TO_GET_JOB_STATUS = 0x100008,
COMPUTE_JOB_FAILED = 0x100009,
- CHANGE_JOB_PRIORITY = 0x10000a,
+ CHANGE_JOB_PRIORITY_NO_JOB = 0x10000a,
+ CHANGE_JOB_PRIORITY_JOB_EXECUTING = 0x10000b,
+ PRIMARY_REPLICA_RESOLVE = 0x10000c,
+ CHANGE_JOB_PRIORITY = 0x10000d,
// Catalog group. Group code: 17
VALIDATION = 0x110001,
diff --git a/modules/platforms/cpp/ignite/odbc/common_types.cpp b/modules/platforms/cpp/ignite/odbc/common_types.cpp
index 8adf041b12..47ea738abd 100644
--- a/modules/platforms/cpp/ignite/odbc/common_types.cpp
+++ b/modules/platforms/cpp/ignite/odbc/common_types.cpp
@@ -272,6 +272,9 @@ sql_state error_code_to_sql_state(error::code code) {
case error::code::RESULT_NOT_FOUND:
case error::code::FAIL_TO_GET_JOB_STATUS:
case error::code::COMPUTE_JOB_FAILED:
+ case error::code::CHANGE_JOB_PRIORITY_NO_JOB:
+ case error::code::PRIMARY_REPLICA_RESOLVE:
+ case error::code::CHANGE_JOB_PRIORITY_JOB_EXECUTING:
case error::code::CHANGE_JOB_PRIORITY:
return sql_state::SHY000_GENERAL_ERROR;
diff --git a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
index 9ccd6f89da..74debb835d 100644
--- a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
@@ -495,8 +495,17 @@ namespace Apache.Ignite
/// <summary> ComputeJobFailed error. </summary>
public const int ComputeJobFailed = (GroupCode << 16) | (9 & 0xFFFF);
+ /// <summary> ChangeJobPriorityNoJob error. </summary>
+ public const int ChangeJobPriorityNoJob = (GroupCode << 16) | (10 & 0xFFFF);
+
+ /// <summary> ChangeJobPriorityJobExecuting error. </summary>
+ public const int ChangeJobPriorityJobExecuting = (GroupCode << 16) | (11 & 0xFFFF);
+
+ /// <summary> PrimaryReplicaResolve error. </summary>
+ public const int PrimaryReplicaResolve = (GroupCode << 16) | (12 & 0xFFFF);
+
/// <summary> ChangeJobPriority error. </summary>
- public const int ChangeJobPriority = (GroupCode << 16) | (10 & 0xFFFF);
+ public const int ChangeJobPriority = (GroupCode << 16) | (13 & 0xFFFF);
}
/// <summary> Catalog errors. </summary>
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index eb686bfb14..453bba117d 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -699,7 +699,14 @@ public class IgniteImpl implements Ignite {
computeCfg
);
- compute = new IgniteComputeImpl(clusterSvc.topologyService(), distributedTblMgr, computeComponent);
+ compute = new IgniteComputeImpl(
+ placementDriverMgr.placementDriver(),
+ clusterSvc.topologyService(),
+ logicalTopologyService,
+ distributedTblMgr,
+ computeComponent,
+ clock
+ );
authenticationManager = createAuthenticationManager();