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/22 10:58:56 UTC
(ignite-3) branch main updated: IGNITE-21148 Java thin: Implement job execution interface (#3038)
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 8860541c01 IGNITE-21148 Java thin: Implement job execution interface (#3038)
8860541c01 is described below
commit 8860541c01134fabbb2549ca0368ddd74fb9fa5c
Author: Vadim Pakhnushev <86...@users.noreply.github.com>
AuthorDate: Mon Jan 22 13:58:51 2024 +0300
IGNITE-21148 Java thin: Implement job execution interface (#3038)
---
.../org/apache/ignite/compute/JobExecution.java | 17 +-
.../java/org/apache/ignite/compute/JobStatus.java | 4 +-
.../java/org/apache/ignite/lang/ErrorGroups.java | 9 +-
.../internal/client/proto/ClientMessagePacker.java | 15 ++
.../client/proto/ClientMessageUnpacker.java | 25 +++
.../ignite/internal/client/proto/ClientOp.java | 9 +
modules/client-handler/build.gradle | 2 +
.../apache/ignite/client/handler/TestServer.java | 4 +-
.../ignite/client/handler/ClientHandlerModule.java | 6 +-
.../handler/ClientInboundMessageHandler.java | 29 ++-
.../compute/ClientComputeCancelRequest.java | 52 +++++
.../ClientComputeChangePriorityRequest.java | 53 +++++
.../ClientComputeExecuteColocatedRequest.java | 21 +-
.../compute/ClientComputeExecuteRequest.java | 30 ++-
.../compute/ClientComputeGetStatusRequest.java | 66 ++++++
modules/client/build.gradle | 1 +
.../apache/ignite/client/ClientOperationType.java | 15 ++
.../org/apache/ignite/client/RetryReadPolicy.java | 3 +
.../apache/ignite/internal/client/ClientUtils.java | 9 +
.../internal/client/compute/ClientCompute.java | 96 ++++----
.../client/compute/ClientJobExecution.java | 137 ++++++++++--
.../apache/ignite/client/ClientComputeTest.java | 70 +++---
.../ignite/client/TestClientHandlerModule.java | 6 +-
.../java/org/apache/ignite/client/TestServer.java | 4 +-
.../apache/ignite/client/fakes/FakeCompute.java | 57 ++++-
.../ignite/internal/compute/ItComputeBaseTest.java | 20 +-
.../internal/compute/ItComputeTestEmbedded.java | 15 +-
.../internal/compute/ItWorkerShutdownTest.java | 6 +-
.../apache/ignite/internal/compute/Cleaner.java | 91 ++++++++
.../ignite/internal/compute/ComputeComponent.java | 30 ++-
.../internal/compute/ComputeComponentImpl.java | 82 ++++---
.../ignite/internal/compute/ComputeUtils.java | 14 +-
.../internal/compute/DelegatingJobExecution.java | 11 +-
.../ignite/internal/compute/ExecutionManager.java | 136 +++++++++++
.../internal/compute/FailSafeJobExecution.java | 11 +-
.../ignite/internal/compute/IgniteComputeImpl.java | 20 +-
.../internal/compute/IgniteComputeInternal.java | 56 +++++
.../compute/JobExecutionFutureWrapper.java | 7 +-
.../internal/compute/JobExecutionWrapper.java | 7 +-
.../internal/compute/executor/ComputeExecutor.java | 6 -
.../compute/executor/ComputeExecutorImpl.java | 11 +-
.../compute/executor/JobExecutionInternal.java | 14 +-
.../compute/message/JobCancelResponse.java | 8 +
.../compute/message/JobChangePriorityResponse.java | 8 +
.../compute/messaging/ComputeMessaging.java | 198 +++++++++++-----
.../compute/messaging/RemoteJobExecution.java | 7 +-
.../compute/queue/CancellingException.java | 32 ---
.../internal/compute/queue/QueueExecution.java | 7 +-
.../internal/compute/queue/QueueExecutionImpl.java | 16 +-
.../compute/state/InMemoryComputeStateMachine.java | 45 +---
.../internal/compute/ComputeComponentImplTest.java | 57 ++---
.../internal/compute/IgniteComputeImplTest.java | 12 +-
.../compute/executor/ComputeExecutorTest.java | 26 ++-
.../compute/queue/PriorityQueueExecutorTest.java | 25 +--
.../state/InMemoryComputeStateMachineTest.java | 4 +-
modules/platforms/cpp/ignite/common/error_codes.h | 3 +-
modules/platforms/cpp/ignite/odbc/common_types.cpp | 3 +-
.../platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs | 7 +-
.../app/client/ItAbstractThinClientTest.java | 2 +
.../runner/app/client/ItThinClientComputeTest.java | 248 ++++++++++++++++++---
.../org/apache/ignite/internal/app/IgniteImpl.java | 9 +-
61 files changed, 1508 insertions(+), 486 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/compute/JobExecution.java b/modules/api/src/main/java/org/apache/ignite/compute/JobExecution.java
index 6977fadee3..c900610ef6 100644
--- a/modules/api/src/main/java/org/apache/ignite/compute/JobExecution.java
+++ b/modules/api/src/main/java/org/apache/ignite/compute/JobExecution.java
@@ -19,6 +19,7 @@ package org.apache.ignite.compute;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import org.jetbrains.annotations.Nullable;
/**
* Job control object, provides information about the job execution process and result, allows cancelling the job.
@@ -39,7 +40,7 @@ public interface JobExecution<R> {
*
* @return The current status of the job, or {@code null} if the job status no longer exists due to exceeding the retention time limit.
*/
- CompletableFuture<JobStatus> statusAsync();
+ CompletableFuture<@Nullable JobStatus> statusAsync();
/**
* Returns the id of the job. The job status may be deleted and thus return {@code null} if the time for retaining job status has been
@@ -47,22 +48,26 @@ public interface JobExecution<R> {
*
* @return The id of the job, or {@code null} if the job status no longer exists due to exceeding the retention time limit.
*/
- default CompletableFuture<UUID> idAsync() {
+ default CompletableFuture<@Nullable UUID> idAsync() {
return statusAsync().thenApply(status -> status != null ? status.id() : null);
}
/**
* Cancels the job.
*
- * @return The future which will be completed when cancel request is processed.
+ * @return The future which will be completed with {@code true} when the job is cancelled, {@code false} when the job couldn't be
+ * cancelled (either it's not yet started, or it's already completed), or {@code null} if the job no longer exists due to
+ * exceeding the retention time limit.
*/
- CompletableFuture<Void> cancelAsync();
+ CompletableFuture<@Nullable Boolean> cancelAsync();
/**
* Changes job priority. After priority change job will be the last in the queue of jobs with the same priority.
*
* @param newPriority new priority.
- * @return The future which will be completed when change priority request is processed.
+ * @return The future which will be completed with {@code true} when the priority is changed, {@code false} when the priority couldn't
+ * be changed (it's already executing or completed), or {@code null} if the job no longer exists due to
+ * exceeding the retention time limit.
*/
- CompletableFuture<Void> changePriorityAsync(int newPriority);
+ CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority);
}
diff --git a/modules/api/src/main/java/org/apache/ignite/compute/JobStatus.java b/modules/api/src/main/java/org/apache/ignite/compute/JobStatus.java
index 1ac64ae706..3ee4ac2910 100644
--- a/modules/api/src/main/java/org/apache/ignite/compute/JobStatus.java
+++ b/modules/api/src/main/java/org/apache/ignite/compute/JobStatus.java
@@ -132,12 +132,12 @@ public class JobStatus implements Serializable {
return this;
}
- public Builder startTime(Instant startTime) {
+ public Builder startTime(@Nullable Instant startTime) {
this.startTime = startTime;
return this;
}
- public Builder finishTime(Instant finishTime) {
+ public Builder finishTime(@Nullable Instant finishTime) {
this.finishTime = finishTime;
return this;
}
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 87af8d63ec..9f21724679 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
@@ -548,17 +548,14 @@ public class ErrorGroups {
/** Compute job result not found error. */
public static final int RESULT_NOT_FOUND_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 7);
- /** Compute job status can not be retrieved. */
+ /** Compute job status can't be retrieved. */
public static final int FAIL_TO_GET_JOB_STATUS_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 8);
/** Compute job failed. */
public static final int COMPUTE_JOB_FAILED_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 9);
- /** Can not change job priority, compute job not found error. */
- public static final int CHANGE_JOB_PRIORITY_NO_JOB_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 10);
-
- /** Can not 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 change job priority. */
+ public static final int CHANGE_JOB_PRIORITY_ERR = COMPUTE_ERR_GROUP.registerErrorCode((short) 10);
}
/** Catalog error group. */
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessagePacker.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessagePacker.java
index a7ada23833..f1f3d4a29b 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessagePacker.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessagePacker.java
@@ -23,6 +23,7 @@ import static org.msgpack.core.MessagePack.Code;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import java.nio.ByteBuffer;
+import java.time.Instant;
import java.util.BitSet;
import java.util.UUID;
import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
@@ -628,6 +629,20 @@ public class ClientMessagePacker implements AutoCloseable {
writePayload(buf);
}
+ /**
+ * Packs an {@link Instant}.
+ *
+ * @param instant Instant object.
+ */
+ public void packInstant(@Nullable Instant instant) {
+ if (instant == null) {
+ packNil();
+ } else {
+ packLong(instant.getEpochSecond());
+ packInt(instant.getNano());
+ }
+ }
+
/**
* Gets metadata.
*
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessageUnpacker.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessageUnpacker.java
index 74ed2d7d2c..f0f3f4c618 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessageUnpacker.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientMessageUnpacker.java
@@ -23,10 +23,12 @@ import io.netty.buffer.ByteBuf;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
+import java.time.Instant;
import java.util.BitSet;
import java.util.UUID;
import org.apache.ignite.internal.binarytuple.BinaryTupleReader;
import org.apache.ignite.internal.util.ArrayUtils;
+import org.jetbrains.annotations.Nullable;
import org.msgpack.core.ExtensionTypeHeader;
import org.msgpack.core.MessageFormat;
import org.msgpack.core.MessageFormatException;
@@ -812,6 +814,29 @@ public class ClientMessageUnpacker implements AutoCloseable {
}
}
+ /**
+ * Reads a nullable {@link Instant}.
+ *
+ * @return {@link Instant} value or {@code null}.
+ */
+ public @Nullable Instant unpackInstantNullable() {
+ if (tryUnpackNil()) {
+ return null;
+ }
+ return unpackInstant();
+ }
+
+ /**
+ * Reads an {@link Instant}.
+ *
+ * @return {@link Instant} value.
+ */
+ public Instant unpackInstant() {
+ long seconds = unpackLong();
+ int nanos = unpackInt();
+ return Instant.ofEpochSecond(seconds, nanos);
+ }
+
private int readLength8() {
return buf.readUnsignedByte();
}
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java
index 829670ab89..699b55a307 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ClientOp.java
@@ -155,4 +155,13 @@ public class ClientOp {
/** JDBC get more results command. */
public static final int JDBC_MORE_RESULTS = 58;
+
+ /** Get compute job status. */
+ public static final int COMPUTE_GET_STATUS = 59;
+
+ /** Cancel compute job. */
+ public static final int COMPUTE_CANCEL = 60;
+
+ /** Change compute job priority. */
+ public static final int COMPUTE_CHANGE_PRIORITY = 61;
}
diff --git a/modules/client-handler/build.gradle b/modules/client-handler/build.gradle
index 18e350e975..22afafdb02 100644
--- a/modules/client-handler/build.gradle
+++ b/modules/client-handler/build.gradle
@@ -38,6 +38,7 @@ dependencies {
implementation project(':ignite-transactions')
implementation project(':ignite-catalog')
implementation project(':ignite-placement-driver-api')
+ implementation project(':ignite-compute')
implementation libs.jetbrains.annotations
implementation libs.fastutil.core
implementation libs.netty.common
@@ -70,6 +71,7 @@ dependencies {
integrationTestImplementation project(':ignite-transactions')
integrationTestImplementation project(':ignite-catalog')
integrationTestImplementation project(':ignite-placement-driver-api')
+ integrationTestImplementation project(':ignite-compute')
integrationTestImplementation(testFixtures(project(':ignite-configuration')))
integrationTestImplementation(testFixtures(project(':ignite-core')))
integrationTestImplementation(testFixtures(project(':ignite-table')))
diff --git a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/TestServer.java b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/TestServer.java
index 07dfcbea2e..619e4e778f 100644
--- a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/TestServer.java
+++ b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/TestServer.java
@@ -25,8 +25,8 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.configuration.ClientConnectorConfiguration;
-import org.apache.ignite.compute.IgniteCompute;
import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
import org.apache.ignite.internal.configuration.ConfigurationManager;
import org.apache.ignite.internal.configuration.ConfigurationTreeGenerator;
import org.apache.ignite.internal.configuration.storage.TestConfigurationStorage;
@@ -125,7 +125,7 @@ public class TestServer {
mock(IgniteTablesInternal.class),
mock(IgniteTransactionsImpl.class),
registry,
- mock(IgniteCompute.class),
+ mock(IgniteComputeInternal.class),
clusterService,
bootstrapFactory,
mock(IgniteSql.class),
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
index 7a894cfc0a..f6c1a51cd2 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
@@ -38,9 +38,9 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import org.apache.ignite.client.handler.configuration.ClientConnectorConfiguration;
import org.apache.ignite.client.handler.configuration.ClientConnectorView;
-import org.apache.ignite.compute.IgniteCompute;
import org.apache.ignite.internal.catalog.CatalogService;
import org.apache.ignite.internal.client.proto.ClientMessageDecoder;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
import org.apache.ignite.internal.configuration.ConfigurationRegistry;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.lang.IgniteInternalException;
@@ -104,7 +104,7 @@ public class ClientHandlerModule implements IgniteComponent {
private final QueryProcessor queryProcessor;
/** Compute. */
- private final IgniteCompute igniteCompute;
+ private final IgniteComputeInternal igniteCompute;
/** Cluster. */
private final ClusterService clusterService;
@@ -150,7 +150,7 @@ public class ClientHandlerModule implements IgniteComponent {
IgniteTablesInternal igniteTables,
IgniteTransactionsImpl igniteTransactions,
ConfigurationRegistry registry,
- IgniteCompute igniteCompute,
+ IgniteComputeInternal igniteCompute,
ClusterService clusterService,
NettyBootstrapFactory bootstrapFactory,
IgniteSql sql,
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
index 7b436b85ff..f13f571024 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
@@ -42,8 +42,11 @@ import java.util.function.Consumer;
import javax.net.ssl.SSLException;
import org.apache.ignite.client.handler.configuration.ClientConnectorView;
import org.apache.ignite.client.handler.requests.cluster.ClientClusterGetNodesRequest;
+import org.apache.ignite.client.handler.requests.compute.ClientComputeCancelRequest;
+import org.apache.ignite.client.handler.requests.compute.ClientComputeChangePriorityRequest;
import org.apache.ignite.client.handler.requests.compute.ClientComputeExecuteColocatedRequest;
import org.apache.ignite.client.handler.requests.compute.ClientComputeExecuteRequest;
+import org.apache.ignite.client.handler.requests.compute.ClientComputeGetStatusRequest;
import org.apache.ignite.client.handler.requests.jdbc.ClientJdbcCloseRequest;
import org.apache.ignite.client.handler.requests.jdbc.ClientJdbcColumnMetadataRequest;
import org.apache.ignite.client.handler.requests.jdbc.ClientJdbcConnectRequest;
@@ -86,7 +89,6 @@ import org.apache.ignite.client.handler.requests.table.ClientTupleUpsertRequest;
import org.apache.ignite.client.handler.requests.tx.ClientTransactionBeginRequest;
import org.apache.ignite.client.handler.requests.tx.ClientTransactionCommitRequest;
import org.apache.ignite.client.handler.requests.tx.ClientTransactionRollbackRequest;
-import org.apache.ignite.compute.IgniteCompute;
import org.apache.ignite.internal.catalog.CatalogService;
import org.apache.ignite.internal.client.proto.ClientMessageCommon;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
@@ -96,6 +98,7 @@ import org.apache.ignite.internal.client.proto.ErrorExtensions;
import org.apache.ignite.internal.client.proto.HandshakeExtension;
import org.apache.ignite.internal.client.proto.ProtocolVersion;
import org.apache.ignite.internal.client.proto.ResponseFlags;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
import org.apache.ignite.internal.event.EventListener;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridTimestamp;
@@ -155,7 +158,7 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
private final ClientConnectorView configuration;
/** Compute. */
- private final IgniteCompute compute;
+ private final IgniteComputeInternal compute;
/** Cluster. */
private final ClusterService clusterService;
@@ -219,7 +222,7 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
IgniteTransactionsImpl igniteTransactions,
QueryProcessor processor,
ClientConnectorView configuration,
- IgniteCompute compute,
+ IgniteComputeInternal compute,
ClusterService clusterService,
IgniteSql sql,
CompletableFuture<UUID> clusterId,
@@ -713,10 +716,26 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
return ClientTransactionRollbackRequest.process(in, resources, metrics);
case ClientOp.COMPUTE_EXECUTE:
- return ClientComputeExecuteRequest.process(in, compute, clusterService, notificationSender(requestId));
+ return ClientComputeExecuteRequest.process(in, out, compute, clusterService, notificationSender(requestId));
case ClientOp.COMPUTE_EXECUTE_COLOCATED:
- return ClientComputeExecuteColocatedRequest.process(in, out, compute, igniteTables, notificationSender(requestId));
+ return ClientComputeExecuteColocatedRequest.process(
+ in,
+ out,
+ compute,
+ igniteTables,
+ clusterService,
+ notificationSender(requestId)
+ );
+
+ case ClientOp.COMPUTE_GET_STATUS:
+ return ClientComputeGetStatusRequest.process(in, out, compute);
+
+ case ClientOp.COMPUTE_CANCEL:
+ return ClientComputeCancelRequest.process(in, out, compute);
+
+ case ClientOp.COMPUTE_CHANGE_PRIORITY:
+ return ClientComputeChangePriorityRequest.process(in, out, compute);
case ClientOp.CLUSTER_GET_NODES:
return ClientClusterGetNodesRequest.process(out, clusterService);
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeCancelRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeCancelRequest.java
new file mode 100644
index 0000000000..2f09072bde
--- /dev/null
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeCancelRequest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.client.handler.requests.compute;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.client.proto.ClientMessagePacker;
+import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
+
+/**
+ * Cancel compute job request.
+ */
+public class ClientComputeCancelRequest {
+ /**
+ * Processes the request.
+ *
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param compute Compute.
+ * @return Future.
+ */
+ public static CompletableFuture<Void> process(
+ ClientMessageUnpacker in,
+ ClientMessagePacker out,
+ IgniteComputeInternal compute
+ ) {
+ UUID jobId = in.unpackUuid();
+ return compute.cancelAsync(jobId).thenAccept(cancelResult -> {
+ if (cancelResult == null) {
+ out.packNil();
+ } else {
+ out.packBoolean(cancelResult);
+ }
+ });
+ }
+}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeChangePriorityRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeChangePriorityRequest.java
new file mode 100644
index 0000000000..802f5909a1
--- /dev/null
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeChangePriorityRequest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.client.handler.requests.compute;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.client.proto.ClientMessagePacker;
+import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
+
+/**
+ * Change compute job priority.
+ */
+public class ClientComputeChangePriorityRequest {
+ /**
+ * Processes the request.
+ *
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param compute Compute.
+ * @return Future.
+ */
+ public static CompletableFuture<Void> process(
+ ClientMessageUnpacker in,
+ ClientMessagePacker out,
+ IgniteComputeInternal compute
+ ) {
+ UUID jobId = in.unpackUuid();
+ int newPriority = in.unpackInt();
+ return compute.changePriorityAsync(jobId, newPriority).thenAccept(result -> {
+ if (result == null) {
+ out.packNil();
+ } else {
+ out.packBoolean(result);
+ }
+ });
+ }
+}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteColocatedRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteColocatedRequest.java
index 13da450c32..0ef9595187 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteColocatedRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteColocatedRequest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.client.handler.requests.compute;
+import static org.apache.ignite.client.handler.requests.compute.ClientComputeExecuteRequest.sendResultAndStatus;
import static org.apache.ignite.client.handler.requests.compute.ClientComputeExecuteRequest.unpackArgs;
import static org.apache.ignite.client.handler.requests.compute.ClientComputeExecuteRequest.unpackDeploymentUnits;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
@@ -27,8 +28,10 @@ import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.NotificationSender;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.IgniteCompute;
+import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.network.ClusterService;
import org.apache.ignite.table.manager.IgniteTables;
/**
@@ -38,10 +41,11 @@ public class ClientComputeExecuteColocatedRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
- * @param compute Compute.
- * @param tables Tables.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param compute Compute.
+ * @param tables Tables.
+ * @param cluster Cluster service
* @return Future.
*/
public static CompletableFuture<Void> process(
@@ -49,18 +53,19 @@ public class ClientComputeExecuteColocatedRequest {
ClientMessagePacker out,
IgniteCompute compute,
IgniteTables tables,
+ ClusterService cluster,
NotificationSender notificationSender) {
return readTableAsync(in, tables).thenCompose(table -> {
- return readTuple(in, table, true).thenAccept(keyTuple -> {
+ return readTuple(in, table, true).thenCompose(keyTuple -> {
List<DeploymentUnit> deploymentUnits = unpackDeploymentUnits(in);
String jobClassName = in.unpackString();
Object[] args = unpackArgs(in);
out.packInt(table.schemaView().lastKnownSchemaVersion());
- compute.executeColocatedAsync(table.name(), keyTuple, deploymentUnits, jobClassName, args)
- .resultAsync()
- .whenComplete((val, err) -> notificationSender.sendNotification(w -> w.packObjectAsBinaryTuple(val), err));
+ JobExecution<Object> execution = compute.executeColocatedAsync(table.name(), keyTuple, deploymentUnits, jobClassName, args);
+ sendResultAndStatus(execution, notificationSender);
+ return execution.idAsync().thenAccept(out::packUuid);
});
});
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteRequest.java
index 82fdfd5e30..b60c8dc46b 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteRequest.java
@@ -17,6 +17,8 @@
package org.apache.ignite.client.handler.requests.compute;
+import static org.apache.ignite.client.handler.requests.compute.ClientComputeGetStatusRequest.packJobStatus;
+
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
@@ -24,10 +26,11 @@ import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.NotificationSender;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.IgniteCompute;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.network.ClusterService;
-import org.jetbrains.annotations.Nullable;
/**
* Compute execute request.
@@ -36,14 +39,16 @@ public class ClientComputeExecuteRequest {
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param compute Compute.
- * @param cluster Cluster.
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param compute Compute.
+ * @param cluster Cluster.
* @param notificationSender Notification sender.
* @return Future.
*/
- public static @Nullable CompletableFuture<Void> process(
+ public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
+ ClientMessagePacker out,
IgniteCompute compute,
ClusterService cluster,
NotificationSender notificationSender) {
@@ -61,11 +66,18 @@ public class ClientComputeExecuteRequest {
String jobClassName = in.unpackString();
Object[] args = unpackArgs(in);
- compute.executeAsync(Set.of(node), deploymentUnits, jobClassName, args)
- .resultAsync()
- .whenComplete((res, err) -> notificationSender.sendNotification(w -> w.packObjectAsBinaryTuple(res), err));
+ JobExecution<Object> execution = compute.executeAsync(Set.of(node), deploymentUnits, jobClassName, args);
+ sendResultAndStatus(execution, notificationSender);
+ return execution.idAsync().thenAccept(out::packUuid);
+ }
- return null;
+ static void sendResultAndStatus(JobExecution<Object> execution, NotificationSender notificationSender) {
+ execution.resultAsync().whenComplete((val, err) ->
+ execution.statusAsync().whenComplete((status, errStatus) ->
+ notificationSender.sendNotification(w -> {
+ w.packObjectAsBinaryTuple(val);
+ packJobStatus(w, status);
+ }, err)));
}
/**
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeGetStatusRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeGetStatusRequest.java
new file mode 100644
index 0000000000..2e09582670
--- /dev/null
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeGetStatusRequest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.client.handler.requests.compute;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.compute.JobStatus;
+import org.apache.ignite.internal.client.proto.ClientMessagePacker;
+import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Compute execute request.
+ */
+public class ClientComputeGetStatusRequest {
+ /**
+ * Processes the request.
+ *
+ * @param in Unpacker.
+ * @param out Packer.
+ * @param compute Compute.
+ * @return Future.
+ */
+ public static CompletableFuture<Void> process(
+ ClientMessageUnpacker in,
+ ClientMessagePacker out,
+ IgniteComputeInternal compute
+ ) {
+ UUID jobId = in.unpackUuid();
+ return compute.statusAsync(jobId).thenAccept(jobStatus -> packJobStatus(out, jobStatus));
+ }
+
+ /**
+ * Writes a {@link JobStatus}.
+ *
+ * @param out Packer.
+ * @param jobStatus Job status.
+ */
+ static void packJobStatus(ClientMessagePacker out, @Nullable JobStatus jobStatus) {
+ if (jobStatus == null) {
+ out.packNil();
+ } else {
+ out.packUuid(jobStatus.id());
+ out.packInt(jobStatus.state().ordinal());
+ out.packInstant(jobStatus.createTime());
+ out.packInstant(jobStatus.startTime());
+ out.packInstant(jobStatus.finishTime());
+ }
+ }
+}
diff --git a/modules/client/build.gradle b/modules/client/build.gradle
index f423b3d6d4..56ce5f33b0 100644
--- a/modules/client/build.gradle
+++ b/modules/client/build.gradle
@@ -53,6 +53,7 @@ dependencies {
testImplementation project(':ignite-metrics')
testImplementation project(':ignite-catalog')
testImplementation project(':ignite-placement-driver-api')
+ testImplementation project(':ignite-compute')
testImplementation(testFixtures(project(':ignite-core')))
testImplementation(testFixtures(project(':ignite-configuration')))
testImplementation(testFixtures(project(':ignite-sql-engine')))
diff --git a/modules/client/src/main/java/org/apache/ignite/client/ClientOperationType.java b/modules/client/src/main/java/org/apache/ignite/client/ClientOperationType.java
index 76ab308314..d57eb8755f 100644
--- a/modules/client/src/main/java/org/apache/ignite/client/ClientOperationType.java
+++ b/modules/client/src/main/java/org/apache/ignite/client/ClientOperationType.java
@@ -129,6 +129,21 @@ public enum ClientOperationType {
*/
COMPUTE_EXECUTE,
+ /**
+ * Get compute job status ({@link org.apache.ignite.compute.JobExecution#statusAsync()}).
+ */
+ COMPUTE_GET_STATUS,
+
+ /**
+ * Cancel compute job ({@link org.apache.ignite.compute.JobExecution#cancelAsync()}).
+ */
+ COMPUTE_CANCEL,
+
+ /**
+ * Change compute job priority ({@link org.apache.ignite.compute.JobExecution#changePriorityAsync(int)}).
+ */
+ COMPUTE_CHANGE_PRIORITY,
+
/**
* SQL Execute ({@link org.apache.ignite.sql.Session#executeAsync(Transaction, String, Object...)}).
*/
diff --git a/modules/client/src/main/java/org/apache/ignite/client/RetryReadPolicy.java b/modules/client/src/main/java/org/apache/ignite/client/RetryReadPolicy.java
index ea15466407..54f8f043ef 100644
--- a/modules/client/src/main/java/org/apache/ignite/client/RetryReadPolicy.java
+++ b/modules/client/src/main/java/org/apache/ignite/client/RetryReadPolicy.java
@@ -39,6 +39,9 @@ public class RetryReadPolicy extends RetryLimitPolicy {
case TUPLE_UPSERT:
case COMPUTE_EXECUTE:
+ case COMPUTE_GET_STATUS:
+ case COMPUTE_CANCEL:
+ case COMPUTE_CHANGE_PRIORITY:
case TUPLE_GET_AND_DELETE:
case TUPLE_DELETE_ALL_EXACT:
case TUPLE_DELETE_EXACT:
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/ClientUtils.java b/modules/client/src/main/java/org/apache/ignite/internal/client/ClientUtils.java
index 88a8d9924a..3368af1ebe 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/ClientUtils.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/ClientUtils.java
@@ -215,6 +215,15 @@ public class ClientUtils {
case ClientOp.COMPUTE_EXECUTE_COLOCATED:
return ClientOperationType.COMPUTE_EXECUTE;
+ case ClientOp.COMPUTE_GET_STATUS:
+ return ClientOperationType.COMPUTE_GET_STATUS;
+
+ case ClientOp.COMPUTE_CANCEL:
+ return ClientOperationType.COMPUTE_CANCEL;
+
+ case ClientOp.COMPUTE_CHANGE_PRIORITY:
+ return ClientOperationType.COMPUTE_CHANGE_PRIORITY;
+
case ClientOp.CLUSTER_GET_NODES:
return null;
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java
index 4b39cbdebb..c4c4f167b8 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.client.compute;
+import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.lang.ErrorGroups.Client.TABLE_ID_NOT_FOUND_ERR;
import java.util.HashMap;
@@ -30,6 +31,8 @@ import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.IgniteCompute;
import org.apache.ignite.compute.JobExecution;
@@ -59,9 +62,6 @@ import org.apache.ignite.table.mapper.Mapper;
public class ClientCompute implements IgniteCompute {
private static final String DEFAULT_SCHEMA_NAME = "PUBLIC";
- /** Indicates a missing table. */
- private static final Object MISSING_TABLE_TOKEN = new Object();
-
/** Channel. */
private final ReliableChannel ch;
@@ -95,7 +95,7 @@ public class ClientCompute implements IgniteCompute {
ClusterNode node = randomNode(nodes);
- return new ClientJobExecution<>(executeOnOneNode(node, units, jobClassName, args));
+ return new ClientJobExecution<>(ch, executeOnOneNode(node, units, jobClassName, args));
}
/** {@inheritDoc} */
@@ -127,7 +127,7 @@ public class ClientCompute implements IgniteCompute {
Objects.requireNonNull(units);
Objects.requireNonNull(jobClassName);
- return new ClientJobExecution<>(doExecuteColocatedAsync(tableName, key, units, jobClassName, args));
+ return new ClientJobExecution<>(ch, doExecuteColocatedAsync(tableName, key, units, jobClassName, args));
}
/** {@inheritDoc} */
@@ -146,10 +146,10 @@ public class ClientCompute implements IgniteCompute {
Objects.requireNonNull(units);
Objects.requireNonNull(jobClassName);
- return new ClientJobExecution<>(doExecuteColocatedAsync(tableName, key, keyMapper, units, jobClassName, args));
+ return new ClientJobExecution<>(ch, doExecuteColocatedAsync(tableName, key, keyMapper, units, jobClassName, args));
}
- private <R> CompletableFuture<R> doExecuteColocatedAsync(
+ private CompletableFuture<PayloadInputChannel> doExecuteColocatedAsync(
String tableName,
Tuple key,
List<DeploymentUnit> units,
@@ -157,16 +157,17 @@ public class ClientCompute implements IgniteCompute {
Object... args
) {
return getTable(tableName)
- .thenCompose(table -> (CompletableFuture<R>) executeColocatedTupleKey(table, key, units, jobClassName, args))
- .handle((res, err) -> handleMissingTable(tableName, res, err))
- .thenCompose(r ->
- // If a table was dropped, try again: maybe a new table was created with the same name and new id.
- r == MISSING_TABLE_TOKEN
- ? doExecuteColocatedAsync(tableName, key, units, jobClassName, args)
- : CompletableFuture.completedFuture(r));
+ .thenCompose(table -> executeColocatedTupleKey(table, key, units, jobClassName, args))
+ .handle((res, err) -> handleMissingTable(
+ tableName,
+ res,
+ err,
+ () -> doExecuteColocatedAsync(tableName, key, units, jobClassName, args)
+ ))
+ .thenCompose(Function.identity());
}
- private <K, R> CompletableFuture<R> doExecuteColocatedAsync(
+ private <K> CompletableFuture<PayloadInputChannel> doExecuteColocatedAsync(
String tableName,
K key,
Mapper<K> keyMapper,
@@ -175,13 +176,14 @@ public class ClientCompute implements IgniteCompute {
Object... args
) {
return getTable(tableName)
- .thenCompose(table -> (CompletableFuture<R>) executeColocatedObjectKey(table, key, keyMapper, units, jobClassName, args))
- .handle((res, err) -> handleMissingTable(tableName, res, err))
- .thenCompose(r ->
- // If a table was dropped, try again: maybe a new table was created with the same name and new id.
- r == MISSING_TABLE_TOKEN
- ? doExecuteColocatedAsync(tableName, key, keyMapper, units, jobClassName, args)
- : CompletableFuture.completedFuture(r));
+ .thenCompose(table -> executeColocatedObjectKey(table, key, keyMapper, units, jobClassName, args))
+ .handle((res, err) -> handleMissingTable(
+ tableName,
+ res,
+ err,
+ () -> doExecuteColocatedAsync(tableName, key, keyMapper, units, jobClassName, args)
+ ))
+ .thenCompose(Function.identity());
}
/** {@inheritDoc} */
@@ -232,7 +234,7 @@ public class ClientCompute implements IgniteCompute {
Map<ClusterNode, JobExecution<R>> map = new HashMap<>(nodes.size());
for (ClusterNode node : nodes) {
- ClientJobExecution<R> execution = new ClientJobExecution<>(executeOnOneNode(node, units, jobClassName, args));
+ ClientJobExecution<R> execution = new ClientJobExecution<>(ch, executeOnOneNode(node, units, jobClassName, args));
if (map.put(node, execution) != null) {
throw new IllegalStateException("Node can't be specified more than once: " + node);
}
@@ -241,8 +243,13 @@ public class ClientCompute implements IgniteCompute {
return map;
}
- private <R> CompletableFuture<R> executeOnOneNode(ClusterNode node, List<DeploymentUnit> units, String jobClassName, Object[] args) {
- var reqFut = ch.serviceAsync(
+ private CompletableFuture<PayloadInputChannel> executeOnOneNode(
+ ClusterNode node,
+ List<DeploymentUnit> units,
+ String jobClassName,
+ Object[] args
+ ) {
+ return ch.serviceAsync(
ClientOp.COMPUTE_EXECUTE,
w -> {
if (w.clientChannel().protocolContext().clusterNode().name().equals(node.name())) {
@@ -257,15 +264,6 @@ public class ClientCompute implements IgniteCompute {
node.name(),
null,
true);
-
- return reqFut
- .thenCompose(PayloadInputChannel::notificationFuture)
- .thenApply(r -> {
- // Notifications require explicit input close.
- try (r) {
- return (R) r.in().unpackObjectFromBinaryTuple();
- }
- });
}
private static ClusterNode randomNode(Set<ClusterNode> nodes) {
@@ -283,7 +281,7 @@ public class ClientCompute implements IgniteCompute {
return iterator.next();
}
- private static <K, R> CompletableFuture<R> executeColocatedObjectKey(
+ private static <K> CompletableFuture<PayloadInputChannel> executeColocatedObjectKey(
ClientTable t,
K key,
Mapper<K> keyMapper,
@@ -299,7 +297,7 @@ public class ClientCompute implements IgniteCompute {
args);
}
- private static <R> CompletableFuture<R> executeColocatedTupleKey(
+ private static CompletableFuture<PayloadInputChannel> executeColocatedTupleKey(
ClientTable t,
Tuple key,
List<DeploymentUnit> units,
@@ -314,14 +312,14 @@ public class ClientCompute implements IgniteCompute {
args);
}
- private static <R> CompletableFuture<R> executeColocatedInternal(
+ private static CompletableFuture<PayloadInputChannel> executeColocatedInternal(
ClientTable t,
BiConsumer<PayloadOutputChannel, ClientSchema> keyWriter,
PartitionAwarenessProvider partitionAwarenessProvider,
List<DeploymentUnit> units,
String jobClassName,
Object[] args) {
- var reqFut = t.doSchemaOutOpAsync(
+ return t.doSchemaOutOpAsync(
ClientOp.COMPUTE_EXECUTE_COLOCATED,
(schema, outputChannel) -> {
ClientMessagePacker w = outputChannel.out();
@@ -336,15 +334,6 @@ public class ClientCompute implements IgniteCompute {
ch -> ch,
partitionAwarenessProvider,
true);
-
- return reqFut
- .thenCompose(PayloadInputChannel::notificationFuture)
- .thenApply(r -> {
- // Notifications require explicit input close.
- try (r) {
- return (R) r.in().unpackObjectFromBinaryTuple();
- }
- });
}
private CompletableFuture<ClientTable> getTable(String tableName) {
@@ -352,7 +341,7 @@ public class ClientCompute implements IgniteCompute {
var cached = tableCache.get(tableName);
if (cached != null) {
- return CompletableFuture.completedFuture(cached);
+ return completedFuture(cached);
}
return tables.tableAsync(tableName).thenApply(t -> {
@@ -367,7 +356,12 @@ public class ClientCompute implements IgniteCompute {
});
}
- private <R> R handleMissingTable(String tableName, R res, Throwable err) {
+ private CompletableFuture<PayloadInputChannel> handleMissingTable(
+ String tableName,
+ PayloadInputChannel res,
+ Throwable err,
+ Supplier<CompletableFuture<PayloadInputChannel>> retry
+ ) {
if (err instanceof CompletionException) {
err = err.getCause();
}
@@ -379,7 +373,7 @@ public class ClientCompute implements IgniteCompute {
// Table was dropped - remove from cache.
tableCache.remove(tableName);
- return (R) MISSING_TABLE_TOKEN;
+ return retry.get();
}
}
@@ -387,7 +381,7 @@ public class ClientCompute implements IgniteCompute {
throw new CompletionException(err);
}
- return res;
+ return completedFuture(res);
}
private static void packJob(ClientMessagePacker w, List<DeploymentUnit> units, String jobClassName, Object[] args) {
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientJobExecution.java b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientJobExecution.java
index d61822ebf0..e55eb6fc28 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientJobExecution.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientJobExecution.java
@@ -17,43 +17,152 @@
package org.apache.ignite.internal.client.compute;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
+import static org.apache.ignite.lang.ErrorGroups.Client.PROTOCOL_ERR;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobState;
import org.apache.ignite.compute.JobStatus;
+import org.apache.ignite.internal.client.PayloadInputChannel;
+import org.apache.ignite.internal.client.ReliableChannel;
+import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.client.proto.ClientOp;
+import org.apache.ignite.lang.IgniteException;
+import org.jetbrains.annotations.Nullable;
/**
* Client job execution implementation.
*/
class ClientJobExecution<R> implements JobExecution<R> {
- private final CompletableFuture<R> result;
+ private static final JobState[] JOB_STATES = JobState.values();
- ClientJobExecution(CompletableFuture<R> result) {
- this.result = result;
+ private final ReliableChannel ch;
+
+ private final CompletableFuture<UUID> jobIdFuture;
+
+ private final CompletableFuture<R> resultAsync;
+
+ // Local status cache
+ private final CompletableFuture<@Nullable JobStatus> statusFuture = new CompletableFuture<>();
+
+ ClientJobExecution(ReliableChannel ch, CompletableFuture<PayloadInputChannel> reqFuture) {
+ this.ch = ch;
+
+ jobIdFuture = reqFuture.thenApply(r -> r.in().unpackUuid());
+
+ resultAsync = reqFuture
+ .thenCompose(PayloadInputChannel::notificationFuture)
+ .thenApply(r -> {
+ // Notifications require explicit input close.
+ try (r) {
+ R result = (R) r.in().unpackObjectFromBinaryTuple();
+ statusFuture.complete(unpackJobStatus(r));
+ return result;
+ }
+ });
}
@Override
public CompletableFuture<R> resultAsync() {
- return result;
+ return resultAsync;
}
@Override
- public CompletableFuture<JobStatus> statusAsync() {
- // TODO https://issues.apache.org/jira/browse/IGNITE-21148
- return nullCompletedFuture();
+ public CompletableFuture<@Nullable JobStatus> statusAsync() {
+ if (statusFuture.isDone()) {
+ return statusFuture;
+ }
+ return jobIdFuture.thenCompose(this::getJobStatus);
}
@Override
- public CompletableFuture<Void> cancelAsync() {
- // TODO https://issues.apache.org/jira/browse/IGNITE-21148
- return nullCompletedFuture();
+ public CompletableFuture<@Nullable Boolean> cancelAsync() {
+ if (statusFuture.isDone()) {
+ return falseCompletedFuture();
+ }
+ return jobIdFuture.thenCompose(this::cancelJob);
}
@Override
- public CompletableFuture<Void> changePriorityAsync(int newPriority) {
- // TODO https://issues.apache.org/jira/browse/IGNITE-21148
- return nullCompletedFuture();
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) {
+ if (statusFuture.isDone()) {
+ return falseCompletedFuture();
+ }
+ return jobIdFuture.thenCompose(jobId -> changePriority(jobId, newPriority));
+ }
+
+ private CompletableFuture<@Nullable JobStatus> getJobStatus(UUID jobId) {
+ // Send the request to any node, the request will be broadcast since client doesn't know which particular node is running the job
+ // especially in case of colocated execution.
+ return ch.serviceAsync(
+ ClientOp.COMPUTE_GET_STATUS,
+ w -> w.out().packUuid(jobId),
+ ClientJobExecution::unpackJobStatus,
+ null,
+ null,
+ false
+ );
+ }
+
+ private CompletableFuture<@Nullable Boolean> cancelJob(UUID jobId) {
+ // Send the request to any node, the request will be broadcast since client doesn't know which particular node is running the job
+ // especially in case of colocated execution.
+ return ch.serviceAsync(
+ ClientOp.COMPUTE_CANCEL,
+ w -> w.out().packUuid(jobId),
+ ClientJobExecution::unpackBooleanResult,
+ null,
+ null,
+ false
+ );
+ }
+
+ private CompletableFuture<@Nullable Boolean> changePriority(UUID jobId, int newPriority) {
+ // Send the request to any node, the request will be broadcast since client doesn't know which particular node is running the job
+ // especially in case of colocated execution.
+ return ch.serviceAsync(
+ ClientOp.COMPUTE_CHANGE_PRIORITY,
+ w -> {
+ w.out().packUuid(jobId);
+ w.out().packInt(newPriority);
+ },
+ ClientJobExecution::unpackBooleanResult,
+ null,
+ null,
+ false
+ );
+ }
+
+ private static @Nullable JobStatus unpackJobStatus(PayloadInputChannel payloadInputChannel) {
+ ClientMessageUnpacker unpacker = payloadInputChannel.in();
+ if (unpacker.tryUnpackNil()) {
+ return null;
+ }
+ return JobStatus.builder()
+ .id(unpacker.unpackUuid())
+ .state(unpackJobState(unpacker))
+ .createTime(unpacker.unpackInstant())
+ .startTime(unpacker.unpackInstantNullable())
+ .finishTime(unpacker.unpackInstantNullable())
+ .build();
+ }
+
+ private static @Nullable Boolean unpackBooleanResult(PayloadInputChannel payloadInputChannel) {
+ ClientMessageUnpacker unpacker = payloadInputChannel.in();
+ if (unpacker.tryUnpackNil()) {
+ return null;
+ }
+ return unpacker.unpackBoolean();
+ }
+
+ private static JobState unpackJobState(ClientMessageUnpacker unpacker) {
+ int id = unpacker.unpackInt();
+ if (id >= 0 && id < JOB_STATES.length) {
+ return JOB_STATES[id];
+ }
+ throw new IgniteException(PROTOCOL_ERR, "Invalid job state id: " + id);
}
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/ClientComputeTest.java b/modules/client/src/test/java/org/apache/ignite/client/ClientComputeTest.java
index 2102d130f5..c9ed725386 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/ClientComputeTest.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/ClientComputeTest.java
@@ -19,8 +19,11 @@ package org.apache.ignite.client;
import static org.apache.ignite.client.AbstractClientTest.getClient;
import static org.apache.ignite.client.AbstractClientTest.getClusterNodes;
+import static org.apache.ignite.compute.JobState.COMPLETED;
+import static org.apache.ignite.compute.JobState.FAILED;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrowFast;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStatusMatcher.jobStatusWithState;
import static org.apache.ignite.lang.ErrorGroups.Table.TABLE_NOT_FOUND_ERR;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
@@ -39,6 +42,7 @@ import org.apache.ignite.client.fakes.FakeCompute;
import org.apache.ignite.client.fakes.FakeIgnite;
import org.apache.ignite.client.fakes.FakeIgniteTables;
import org.apache.ignite.compute.DeploymentUnit;
+import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.version.Version;
import org.apache.ignite.internal.client.table.ClientTable;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
@@ -69,6 +73,7 @@ public class ClientComputeTest extends BaseIgniteAbstractTest {
void tearDown() throws Exception {
IgniteUtils.closeAll(server1, server2, server3);
FakeCompute.future = null;
+ FakeCompute.status = null;
}
@Test
@@ -79,18 +84,17 @@ public class ClientComputeTest extends BaseIgniteAbstractTest {
try (var client = getClient(server1, server2, server3, server1, server2)) {
assertTrue(IgniteTestUtils.waitForCondition(() -> client.connections().size() == 3, 3000));
- CompletableFuture<String> fut1 = client.compute()
- .<String>executeAsync(getClusterNodes("s1"), List.of(), "job", new Object[]{}).resultAsync();
+ JobExecution<String> execution1 = client.compute().executeAsync(getClusterNodes("s1"), List.of(), "job");
+ JobExecution<String> execution2 = client.compute().executeAsync(getClusterNodes("s2"), List.of(), "job");
+ JobExecution<String> execution3 = client.compute().executeAsync(getClusterNodes("s3"), List.of(), "job");
- CompletableFuture<String> fut2 = client.compute()
- .<String>executeAsync(getClusterNodes("s2"), List.of(), "job", new Object[]{}).resultAsync();
+ assertThat(execution1.resultAsync(), willBe("s1"));
+ assertThat(execution2.resultAsync(), willBe("s2"));
+ assertThat(execution3.resultAsync(), willBe("s3"));
- CompletableFuture<String> fut3 = client.compute()
- .<String>executeAsync(getClusterNodes("s3"), List.of(), "job", new Object[]{}).resultAsync();
-
- assertThat(fut1, willBe("s1"));
- assertThat(fut2, willBe("s2"));
- assertThat(fut3, willBe("s3"));
+ assertThat(execution1.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ assertThat(execution2.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ assertThat(execution3.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
}
}
@@ -99,18 +103,17 @@ public class ClientComputeTest extends BaseIgniteAbstractTest {
initServers(reqId -> false);
try (var client = getClient(server3)) {
- CompletableFuture<String> fut1 = client.compute()
- .<String>executeAsync(getClusterNodes("s1"), List.of(), "job", new Object[]{}).resultAsync();
-
- CompletableFuture<String> fut2 = client.compute()
- .<String>executeAsync(getClusterNodes("s2"), List.of(), "job", new Object[]{}).resultAsync();
+ JobExecution<String> execution1 = client.compute().executeAsync(getClusterNodes("s1"), List.of(), "job");
+ JobExecution<String> execution2 = client.compute().executeAsync(getClusterNodes("s2"), List.of(), "job");
+ JobExecution<String> execution3 = client.compute().executeAsync(getClusterNodes("s3"), List.of(), "job");
- CompletableFuture<String> fut3 = client.compute()
- .<String>executeAsync(getClusterNodes("s3"), List.of(), "job", new Object[]{}).resultAsync();
+ assertThat(execution1.resultAsync(), willBe("s3"));
+ assertThat(execution2.resultAsync(), willBe("s3"));
+ assertThat(execution3.resultAsync(), willBe("s3"));
- assertThat(fut1, willBe("s3"));
- assertThat(fut2, willBe("s3"));
- assertThat(fut3, willBe("s3"));
+ assertThat(execution1.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ assertThat(execution2.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ assertThat(execution3.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
}
}
@@ -124,7 +127,7 @@ public class ClientComputeTest extends BaseIgniteAbstractTest {
var nodeName = "s" + nodeId;
CompletableFuture<String> fut = client.compute()
- .<String>executeAsync(getClusterNodes(nodeName), List.of(), "job", new Object[]{}).resultAsync();
+ .<String>executeAsync(getClusterNodes(nodeName), List.of(), "job").resultAsync();
assertThat(fut, willBe("s3"));
}
@@ -153,14 +156,17 @@ public class ClientComputeTest extends BaseIgniteAbstractTest {
try (var client = getClient(server2)) {
Tuple key = Tuple.create().set("key", "k");
- CompletableFuture<String> fut1 = client.compute()
- .<String>executeColocatedAsync(TABLE_NAME, key, List.of(), "job").resultAsync();
+ JobExecution<String> execution1 = client.compute()
+ .executeColocatedAsync(TABLE_NAME, key, List.of(), "job");
+
+ JobExecution<String> execution2 = client.compute()
+ .executeColocatedAsync(TABLE_NAME, 1L, Mapper.of(Long.class), List.of(), "job");
- CompletableFuture<String> fut2 = client.compute()
- .<Long, String>executeColocatedAsync(TABLE_NAME, 1L, Mapper.of(Long.class), List.of(), "job").resultAsync();
+ assertThat(execution1.resultAsync(), willBe("s2"));
+ assertThat(execution2.resultAsync(), willBe("s2"));
- assertThat(fut1, willBe("s2"));
- assertThat(fut2, willBe("s2"));
+ assertThat(execution1.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ assertThat(execution2.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
}
}
@@ -230,7 +236,7 @@ public class ClientComputeTest extends BaseIgniteAbstractTest {
private static CompletableFuture<String> getUnits(IgniteClient client, List<DeploymentUnit> units) {
return client.compute()
- .<String>executeAsync(getClusterNodes("s1"), units, FakeCompute.GET_UNITS, new Object[]{}).resultAsync();
+ .<String>executeAsync(getClusterNodes("s1"), units, FakeCompute.GET_UNITS).resultAsync();
}
@Test
@@ -240,12 +246,10 @@ public class ClientComputeTest extends BaseIgniteAbstractTest {
try (var client = getClient(server1)) {
FakeCompute.future = CompletableFuture.failedFuture(new RuntimeException("job failed"));
- CompletableFuture<Object> fut = client.compute().executeAsync(getClusterNodes("s1"), List.of(), "job").resultAsync();
+ JobExecution<String> execution = client.compute().executeAsync(getClusterNodes("s1"), List.of(), "job");
- assertThat(
- fut,
- willThrowFast(IgniteException.class)
- );
+ assertThat(execution.resultAsync(), willThrowFast(IgniteException.class));
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(FAILED)));
}
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java b/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java
index c438797282..e767369599 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java
@@ -42,9 +42,9 @@ import org.apache.ignite.client.handler.ClientInboundMessageHandler;
import org.apache.ignite.client.handler.ClientPrimaryReplicaTracker;
import org.apache.ignite.client.handler.FakeCatalogService;
import org.apache.ignite.client.handler.configuration.ClientConnectorConfiguration;
-import org.apache.ignite.compute.IgniteCompute;
import org.apache.ignite.internal.catalog.CatalogService;
import org.apache.ignite.internal.client.proto.ClientMessageDecoder;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
import org.apache.ignite.internal.configuration.ConfigurationRegistry;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.manager.IgniteComponent;
@@ -79,7 +79,7 @@ public class TestClientHandlerModule implements IgniteComponent {
private final ClusterService clusterService;
/** Compute. */
- private final IgniteCompute compute;
+ private final IgniteComputeInternal compute;
/** Cluster id. */
private final UUID clusterId;
@@ -125,7 +125,7 @@ public class TestClientHandlerModule implements IgniteComponent {
Function<Integer, Boolean> shouldDropConnection,
@Nullable Function<Integer, Integer> responseDelay,
ClusterService clusterService,
- IgniteCompute compute,
+ IgniteComputeInternal compute,
UUID clusterId,
ClientHandlerMetricSource metrics,
AuthenticationManager authenticationManager,
diff --git a/modules/client/src/test/java/org/apache/ignite/client/TestServer.java b/modules/client/src/test/java/org/apache/ignite/client/TestServer.java
index 32c376a0cd..02433a632b 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/TestServer.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/TestServer.java
@@ -43,8 +43,8 @@ import org.apache.ignite.client.handler.DummyAuthenticationManager;
import org.apache.ignite.client.handler.FakeCatalogService;
import org.apache.ignite.client.handler.FakePlacementDriver;
import org.apache.ignite.client.handler.configuration.ClientConnectorConfiguration;
-import org.apache.ignite.compute.IgniteCompute;
import org.apache.ignite.internal.client.ClientClusterNode;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
import org.apache.ignite.internal.configuration.ConfigurationRegistry;
import org.apache.ignite.internal.configuration.ConfigurationTreeGenerator;
import org.apache.ignite.internal.configuration.storage.TestConfigurationStorage;
@@ -189,7 +189,7 @@ public class TestServer implements AutoCloseable {
Mockito.when(clusterService.topologyService().getByConsistentId(anyString())).thenAnswer(
i -> getClusterNode(i.getArgument(0, String.class)));
- IgniteCompute compute = new FakeCompute(nodeName);
+ IgniteComputeInternal compute = new FakeCompute(nodeName);
metrics = new ClientHandlerMetricSource();
metrics.enable();
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCompute.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCompute.java
index 9f9e195f6e..28d2bc00bf 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCompute.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeCompute.java
@@ -18,12 +18,17 @@
package org.apache.ignite.client.fakes;
import static java.util.concurrent.CompletableFuture.completedFuture;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+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.util.CompletableFutures.trueCompletedFuture;
+import java.time.Instant;
import java.util.List;
import java.util.Map;
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.CountDownLatch;
@@ -31,7 +36,9 @@ import java.util.stream.Collectors;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.IgniteCompute;
import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobState;
import org.apache.ignite.compute.JobStatus;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
import org.apache.ignite.internal.util.ExceptionUtils;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.table.Tuple;
@@ -42,13 +49,15 @@ import org.jetbrains.annotations.Nullable;
* Fake {@link IgniteCompute}.
*/
@SuppressWarnings({"unchecked", "rawtypes"})
-public class FakeCompute implements IgniteCompute {
+public class FakeCompute implements IgniteComputeInternal {
public static final String GET_UNITS = "get-units";
public static volatile @Nullable CompletableFuture future;
public static volatile @Nullable RuntimeException err;
+ public static volatile @Nullable JobStatus status;
+
public static volatile CountDownLatch latch = new CountDownLatch(0);
private final String nodeName;
@@ -59,6 +68,8 @@ public class FakeCompute implements IgniteCompute {
@Override
public <R> JobExecution<R> executeAsync(Set<ClusterNode> nodes, List<DeploymentUnit> units, String jobClassName, Object... args) {
+ initStatus();
+
if (Objects.equals(jobClassName, GET_UNITS)) {
String unitString = units.stream().map(DeploymentUnit::render).collect(Collectors.joining(","));
return completedExecution((R) unitString);
@@ -100,6 +111,7 @@ public class FakeCompute implements IgniteCompute {
String jobClassName,
Object... args
) {
+ initStatus();
return jobExecution(future != null ? future : completedFuture((R) nodeName));
}
@@ -112,6 +124,7 @@ public class FakeCompute implements IgniteCompute {
String jobClassName,
Object... args
) {
+ initStatus();
return jobExecution(future != null ? future : completedFuture((R) nodeName));
}
@@ -162,7 +175,20 @@ public class FakeCompute implements IgniteCompute {
return jobExecution(completedFuture(result));
}
+ private static void initStatus() {
+ status = JobStatus.builder()
+ .id(UUID.randomUUID())
+ .state(EXECUTING)
+ .createTime(Instant.now())
+ .startTime(Instant.now())
+ .build();
+ }
+
private static <R> JobExecution<R> jobExecution(CompletableFuture<R> result) {
+ result.whenComplete((r, throwable) -> {
+ JobState state = throwable != null ? FAILED : COMPLETED;
+ status = status.toBuilder().state(state).finishTime(Instant.now()).build();
+ });
return new JobExecution<>() {
@Override
public CompletableFuture<R> resultAsync() {
@@ -170,19 +196,34 @@ public class FakeCompute implements IgniteCompute {
}
@Override
- public @Nullable CompletableFuture<JobStatus> statusAsync() {
- return nullCompletedFuture();
+ public CompletableFuture<@Nullable JobStatus> statusAsync() {
+ return completedFuture(status);
}
@Override
- public CompletableFuture<Void> cancelAsync() {
- return nullCompletedFuture();
+ public CompletableFuture<@Nullable Boolean> cancelAsync() {
+ return trueCompletedFuture();
}
@Override
- public CompletableFuture<Void> changePriorityAsync(int newPriority) {
- return nullCompletedFuture();
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) {
+ return trueCompletedFuture();
}
};
}
+
+ @Override
+ public CompletableFuture<@Nullable JobStatus> statusAsync(UUID jobId) {
+ return completedFuture(status);
+ }
+
+ @Override
+ public CompletableFuture<@Nullable Boolean> cancelAsync(UUID jobId) {
+ return trueCompletedFuture();
+ }
+
+ @Override
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(UUID jobId, int newPriority) {
+ return trueCompletedFuture();
+ }
}
diff --git a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeBaseTest.java b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeBaseTest.java
index 6d7cada36a..abf492e3e8 100644
--- a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeBaseTest.java
+++ b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeBaseTest.java
@@ -21,7 +21,6 @@ import static java.util.stream.Collectors.toList;
import static org.apache.ignite.compute.JobState.COMPLETED;
import static org.apache.ignite.compute.JobState.FAILED;
import static org.apache.ignite.internal.compute.utils.ComputeTestUtils.assertPublicException;
-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.apache.ignite.lang.ErrorGroups.Common.COMMON_ERR_GROUP;
@@ -48,7 +47,6 @@ import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
import org.apache.ignite.internal.app.IgniteImpl;
-import org.apache.ignite.internal.compute.queue.CancellingException;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.lang.TableNotFoundException;
import org.apache.ignite.network.ClusterNode;
@@ -87,7 +85,7 @@ public abstract class ItComputeBaseTest extends ClusterPerTestIntegrationTest {
assertThat(execution.resultAsync(), willBe("a42"));
assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
}
@Test
@@ -109,7 +107,7 @@ public abstract class ItComputeBaseTest extends ClusterPerTestIntegrationTest {
assertThat(execution.resultAsync(), willBe("a42"));
assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
}
@Test
@@ -157,7 +155,7 @@ public abstract class ItComputeBaseTest extends ClusterPerTestIntegrationTest {
assertPublicException(ex, COMMON_ERR_GROUP, INTERNAL_ERR, "Oops");
assertThat(execution.statusAsync(), willBe(jobStatusWithState(FAILED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
}
@Test
@@ -182,7 +180,7 @@ public abstract class ItComputeBaseTest extends ClusterPerTestIntegrationTest {
assertPublicException(ex, COMMON_ERR_GROUP, INTERNAL_ERR, "Oops");
assertThat(execution.statusAsync(), willBe(jobStatusWithState(FAILED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
}
@Test
@@ -198,7 +196,7 @@ public abstract class ItComputeBaseTest extends ClusterPerTestIntegrationTest {
JobExecution<String> execution = results.get(node);
assertThat(execution.resultAsync(), willBe("a42"));
assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
}
}
@@ -215,7 +213,7 @@ public abstract class ItComputeBaseTest extends ClusterPerTestIntegrationTest {
JobExecution<String> execution = results.get(node);
assertThat(execution.resultAsync(), willBe(node.name()));
assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
}
}
@@ -237,7 +235,7 @@ public abstract class ItComputeBaseTest extends ClusterPerTestIntegrationTest {
assertPublicException(result, COMMON_ERR_GROUP, INTERNAL_ERR, "Oops");
assertThat(execution.statusAsync(), willBe(jobStatusWithState(FAILED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
}
}
@@ -264,7 +262,7 @@ public abstract class ItComputeBaseTest extends ClusterPerTestIntegrationTest {
assertThat(execution.resultAsync(), willBe(in(allNodeNames())));
assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
}
@Test
@@ -314,6 +312,6 @@ public abstract class ItComputeBaseTest extends ClusterPerTestIntegrationTest {
assertThat(execution.resultAsync(), willBe(in(allNodeNames())));
assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
}
}
diff --git a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeTestEmbedded.java b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeTestEmbedded.java
index 2ce4432526..783942ab4a 100644
--- a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeTestEmbedded.java
+++ b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeTestEmbedded.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.compute;
import static java.util.stream.Collectors.joining;
import static org.apache.ignite.internal.compute.utils.ComputeTestUtils.assertPublicException;
-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.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.testframework.matchers.JobStatusMatcher.jobStatusWithState;
@@ -132,7 +131,7 @@ class ItComputeTestEmbedded extends ItComputeBaseTest {
await().until(execution::statusAsync, willBe(jobStatusWithState(JobState.EXECUTING)));
- assertThat(execution.cancelAsync(), willCompleteSuccessfully());
+ assertThat(execution.cancelAsync(), willBe(true));
await().until(execution::statusAsync, willBe(jobStatusWithState(JobState.CANCELED)));
}
@@ -145,29 +144,29 @@ class ItComputeTestEmbedded extends ItComputeBaseTest {
await().until(execution::statusAsync, willBe(jobStatusWithState(JobState.EXECUTING)));
- assertThat(execution.cancelAsync(), willCompleteSuccessfully());
+ assertThat(execution.cancelAsync(), willBe(true));
await().until(execution::statusAsync, willBe(jobStatusWithState(JobState.CANCELED)));
}
@Test
- void changeJobPriorityLocallyComputeException() {
+ void changeExecutingJobPriorityLocally() {
IgniteImpl entryNode = node(0);
JobExecution<String> execution = entryNode.compute().executeAsync(Set.of(entryNode.node()), units(), LongJob.class.getName());
await().until(execution::statusAsync, willBe(jobStatusWithState(JobState.EXECUTING)));
- assertThat(execution.changePriorityAsync(2), willThrow(ComputeException.class));
+ assertThat(execution.changePriorityAsync(2), willBe(false));
}
@Test
- void changeJobPriorityRemotelyComputeException() {
+ void changeExecutingJobPriorityRemotely() {
IgniteImpl entryNode = node(0);
JobExecution<String> execution = entryNode.compute().executeAsync(Set.of(node(1).node()), units(), LongJob.class.getName());
await().until(execution::statusAsync, willBe(jobStatusWithState(JobState.EXECUTING)));
- assertThat(execution.changePriorityAsync(2), willThrow(ComputeException.class));
+ assertThat(execution.changePriorityAsync(2), willBe(false));
}
@Test
@@ -191,7 +190,7 @@ class ItComputeTestEmbedded extends ItComputeBaseTest {
assertThat(execution3.resultAsync().isDone(), is(false));
// Change priority of task 3, so it should be executed before task 2
- assertThat(execution3.changePriorityAsync(2), willCompleteSuccessfully());
+ assertThat(execution3.changePriorityAsync(2), willBe(true));
// Run 1 and 3 task
WaitLatchJob.latch.countDown();
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 307ae2c627..b1a2496393 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
@@ -23,6 +23,8 @@ 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;
@@ -366,12 +368,12 @@ class ItWorkerShutdownTest extends ClusterPerTestIntegrationTest {
assertThat(remoteWorkerCandidates, hasItem(failoverWorker));
// When cancel job.
- execution.cancelAsync().get(10, TimeUnit.SECONDS);
+ assertThat(execution.cancelAsync(), willBe(true));
// Then it is cancelled.
assertThat(execution.resultAsync(), willThrow(IgniteException.class));
// And.
- assertThat(statusSync(execution).state(), is(CANCELED));
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(CANCELED)));
}
@Test
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/Cleaner.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/Cleaner.java
new file mode 100644
index 0000000000..65315d7532
--- /dev/null
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/Cleaner.java
@@ -0,0 +1,91 @@
+/*
+ * 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 org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
+
+import java.util.HashSet;
+import java.util.Queue;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.thread.NamedThreadFactory;
+
+/**
+ * Cleans entries after TTL is expired.
+ *
+ * @param <T> Type of entry.
+ */
+public class Cleaner<T> {
+ private static final IgniteLogger LOG = Loggers.forClass(Cleaner.class);
+
+ private ExecutorService cleaner;
+
+ private final Set<UUID> toRemove = new HashSet<>();
+
+ private final Queue<UUID> waitToRemove = new ConcurrentLinkedQueue<>();
+
+ /**
+ * Starts the cleaner.
+ *
+ * @param clean Function to clean the entry.
+ * @param ttlMillis Time after which the clean function will be called for the scheduled entry, in milliseconds.
+ * @param nodeName Node name.
+ */
+ public void start(Consumer<UUID> clean, long ttlMillis, String nodeName) {
+ ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
+ NamedThreadFactory.create(nodeName, "compute-execution-cleanup", true, LOG)
+ );
+
+ executor.scheduleAtFixedRate(() -> {
+ toRemove.forEach(clean);
+ toRemove.clear();
+
+ UUID nextToRemove;
+ //noinspection NestedAssignment
+ while ((nextToRemove = waitToRemove.poll()) != null) {
+ toRemove.add(nextToRemove);
+ }
+ }, ttlMillis, ttlMillis, TimeUnit.MILLISECONDS);
+
+ cleaner = executor;
+ }
+
+ /**
+ * Stops the cleaner.
+ */
+ public void stop() {
+ shutdownAndAwaitTermination(cleaner, 1000, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Schedules entry to remove.
+ *
+ * @param entryId Entry id.
+ */
+ public void scheduleRemove(UUID entryId) {
+ waitToRemove.add(entryId);
+ }
+}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponent.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponent.java
index 4313f83a32..22a0737a6a 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponent.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponent.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.compute;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.JobStatus;
@@ -103,11 +104,30 @@ public interface ComputeComponent extends IgniteComponent {
}
/**
- * Returns job status by ID.
+ * Retrieves the current status of the job on any node in the cluster. The job status may be deleted and thus return {@code null} if the
+ * time for retaining job status has been exceeded.
*
- * @param jobId Job ID.
- * @return Job status. {@code null} if job with the given ID does not exist.
+ * @param jobId Job id.
+ * @return The current status of the job, or {@code null} if the job status no longer exists due to exceeding the retention time limit.
*/
- @Nullable
- JobStatus getJobStatus(UUID jobId);
+ CompletableFuture<@Nullable JobStatus> statusAsync(UUID jobId);
+
+ /**
+ * Cancels the job running on any node in the cluster.
+ *
+ * @param jobId Job id.
+ * @return The future which will be completed with {@code true} when the job is cancelled, {@code false} when the job couldn't be
+ * cancelled (either it's not yet started, or it's already completed), or {@code null} if there's no job with the specified id.
+ */
+ CompletableFuture<@Nullable Boolean> cancelAsync(UUID jobId);
+
+ /**
+ * Changes compute job priority.
+ *
+ * @param jobId Job id.
+ * @param newPriority New priority.
+ * @return The future which will be completed with {@code true} when the priority is changed, {@code false} when the priority couldn't
+ * be changed (it's already executing or completed), or {@code null} if there's no job with the specified id.
+ */
+ CompletableFuture<@Nullable Boolean> changePriorityAsync(UUID jobId, int newPriority);
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponentImpl.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponentImpl.java
index db297e6ff9..1dc4598500 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponentImpl.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeComponentImpl.java
@@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.JobStatus;
+import org.apache.ignite.internal.compute.configuration.ComputeConfiguration;
import org.apache.ignite.internal.compute.executor.ComputeExecutor;
import org.apache.ignite.internal.compute.executor.JobExecutionInternal;
import org.apache.ignite.internal.compute.loader.JobContext;
@@ -41,6 +42,7 @@ import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.TopologyService;
import org.jetbrains.annotations.Nullable;
/**
@@ -61,17 +63,22 @@ public class ComputeComponentImpl implements ComputeComponent {
private final ComputeMessaging messaging;
+ private final ExecutionManager executionManager;
+
/**
* Creates a new instance.
*/
public ComputeComponentImpl(
MessagingService messagingService,
+ TopologyService topologyService,
JobContextManager jobContextManager,
- ComputeExecutor executor
+ ComputeExecutor executor,
+ ComputeConfiguration computeConfiguration
) {
this.jobContextManager = jobContextManager;
this.executor = executor;
- messaging = new ComputeMessaging(messagingService);
+ executionManager = new ExecutionManager(computeConfiguration, topologyService);
+ messaging = new ComputeMessaging(executionManager, messagingService, topologyService);
}
/** {@inheritDoc} */
@@ -82,7 +89,30 @@ public class ComputeComponentImpl implements ComputeComponent {
String jobClassName,
Object... args
) {
- return start(options, units, jobClassName, args);
+ if (!busyLock.enterBusy()) {
+ return new DelegatingJobExecution<>(
+ failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()))
+ );
+ }
+
+ try {
+ CompletableFuture<JobExecutionInternal<R>> future =
+ mapClassLoaderExceptions(jobContextManager.acquireClassLoader(units), jobClassName)
+ .thenApply(context -> {
+ JobExecutionInternal<R> execution = exec(context, options, jobClassName, args);
+ execution.resultAsync().whenComplete((result, e) -> context.close());
+ inFlightFutures.registerFuture(execution.resultAsync());
+ return execution;
+ });
+
+ inFlightFutures.registerFuture(future);
+
+ JobExecution<R> result = new DelegatingJobExecution<>(future);
+ result.idAsync().thenAccept(jobId -> executionManager.addExecution(jobId, result));
+ return result;
+ } finally {
+ busyLock.leaveBusy();
+ }
}
/** {@inheritDoc} */
@@ -107,52 +137,35 @@ public class ComputeComponentImpl implements ComputeComponent {
inFlightFutures.registerFuture(jobIdFuture);
inFlightFutures.registerFuture(resultFuture);
- return new RemoteJobExecution<>(remoteNode, jobIdFuture, resultFuture, inFlightFutures, messaging);
+ JobExecution<R> result = new RemoteJobExecution<>(remoteNode, jobIdFuture, resultFuture, inFlightFutures, messaging);
+ jobIdFuture.thenAccept(jobId -> executionManager.addExecution(jobId, result));
+ return result;
} finally {
busyLock.leaveBusy();
}
}
@Override
- public @Nullable JobStatus getJobStatus(UUID jobId) {
- return executor.status(jobId);
+ public CompletableFuture<@Nullable JobStatus> statusAsync(UUID jobId) {
+ return messaging.broadcastStatusAsync(jobId);
}
- private <R> JobExecution<R> start(
- ExecutionOptions options,
- List<DeploymentUnit> units,
- String jobClassName,
- Object... args
- ) {
- if (!busyLock.enterBusy()) {
- return new DelegatingJobExecution<>(
- failedFuture(new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException()))
- );
- }
-
- try {
- CompletableFuture<JobExecutionInternal<R>> future =
- mapClassLoaderExceptions(jobContextManager.acquireClassLoader(units), jobClassName)
- .thenApply(context -> {
- JobExecutionInternal<R> execution = exec(context, options, jobClassName, args);
- execution.resultAsync().whenComplete((result, e) -> context.close());
- inFlightFutures.registerFuture(execution.resultAsync());
- return execution;
- });
-
- inFlightFutures.registerFuture(future);
+ @Override
+ public CompletableFuture<@Nullable Boolean> cancelAsync(UUID jobId) {
+ return messaging.broadcastCancelAsync(jobId);
+ }
- return new DelegatingJobExecution<>(future);
- } finally {
- busyLock.leaveBusy();
- }
+ @Override
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(UUID jobId, int newPriority) {
+ return messaging.broadcastChangePriorityAsync(jobId, newPriority);
}
/** {@inheritDoc} */
@Override
public CompletableFuture<Void> start() {
executor.start();
- messaging.start(this::start, this::getJobStatus);
+ messaging.start(this::executeLocally);
+ executionManager.start();
return nullCompletedFuture();
}
@@ -167,6 +180,7 @@ public class ComputeComponentImpl implements ComputeComponent {
busyLock.block();
inFlightFutures.cancelInFlightFutures();
+ executionManager.stop();
messaging.stop();
executor.stop();
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeUtils.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeUtils.java
index 990d2f9456..5d1f62d420 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeUtils.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ComputeUtils.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.compute;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.lang.ErrorGroups.Compute.CLASS_INITIALIZATION_ERR;
import java.lang.reflect.Constructor;
@@ -38,6 +37,7 @@ import org.apache.ignite.internal.compute.message.JobCancelResponse;
import org.apache.ignite.internal.compute.message.JobChangePriorityResponse;
import org.apache.ignite.internal.compute.message.JobResultResponse;
import org.apache.ignite.internal.compute.message.JobStatusResponse;
+import org.jetbrains.annotations.Nullable;
/**
* Utility class for compute.
@@ -147,7 +147,7 @@ public class ComputeUtils {
* @param jobStatusResponse Job status result message response.
* @return Completable future with result.
*/
- public static CompletableFuture<JobStatus> statusFromJobStatusResponse(JobStatusResponse jobStatusResponse) {
+ public static CompletableFuture<@Nullable JobStatus> statusFromJobStatusResponse(JobStatusResponse jobStatusResponse) {
Throwable throwable = jobStatusResponse.throwable();
if (throwable != null) {
return failedFuture(throwable);
@@ -162,13 +162,13 @@ public class ComputeUtils {
* @param jobCancelResponse Job cancel message response.
* @return Completable future with result.
*/
- public static CompletableFuture<Void> cancelFromJobCancelResponse(JobCancelResponse jobCancelResponse) {
+ public static CompletableFuture<@Nullable Boolean> cancelFromJobCancelResponse(JobCancelResponse jobCancelResponse) {
Throwable throwable = jobCancelResponse.throwable();
if (throwable != null) {
return failedFuture(throwable);
}
- return nullCompletedFuture();
+ return completedFuture(jobCancelResponse.result());
}
/**
@@ -177,13 +177,15 @@ public class ComputeUtils {
* @param jobChangePriorityResponse Job change priority message response.
* @return Completable future with result.
*/
- public static CompletableFuture<Void> changePriorityFromJobChangePriorityResponse(JobChangePriorityResponse jobChangePriorityResponse) {
+ public static CompletableFuture<@Nullable Boolean> changePriorityFromJobChangePriorityResponse(
+ JobChangePriorityResponse jobChangePriorityResponse
+ ) {
Throwable throwable = jobChangePriorityResponse.throwable();
if (throwable != null) {
return failedFuture(throwable);
}
- return nullCompletedFuture();
+ return completedFuture(jobChangePriorityResponse.result());
}
/**
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/DelegatingJobExecution.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/DelegatingJobExecution.java
index bc1bdeedf3..32b45d1ddd 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/DelegatingJobExecution.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/DelegatingJobExecution.java
@@ -21,6 +21,7 @@ import java.util.concurrent.CompletableFuture;
import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.JobStatus;
import org.apache.ignite.internal.compute.executor.JobExecutionInternal;
+import org.jetbrains.annotations.Nullable;
/**
* Delegates {@link JobExecution} to the future of {@link JobExecutionInternal}.
@@ -40,17 +41,17 @@ class DelegatingJobExecution<R> implements JobExecution<R> {
}
@Override
- public CompletableFuture<JobStatus> statusAsync() {
+ public CompletableFuture<@Nullable JobStatus> statusAsync() {
return delegate.thenApply(JobExecutionInternal::status);
}
@Override
- public CompletableFuture<Void> cancelAsync() {
- return delegate.thenAccept(JobExecutionInternal::cancel);
+ public CompletableFuture<@Nullable Boolean> cancelAsync() {
+ return delegate.thenApply(JobExecutionInternal::cancel);
}
@Override
- public CompletableFuture<Void> changePriorityAsync(int newPriority) {
- return delegate.thenAccept(jobExecutionInternal -> jobExecutionInternal.changePriority(newPriority));
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) {
+ return delegate.thenApply(jobExecutionInternal -> jobExecutionInternal.changePriority(newPriority));
}
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/ExecutionManager.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ExecutionManager.java
new file mode 100644
index 0000000000..4a7d20423f
--- /dev/null
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/ExecutionManager.java
@@ -0,0 +1,136 @@
+/*
+ * 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.concurrent.CompletableFuture.failedFuture;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.lang.ErrorGroups.Compute.RESULT_NOT_FOUND_ERR;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.compute.ComputeException;
+import org.apache.ignite.compute.JobExecution;
+import org.apache.ignite.compute.JobStatus;
+import org.apache.ignite.internal.compute.configuration.ComputeConfiguration;
+import org.apache.ignite.network.TopologyService;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Manages job executions. Stores them in the TTL-based cache.
+ */
+public class ExecutionManager {
+ private final ComputeConfiguration computeConfiguration;
+
+ private final TopologyService topologyService;
+
+ private final Cleaner<JobExecution<?>> cleaner = new Cleaner<>();
+
+ private final Map<UUID, JobExecution<?>> executions = new ConcurrentHashMap<>();
+
+ ExecutionManager(ComputeConfiguration computeConfiguration, TopologyService topologyService) {
+ this.computeConfiguration = computeConfiguration;
+ this.topologyService = topologyService;
+ }
+
+ /**
+ * Puts an execution to the cache. When the job completes, it will be evicted from the cache after some time.
+ *
+ * @param jobId Job id.
+ * @param execution Job execution.
+ */
+ void addExecution(UUID jobId, JobExecution<?> execution) {
+ executions.put(jobId, execution);
+ execution.resultAsync().whenComplete((r, throwable) -> cleaner.scheduleRemove(jobId));
+ }
+
+ /**
+ * Starts execution manager.
+ */
+ void start() {
+ long ttlMillis = computeConfiguration.statesLifetimeMillis().value();
+ String nodeName = topologyService.localMember().name();
+ cleaner.start(executions::remove, ttlMillis, nodeName);
+ }
+
+ /**
+ * Stops execution manager.
+ */
+ void stop() {
+ cleaner.stop();
+ }
+
+ /**
+ * Returns job's execution result.
+ *
+ * @param jobId Job id.
+ * @return Job's execution result future.
+ */
+ public CompletableFuture<?> resultAsync(UUID jobId) {
+ JobExecution<?> execution = executions.get(jobId);
+ if (execution != null) {
+ return execution.resultAsync();
+ }
+ return failedFuture(new ComputeException(RESULT_NOT_FOUND_ERR, "Job result not found for the job with ID: " + jobId));
+ }
+
+ /**
+ * Retrieves the current status of the job.
+ *
+ * @param jobId Job id.
+ * @return The current status of the job, or {@code null} if there's no job with the specified id.
+ */
+ public CompletableFuture<@Nullable JobStatus> statusAsync(UUID jobId) {
+ JobExecution<?> execution = executions.get(jobId);
+ if (execution != null) {
+ return execution.statusAsync();
+ }
+ return nullCompletedFuture();
+ }
+
+ /**
+ * Cancels the running job.
+ *
+ * @param jobId Job id.
+ * @return The future which will be completed with {@code true} when the job is cancelled, {@code false} when the job couldn't be
+ * cancelled (either it's not yet started, or it's already completed), or {@code null} if there's no job with the specified id.
+ */
+ public CompletableFuture<@Nullable Boolean> cancelAsync(UUID jobId) {
+ JobExecution<?> execution = executions.get(jobId);
+ if (execution != null) {
+ return execution.cancelAsync();
+ }
+ return nullCompletedFuture();
+ }
+
+ /**
+ * Changes job priority.
+ *
+ * @param jobId Job id.
+ * @return The future which will be completed with {@code true} when the priority is changed, {@code false} when the priority couldn't
+ * be changed (it's already executing or completed), or {@code null} iif there's no job with the specified id.
+ */
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(UUID jobId, int newPriority) {
+ JobExecution<?> execution = executions.get(jobId);
+ if (execution != null) {
+ return execution.changePriorityAsync(newPriority);
+ }
+ return nullCompletedFuture();
+ }
+}
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 b2f16adc3f..46e6e31b1c 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
@@ -29,6 +29,7 @@ import org.apache.ignite.compute.JobState;
import org.apache.ignite.compute.JobStatus;
import org.apache.ignite.internal.logger.IgniteLogger;
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
@@ -137,13 +138,13 @@ class FailSafeJobExecution<T> implements JobExecution<T> {
}
/**
- * Returns the transformed status of the running job execution. The transformation is needed because we do not want to change
- * some fields of the status (e.g. creation time) when the job is restarted.
+ * Returns the transformed status of the running job execution. The transformation is needed because we do not want to change some
+ * fields of the status (e.g. creation time) when the job is restarted.
*
* @return the transformed status.
*/
@Override
- public CompletableFuture<JobStatus> statusAsync() {
+ public CompletableFuture<@Nullable JobStatus> statusAsync() {
if (exception.get() != null) {
return CompletableFuture.failedFuture(exception.get());
}
@@ -154,13 +155,13 @@ class FailSafeJobExecution<T> implements JobExecution<T> {
}
@Override
- public CompletableFuture<Void> cancelAsync() {
+ public CompletableFuture<@Nullable Boolean> cancelAsync() {
resultFuture.cancel(false);
return runningJobExecution.get().cancelAsync();
}
@Override
- public CompletableFuture<Void> changePriorityAsync(int newPriority) {
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) {
return runningJobExecution.get().changePriorityAsync(newPriority);
}
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 5ebb6991ee..19a70a5c2b 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
@@ -26,12 +26,14 @@ import java.util.List;
import java.util.Map;
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.ThreadLocalRandom;
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.table.IgniteTablesInternal;
import org.apache.ignite.internal.table.TableViewInternal;
@@ -43,11 +45,12 @@ 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;
/**
* Implementation of {@link IgniteCompute}.
*/
-public class IgniteComputeImpl implements IgniteCompute {
+public class IgniteComputeImpl implements IgniteComputeInternal {
private static final String DEFAULT_SCHEMA_NAME = "PUBLIC";
private final TopologyService topologyService;
@@ -271,4 +274,19 @@ public class IgniteComputeImpl implements IgniteCompute {
// if worker node has left the cluster.
node -> new JobExecutionWrapper<>(executeOnOneNodeWithFailover(node, Set.of(), units, jobClassName, args))));
}
+
+ @Override
+ public CompletableFuture<@Nullable JobStatus> statusAsync(UUID jobId) {
+ return computeComponent.statusAsync(jobId);
+ }
+
+ @Override
+ public CompletableFuture<@Nullable Boolean> cancelAsync(UUID jobId) {
+ return computeComponent.cancelAsync(jobId);
+ }
+
+ @Override
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(UUID jobId, int newPriority) {
+ return computeComponent.changePriorityAsync(jobId, newPriority);
+ }
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeInternal.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeInternal.java
new file mode 100644
index 0000000000..417e07c236
--- /dev/null
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeInternal.java
@@ -0,0 +1,56 @@
+/*
+ * 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.UUID;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.compute.IgniteCompute;
+import org.apache.ignite.compute.JobStatus;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Internal compute facade.
+ */
+public interface IgniteComputeInternal extends IgniteCompute {
+ /**
+ * Gets job status by id.
+ *
+ * @param jobId Job id.
+ * @return Job status or {@code null} if there's no status registered for this id.
+ */
+ CompletableFuture<@Nullable JobStatus> statusAsync(UUID jobId);
+
+ /**
+ * Cancels compute job.
+ *
+ * @param jobId Job id.
+ * @return The future which will be completed with {@code true} when the job is cancelled, {@code false} when the job couldn't be
+ * cancelled (either it's not yet started, or it's already completed), or {@code null} if there's no job with the specified id.
+ */
+ CompletableFuture<@Nullable Boolean> cancelAsync(UUID jobId);
+
+ /**
+ * Changes compute job priority.
+ *
+ * @param jobId Job id.
+ * @param newPriority New priority.
+ * @return The future which will be completed with {@code true} when the priority is changed, {@code false} when the priority couldn't
+ * be changed (it's already executing or completed), or {@code null} if there's no job with the specified id.
+ */
+ CompletableFuture<@Nullable Boolean> changePriorityAsync(UUID jobId, int newPriority);
+}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionFutureWrapper.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionFutureWrapper.java
index 95ed3ce85f..fa74709eaf 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionFutureWrapper.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionFutureWrapper.java
@@ -22,6 +22,7 @@ import static org.apache.ignite.internal.lang.IgniteExceptionMapperUtil.convertT
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.JobStatus;
+import org.jetbrains.annotations.Nullable;
/**
* Wraps the future of {@link JobExecution} converting exceptions thrown by the delegate to public.
@@ -41,17 +42,17 @@ class JobExecutionFutureWrapper<R> implements JobExecution<R> {
}
@Override
- public CompletableFuture<JobStatus> statusAsync() {
+ public CompletableFuture<@Nullable JobStatus> statusAsync() {
return convertToPublicFuture(delegate.thenCompose(JobExecution::statusAsync));
}
@Override
- public CompletableFuture<Void> cancelAsync() {
+ public CompletableFuture<@Nullable Boolean> cancelAsync() {
return convertToPublicFuture(delegate.thenCompose(JobExecution::cancelAsync));
}
@Override
- public CompletableFuture<Void> changePriorityAsync(int newPriority) {
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) {
return convertToPublicFuture(delegate.thenCompose(jobExecution -> jobExecution.changePriorityAsync(newPriority)));
}
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionWrapper.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionWrapper.java
index 46a59888c6..24127cb16a 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionWrapper.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/JobExecutionWrapper.java
@@ -22,6 +22,7 @@ import static org.apache.ignite.internal.lang.IgniteExceptionMapperUtil.convertT
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.JobStatus;
+import org.jetbrains.annotations.Nullable;
/**
* Wraps the {@link JobExecution} converting exceptions thrown by the delegate to public.
@@ -41,17 +42,17 @@ class JobExecutionWrapper<R> implements JobExecution<R> {
}
@Override
- public CompletableFuture<JobStatus> statusAsync() {
+ public CompletableFuture<@Nullable JobStatus> statusAsync() {
return convertToPublicFuture(delegate.statusAsync());
}
@Override
- public CompletableFuture<Void> cancelAsync() {
+ public CompletableFuture<@Nullable Boolean> cancelAsync() {
return convertToPublicFuture(delegate.cancelAsync());
}
@Override
- public CompletableFuture<Void> changePriorityAsync(int newPriority) {
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) {
return convertToPublicFuture(delegate.changePriorityAsync(newPriority));
}
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutor.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutor.java
index 6294168ab8..bf4866ad4a 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutor.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutor.java
@@ -17,11 +17,8 @@
package org.apache.ignite.internal.compute.executor;
-import java.util.UUID;
import org.apache.ignite.compute.ComputeJob;
-import org.apache.ignite.compute.JobStatus;
import org.apache.ignite.internal.compute.ExecutionOptions;
-import org.jetbrains.annotations.Nullable;
/**
* Executor of Compute jobs.
@@ -31,8 +28,5 @@ public interface ComputeExecutor {
void start();
- @Nullable
- JobStatus status(UUID jobId);
-
void stop();
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutorImpl.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutorImpl.java
index 30cdc8486f..8cdf3c2195 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutorImpl.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/ComputeExecutorImpl.java
@@ -17,12 +17,10 @@
package org.apache.ignite.internal.compute.executor;
-import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.Ignite;
import org.apache.ignite.compute.ComputeJob;
import org.apache.ignite.compute.JobExecutionContext;
-import org.apache.ignite.compute.JobStatus;
import org.apache.ignite.internal.compute.ComputeUtils;
import org.apache.ignite.internal.compute.ExecutionOptions;
import org.apache.ignite.internal.compute.JobExecutionContextImpl;
@@ -33,7 +31,6 @@ import org.apache.ignite.internal.compute.state.ComputeStateMachine;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.thread.NamedThreadFactory;
-import org.jetbrains.annotations.Nullable;
/**
* Base implementation of {@link ComputeExecutor}.
@@ -91,17 +88,11 @@ public class ComputeExecutorImpl implements ComputeExecutor {
stateMachine.start();
executorService = new PriorityQueueExecutor(
configuration,
- new NamedThreadFactory(NamedThreadFactory.threadPrefix(ignite.name(), "compute"), LOG),
+ NamedThreadFactory.create(ignite.name(), "compute", LOG),
stateMachine
);
}
- @Override
- @Nullable
- public JobStatus status(UUID jobId) {
- return stateMachine.currentStatus(jobId);
- }
-
@Override
public void stop() {
stateMachine.stop();
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/JobExecutionInternal.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/JobExecutionInternal.java
index b7c9b8398d..ce8c026351 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/JobExecutionInternal.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/executor/JobExecutionInternal.java
@@ -53,17 +53,23 @@ public class JobExecutionInternal<R> {
return execution.status();
}
- public void cancel() {
+ /**
+ * Cancel job execution.
+ *
+ * @return {@code true} if job was successfully cancelled.
+ */
+ public boolean cancel() {
isInterrupted.set(true);
- execution.cancel();
+ return execution.cancel();
}
/**
* Change priority of job execution.
*
* @param newPriority new priority.
+ * @return {@code true} if priority was successfully changed.
*/
- public void changePriority(int newPriority) {
- execution.changePriority(newPriority);
+ public boolean changePriority(int newPriority) {
+ return execution.changePriority(newPriority);
}
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/message/JobCancelResponse.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/message/JobCancelResponse.java
index 596bd1cae4..30ae9b8cd7 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/message/JobCancelResponse.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/message/JobCancelResponse.java
@@ -36,4 +36,12 @@ public interface JobCancelResponse extends NetworkMessage {
@Nullable
@Marshallable
Throwable throwable();
+
+ /**
+ * Returns {@code true} if the job was cancelled, {@code false} if not and {@code null} if the job was not found.
+ *
+ * @return {@code true} if the job was cancelled, {@code false} if not and {@code null} if the job was not found.
+ */
+ @Nullable
+ Boolean result();
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/message/JobChangePriorityResponse.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/message/JobChangePriorityResponse.java
index fd822849fc..9e894409fe 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/message/JobChangePriorityResponse.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/message/JobChangePriorityResponse.java
@@ -37,4 +37,12 @@ public interface JobChangePriorityResponse extends NetworkMessage {
@Nullable
@Marshallable
Throwable throwable();
+
+ /**
+ * Returns {@code true} if the priority was changed, {@code false} if not and {@code null} if the job was not found.
+ *
+ * @return {@code true} if the priority was changed, {@code false} if not and {@code null} if the job was not found.
+ */
+ @Nullable
+ Boolean result();
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/messaging/ComputeMessaging.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/messaging/ComputeMessaging.java
index cd94e83b8e..cbe1f2285c 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/messaging/ComputeMessaging.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/messaging/ComputeMessaging.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.compute.messaging;
+import static java.util.concurrent.CompletableFuture.allOf;
import static org.apache.ignite.internal.compute.ComputeUtils.cancelFromJobCancelResponse;
import static org.apache.ignite.internal.compute.ComputeUtils.changePriorityFromJobChangePriorityResponse;
import static org.apache.ignite.internal.compute.ComputeUtils.jobIdFromExecuteResponse;
@@ -24,12 +25,13 @@ import static org.apache.ignite.internal.compute.ComputeUtils.resultFromJobResul
import static org.apache.ignite.internal.compute.ComputeUtils.statusFromJobStatusResponse;
import static org.apache.ignite.internal.compute.ComputeUtils.toDeploymentUnit;
import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Compute.CANCELLING_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Compute.CHANGE_JOB_PRIORITY_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Compute.FAIL_TO_GET_JOB_STATUS_ERR;
import java.util.List;
-import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.ignite.compute.ComputeException;
@@ -39,6 +41,7 @@ import org.apache.ignite.compute.JobStatus;
import org.apache.ignite.internal.compute.ComputeMessageTypes;
import org.apache.ignite.internal.compute.ComputeMessagesFactory;
import org.apache.ignite.internal.compute.ComputeUtils;
+import org.apache.ignite.internal.compute.ExecutionManager;
import org.apache.ignite.internal.compute.ExecutionOptions;
import org.apache.ignite.internal.compute.JobStarter;
import org.apache.ignite.internal.compute.message.DeploymentUnitMsg;
@@ -52,14 +55,13 @@ import org.apache.ignite.internal.compute.message.JobResultRequest;
import org.apache.ignite.internal.compute.message.JobResultResponse;
import org.apache.ignite.internal.compute.message.JobStatusRequest;
import org.apache.ignite.internal.compute.message.JobStatusResponse;
-import org.apache.ignite.internal.compute.queue.CancellingException;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
-import org.apache.ignite.lang.ErrorGroups.Compute;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.MessagingService;
import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.TopologyService;
import org.jetbrains.annotations.Nullable;
/**
@@ -70,23 +72,31 @@ public class ComputeMessaging {
private final ComputeMessagesFactory messagesFactory = new ComputeMessagesFactory();
+ private final ExecutionManager executionManager;
+
private final MessagingService messagingService;
- private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+ private final TopologyService topologyService;
- //TODO https://issues.apache.org/jira/browse/IGNITE-21168
- private final Map<UUID, JobExecution<Object>> executions = new ConcurrentHashMap<>();
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
- public ComputeMessaging(MessagingService messagingService) {
+ /**
+ * Constructor.
+ *
+ * @param executionManager Execution manager.
+ * @param messagingService Messaging service.
+ * @param topologyService Topology service.
+ */
+ public ComputeMessaging(ExecutionManager executionManager, MessagingService messagingService, TopologyService topologyService) {
+ this.executionManager = executionManager;
this.messagingService = messagingService;
+ this.topologyService = topologyService;
}
/**
* Start messaging service.
- *
- * @param starter Compute job starter.
*/
- public void start(JobStarter starter, Function<UUID, JobStatus> jobStatus) {
+ public void start(JobStarter starter) {
messagingService.addMessageHandler(ComputeMessageTypes.class, (message, senderConsistentId, correlationId) -> {
assert correlationId != null;
@@ -101,7 +111,7 @@ public class ComputeMessaging {
}
try {
- processRequest(message, senderConsistentId, correlationId, starter, jobStatus);
+ processRequest(message, senderConsistentId, correlationId, starter);
} finally {
busyLock.leaveBusy();
}
@@ -121,9 +131,9 @@ public class ComputeMessaging {
} else if (message instanceof JobStatusRequest) {
sendJobStatusResponse(null, ex, senderConsistentId, correlationId);
} else if (message instanceof JobCancelRequest) {
- sendJobCancelResponse(ex, senderConsistentId, correlationId);
+ sendJobCancelResponse(null, ex, senderConsistentId, correlationId);
} else if (message instanceof JobChangePriorityRequest) {
- sendJobChangePriorityResponse(ex, senderConsistentId, correlationId);
+ sendJobChangePriorityResponse(null, ex, senderConsistentId, correlationId);
}
}
@@ -131,15 +141,14 @@ public class ComputeMessaging {
NetworkMessage message,
String senderConsistentId,
long correlationId,
- JobStarter starter,
- Function<UUID, JobStatus> jobStatus
+ JobStarter starter
) {
if (message instanceof ExecuteRequest) {
processExecuteRequest(starter, (ExecuteRequest) message, senderConsistentId, correlationId);
} else if (message instanceof JobResultRequest) {
processJobResultRequest((JobResultRequest) message, senderConsistentId, correlationId);
} else if (message instanceof JobStatusRequest) {
- processJobStatusRequest(jobStatus, (JobStatusRequest) message, senderConsistentId, correlationId);
+ processJobStatusRequest((JobStatusRequest) message, senderConsistentId, correlationId);
} else if (message instanceof JobCancelRequest) {
processJobCancelRequest((JobCancelRequest) message, senderConsistentId, correlationId);
} else if (message instanceof JobChangePriorityRequest) {
@@ -190,12 +199,7 @@ public class ComputeMessaging {
List<DeploymentUnit> units = toDeploymentUnit(request.deploymentUnits());
JobExecution<Object> execution = starter.start(request.executeOptions(), units, request.jobClassName(), request.args());
- execution.idAsync().whenComplete((jobId, err) -> {
- if (jobId != null) {
- executions.put(jobId, execution);
- }
- sendExecuteResponse(jobId, err, senderConsistentId, correlationId);
- });
+ execution.idAsync().whenComplete((jobId, err) -> sendExecuteResponse(jobId, err, senderConsistentId, correlationId));
}
private void sendExecuteResponse(@Nullable UUID jobId, @Nullable Throwable ex, String senderConsistentId, Long correlationId) {
@@ -225,19 +229,11 @@ public class ComputeMessaging {
}
private void processJobResultRequest(JobResultRequest request, String senderConsistentId, long correlationId) {
- UUID jobId = request.jobId();
- JobExecution<Object> execution = executions.get(jobId);
- if (execution != null) {
- execution.resultAsync()
- .whenComplete((result, err) -> sendJobResultResponse(result, err, senderConsistentId, correlationId)
- .whenComplete((unused, throwable) -> executions.remove(jobId)));
- } else {
- ComputeException ex = new ComputeException(Compute.RESULT_NOT_FOUND_ERR, "Job result not found for the job id " + jobId);
- sendJobResultResponse(null, ex, senderConsistentId, correlationId);
- }
+ executionManager.resultAsync(request.jobId())
+ .whenComplete((result, err) -> sendJobResultResponse(result, err, senderConsistentId, correlationId));
}
- private CompletableFuture<Void> sendJobResultResponse(
+ private void sendJobResultResponse(
@Nullable Object result,
@Nullable Throwable ex,
String senderConsistentId,
@@ -248,7 +244,7 @@ public class ComputeMessaging {
.throwable(ex)
.build();
- return messagingService.respond(senderConsistentId, jobResultResponse, correlationId);
+ messagingService.respond(senderConsistentId, jobResultResponse, correlationId);
}
/**
@@ -256,9 +252,9 @@ public class ComputeMessaging {
*
* @param remoteNode The job will be executed on this node.
* @param jobId Compute job id.
- * @return Job status future.
+ * @return The current status of the job, or {@code null} if there's no job with the specified id.
*/
- CompletableFuture<JobStatus> remoteStatusAsync(ClusterNode remoteNode, UUID jobId) {
+ CompletableFuture<@Nullable JobStatus> remoteStatusAsync(ClusterNode remoteNode, UUID jobId) {
JobStatusRequest jobStatusRequest = messagesFactory.jobStatusRequest()
.jobId(jobId)
.build();
@@ -268,12 +264,12 @@ public class ComputeMessaging {
}
private void processJobStatusRequest(
- Function<UUID, JobStatus> jobStatus,
JobStatusRequest request,
String senderConsistentId,
long correlationId
) {
- sendJobStatusResponse(jobStatus.apply(request.jobId()), null, senderConsistentId, correlationId);
+ executionManager.statusAsync(request.jobId())
+ .whenComplete((status, throwable) -> sendJobStatusResponse(status, throwable, senderConsistentId, correlationId));
}
private void sendJobStatusResponse(
@@ -295,9 +291,10 @@ public class ComputeMessaging {
*
* @param remoteNode The job will be canceled on this node.
* @param jobId Compute job id.
- * @return Job cancel future (will be completed when cancel request is processed).
+ * @return The future which will be completed with {@code true} when the job is cancelled, {@code false} when the job couldn't be
+ * cancelled (either it's not yet started, or it's already completed), or {@code null} if there's no job with the specified id.
*/
- CompletableFuture<Void> remoteCancelAsync(ClusterNode remoteNode, UUID jobId) {
+ CompletableFuture<@Nullable Boolean> remoteCancelAsync(ClusterNode remoteNode, UUID jobId) {
JobCancelRequest jobCancelRequest = messagesFactory.jobCancelRequest()
.jobId(jobId)
.build();
@@ -307,22 +304,18 @@ public class ComputeMessaging {
}
private void processJobCancelRequest(JobCancelRequest request, String senderConsistentId, long correlationId) {
- UUID jobId = request.jobId();
- JobExecution<Object> execution = executions.get(jobId);
- if (execution != null) {
- execution.cancelAsync()
- .whenComplete((result, err) -> sendJobCancelResponse(err, senderConsistentId, correlationId));
- } else {
- sendJobCancelResponse(new CancellingException(jobId), senderConsistentId, correlationId);
- }
+ executionManager.cancelAsync(request.jobId())
+ .whenComplete((result, err) -> sendJobCancelResponse(result, err, senderConsistentId, correlationId));
}
private void sendJobCancelResponse(
+ @Nullable Boolean result,
@Nullable Throwable throwable,
String senderConsistentId,
Long correlationId
) {
JobCancelResponse jobCancelResponse = messagesFactory.jobCancelResponse()
+ .result(result)
.throwable(throwable)
.build();
@@ -338,7 +331,7 @@ public class ComputeMessaging {
*
* @return Job change priority future (will be completed when change priority request is processed).
*/
- CompletableFuture<Void> remoteChangePriorityAsync(ClusterNode remoteNode, UUID jobId, int newPriority) {
+ CompletableFuture<@Nullable Boolean> remoteChangePriorityAsync(ClusterNode remoteNode, UUID jobId, int newPriority) {
JobChangePriorityRequest jobChangePriorityRequest = messagesFactory.jobChangePriorityRequest()
.jobId(jobId)
.priority(newPriority)
@@ -349,27 +342,112 @@ public class ComputeMessaging {
}
private void processJobChangePriorityRequest(JobChangePriorityRequest request, String senderConsistentId, long correlationId) {
- UUID jobId = request.jobId();
- JobExecution<Object> execution = executions.get(jobId);
- if (execution != null) {
- execution.changePriorityAsync(request.priority())
- .whenComplete((result, err) -> sendJobChangePriorityResponse(err, senderConsistentId, correlationId));
- } else {
- ComputeException ex = new ComputeException(Compute.CHANGE_JOB_PRIORITY_NO_JOB_ERR, "Can not change job priority,"
- + " job not found for the job id " + jobId);
- sendJobChangePriorityResponse(ex, senderConsistentId, correlationId);
- }
+ executionManager.changePriorityAsync(request.jobId(), request.priority())
+ .whenComplete((result, err) -> sendJobChangePriorityResponse(result, err, senderConsistentId, correlationId));
}
private void sendJobChangePriorityResponse(
+ @Nullable Boolean result,
@Nullable Throwable throwable,
String senderConsistentId,
Long correlationId
) {
JobChangePriorityResponse jobChangePriorityResponse = messagesFactory.jobChangePriorityResponse()
.throwable(throwable)
+ .result(result)
.build();
messagingService.respond(senderConsistentId, jobChangePriorityResponse, correlationId);
}
+
+ /**
+ * Broadcasts job status request to all nodes in the cluster.
+ *
+ * @param jobId Job id.
+ * @return The current status of the job, or {@code null} if the job status no longer exists due to exceeding the retention time limit.
+ */
+ public CompletableFuture<@Nullable JobStatus> broadcastStatusAsync(UUID jobId) {
+ return broadcastAsync(
+ node -> remoteStatusAsync(node, jobId),
+ throwable -> new ComputeException(
+ FAIL_TO_GET_JOB_STATUS_ERR,
+ "Failed to retrieve status of the job with ID: " + jobId,
+ throwable
+ ));
+ }
+
+ /**
+ * Broadcasts job cancel request to all nodes in the cluster.
+ *
+ * @param jobId Job id.
+ * @return The future which will be completed with {@code true} when the job is cancelled, {@code false} when the job couldn't be
+ * cancelled (either it's not yet started, or it's already completed), or {@code null} if there's no job with the specified id.
+ */
+ public CompletableFuture<@Nullable Boolean> broadcastCancelAsync(UUID jobId) {
+ return broadcastAsync(
+ node -> remoteCancelAsync(node, jobId),
+ throwable -> new ComputeException(
+ CANCELLING_ERR,
+ "Failed to cancel job with ID: " + jobId,
+ throwable
+ ));
+ }
+
+ /**
+ * Broadcasts job priority change request to all nodes in the cluster.
+ *
+ * @param jobId Job id.
+ * @param newPriority New priority.
+ * @return The future which will be completed with {@code true} when the priority is changed, {@code false} when the priority couldn't
+ * be changed (it's already executing or completed), or {@code null} if there's no job with the specified id.
+ */
+ public CompletableFuture<@Nullable Boolean> broadcastChangePriorityAsync(UUID jobId, int newPriority) {
+ return broadcastAsync(
+ node -> remoteChangePriorityAsync(node, jobId, newPriority),
+ throwable -> new ComputeException(
+ CHANGE_JOB_PRIORITY_ERR,
+ "Failed to change priority for job with ID: " + jobId,
+ throwable
+ ));
+ }
+
+ /**
+ * Broadcasts a request to all nodes in the cluster.
+ *
+ * @param request Function which maps a node to the request future.
+ * @param error Function which creates a specific error from the exception thrown from the request.
+ * @return The future which will be completed when request is processed.
+ */
+ private <R> CompletableFuture<@Nullable R> broadcastAsync(
+ Function<ClusterNode, CompletableFuture<@Nullable R>> request,
+ Function<Throwable, Throwable> error
+ ) {
+ CompletableFuture<@Nullable R> result = new CompletableFuture<>();
+
+ CompletableFuture<?>[] futures = topologyService.allMembers()
+ .stream()
+ .map(node -> request.apply(node)
+ .thenAccept(response -> {
+ if (response != null) {
+ result.complete(response);
+ }
+ }))
+ .toArray(CompletableFuture[]::new);
+
+ allOf(futures).whenComplete((unused, throwable) -> {
+ // If none of the nodes returned non-null status it means that either we couldn't find the status
+ // or the node which had the status thrown an exception. If any of the futures completed exceptionally
+ // but the result is non-null, then ignore the exceptions from other futures.
+ if (!result.isDone()) {
+ // allOf will complete exceptionally if any of the futures failed, so this condition means that we
+ // successfully couldn't find a status.
+ if (throwable == null) {
+ result.complete(null);
+ return;
+ }
+ result.completeExceptionally(error.apply(throwable));
+ }
+ });
+ return result;
+ }
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/messaging/RemoteJobExecution.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/messaging/RemoteJobExecution.java
index 1ece202966..a4affd440e 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/messaging/RemoteJobExecution.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/messaging/RemoteJobExecution.java
@@ -23,6 +23,7 @@ import org.apache.ignite.compute.JobExecution;
import org.apache.ignite.compute.JobStatus;
import org.apache.ignite.internal.future.InFlightFutures;
import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.Nullable;
/**
* Remote job execution implementation.
@@ -69,21 +70,21 @@ public class RemoteJobExecution<R> implements JobExecution<R> {
}
@Override
- public CompletableFuture<JobStatus> statusAsync() {
+ public CompletableFuture<@Nullable JobStatus> statusAsync() {
return inFlightFutures.registerFuture(
jobIdFuture.thenCompose(jobId -> messaging.remoteStatusAsync(remoteNode, jobId))
);
}
@Override
- public CompletableFuture<Void> cancelAsync() {
+ public CompletableFuture<@Nullable Boolean> cancelAsync() {
return inFlightFutures.registerFuture(
jobIdFuture.thenCompose(jobId -> messaging.remoteCancelAsync(remoteNode, jobId))
);
}
@Override
- public CompletableFuture<Void> changePriorityAsync(int newPriority) {
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) {
return inFlightFutures.registerFuture(
jobIdFuture.thenCompose(jobId -> messaging.remoteChangePriorityAsync(remoteNode, jobId, newPriority))
);
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/CancellingException.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/CancellingException.java
deleted file mode 100644
index defc72cde9..0000000000
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/CancellingException.java
+++ /dev/null
@@ -1,32 +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.queue;
-
-import static org.apache.ignite.lang.ErrorGroups.Compute.CANCELLING_ERR;
-
-import java.util.UUID;
-import org.apache.ignite.compute.ComputeException;
-
-/**
- * Thrown when job cancel request failed.
- */
-public class CancellingException extends ComputeException {
- public CancellingException(UUID jobId) {
- super(CANCELLING_ERR, "Cancelling job " + jobId + " failed.");
- }
-}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecution.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecution.java
index f603ec3530..a5710412a8 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecution.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecution.java
@@ -46,8 +46,10 @@ public interface QueueExecution<R> {
/**
* Cancels the job.
+ *
+ * @return {@code true} if job was successfully cancelled.
*/
- void cancel();
+ boolean cancel();
/**
* Change job priority. Priority can be changed only if task still in executor's queue.
@@ -55,6 +57,7 @@ public interface QueueExecution<R> {
* Queue entry will be executed last in the queue of entries with the same priority (FIFO).
*
* @param newPriority new priority.
+ * @return {@code true} if job priority was successfully changed.
*/
- void changePriority(int newPriority);
+ boolean changePriority(int newPriority);
}
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecutionImpl.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecutionImpl.java
index bac4da281b..02f195fd08 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecutionImpl.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/queue/QueueExecutionImpl.java
@@ -30,7 +30,6 @@ import org.apache.ignite.internal.compute.state.ComputeStateMachine;
import org.apache.ignite.internal.compute.state.IllegalJobStateTransition;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.lang.ErrorGroups.Compute;
import org.jetbrains.annotations.Nullable;
/**
@@ -86,7 +85,7 @@ class QueueExecutionImpl<R> implements QueueExecution<R> {
}
@Override
- public void cancel() {
+ public boolean cancel() {
try {
stateMachine.cancelingJob(jobId);
@@ -94,27 +93,28 @@ class QueueExecutionImpl<R> implements QueueExecution<R> {
if (queueEntry != null) {
executor.remove(queueEntry);
queueEntry.interrupt();
+ return true;
}
} catch (IllegalJobStateTransition e) {
LOG.info("Cannot cancel the job", e);
- throw new CancellingException(jobId);
}
+ return false;
}
@Override
- public void changePriority(int newPriority) {
+ public boolean changePriority(int newPriority) {
if (newPriority == priority.get()) {
- return;
+ return false;
}
QueueEntry<R> queueEntry = this.queueEntry.get();
if (executor.removeFromQueue(queueEntry)) {
this.priority.set(newPriority);
this.queueEntry.set(null);
run();
- } else {
- throw new ComputeException(Compute.CHANGE_JOB_PRIORITY_JOB_EXECUTING_ERR, "Can not change job priority,"
- + " job already processing. [job id = " + jobId + "]");
+ return true;
}
+ LOG.info("Cannot change job priority, job already processing. [job id = {}]", job);
+ return false;
}
/**
diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/state/InMemoryComputeStateMachine.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/state/InMemoryComputeStateMachine.java
index f35b84f4bb..4654ee1d17 100644
--- a/modules/compute/src/main/java/org/apache/ignite/internal/compute/state/InMemoryComputeStateMachine.java
+++ b/modules/compute/src/main/java/org/apache/ignite/internal/compute/state/InMemoryComputeStateMachine.java
@@ -23,25 +23,18 @@ 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.compute.JobState.QUEUED;
-import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
import java.time.Instant;
-import java.util.HashSet;
import java.util.Map;
-import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import org.apache.ignite.compute.JobState;
import org.apache.ignite.compute.JobStatus;
+import org.apache.ignite.internal.compute.Cleaner;
import org.apache.ignite.internal.compute.configuration.ComputeConfiguration;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
-import org.apache.ignite.internal.thread.NamedThreadFactory;
/**
* In memory implementation of {@link ComputeStateMachine}.
@@ -51,40 +44,26 @@ public class InMemoryComputeStateMachine implements ComputeStateMachine {
private final ComputeConfiguration configuration;
- private ExecutorService cleaner;
+ private final String nodeName;
- private final Set<UUID> toRemove = new HashSet<>();
-
- private final Set<UUID> waitToRemove = ConcurrentHashMap.newKeySet();
+ private final Cleaner<JobStatus> cleaner = new Cleaner<>();
private final Map<UUID, JobStatus> statuses = new ConcurrentHashMap<>();
- public InMemoryComputeStateMachine(ComputeConfiguration configuration) {
+ public InMemoryComputeStateMachine(ComputeConfiguration configuration, String nodeName) {
this.configuration = configuration;
+ this.nodeName = nodeName;
}
@Override
public void start() {
- Long lifetime = configuration.statesLifetimeMillis().value();
- ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
- new NamedThreadFactory("InMemoryComputeStateMachine-pool", LOG)
- );
- executor.scheduleAtFixedRate(() -> {
- Set<UUID> nextToRemove = Set.of(waitToRemove.toArray(UUID[]::new));
- this.waitToRemove.removeAll(nextToRemove);
-
- for (UUID jobId : toRemove) {
- statuses.remove(jobId);
- }
- toRemove.clear();
- toRemove.addAll(nextToRemove);
- }, lifetime, lifetime, TimeUnit.MILLISECONDS);
- cleaner = executor;
+ long ttlMillis = configuration.statesLifetimeMillis().value();
+ cleaner.start(statuses::remove, ttlMillis, nodeName);
}
@Override
public void stop() {
- shutdownAndAwaitTermination(cleaner, 1000, TimeUnit.MILLISECONDS);
+ cleaner.stop();
}
@Override
@@ -117,7 +96,7 @@ public class InMemoryComputeStateMachine implements ComputeStateMachine {
@Override
public void failJob(UUID jobId) {
changeJobState(jobId, FAILED);
- waitToRemove.add(jobId);
+ cleaner.scheduleRemove(jobId);
}
@Override
@@ -128,14 +107,14 @@ public class InMemoryComputeStateMachine implements ComputeStateMachine {
@Override
public void completeJob(UUID jobId) {
changeJobState(jobId, COMPLETED);
- waitToRemove.add(jobId);
+ cleaner.scheduleRemove(jobId);
}
@Override
public void cancelingJob(UUID jobId) {
changeJobState(jobId, currentState -> {
if (currentState == QUEUED) {
- waitToRemove.add(jobId);
+ cleaner.scheduleRemove(jobId);
return CANCELED;
} else if (currentState == EXECUTING) {
return CANCELING;
@@ -148,7 +127,7 @@ public class InMemoryComputeStateMachine implements ComputeStateMachine {
@Override
public void cancelJob(UUID jobId) {
changeJobState(jobId, CANCELED);
- waitToRemove.add(jobId);
+ cleaner.scheduleRemove(jobId);
}
private void changeJobState(UUID jobId, JobState newState) {
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/ComputeComponentImplTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/ComputeComponentImplTest.java
index 6a4ca07d31..efe77210d8 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/ComputeComponentImplTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/ComputeComponentImplTest.java
@@ -40,6 +40,7 @@ import static org.hamcrest.Matchers.startsWith;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Answers.RETURNS_DEEP_STUBS;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -65,7 +66,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.Ignite;
-import org.apache.ignite.compute.ComputeException;
import org.apache.ignite.compute.ComputeJob;
import org.apache.ignite.compute.DeploymentUnit;
import org.apache.ignite.compute.JobExecution;
@@ -89,7 +89,6 @@ import org.apache.ignite.internal.compute.message.JobResultRequest;
import org.apache.ignite.internal.compute.message.JobResultResponse;
import org.apache.ignite.internal.compute.message.JobStatusRequest;
import org.apache.ignite.internal.compute.message.JobStatusResponse;
-import org.apache.ignite.internal.compute.queue.CancellingException;
import org.apache.ignite.internal.compute.state.InMemoryComputeStateMachine;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
@@ -106,6 +105,7 @@ import org.apache.ignite.network.MessagingService;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.NetworkMessageHandler;
+import org.apache.ignite.network.TopologyService;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -113,7 +113,6 @@ import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
-import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -129,14 +128,16 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
@Mock
private MessagingService messagingService;
+ @Mock(answer = RETURNS_DEEP_STUBS)
+ private TopologyService topologyService;
+
@InjectConfiguration
private ComputeConfiguration computeConfiguration;
@Mock
private JobContextManager jobContextManager;
- @InjectMocks
- private ComputeComponentImpl computeComponent;
+ private ComputeComponent computeComponent;
private ComputeExecutor computeExecutor;
@@ -179,6 +180,7 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
@BeforeEach
void setUp() {
lenient().when(ignite.name()).thenReturn(INSTANCE_NAME);
+ lenient().when(topologyService.localMember().name()).thenReturn(INSTANCE_NAME);
JobClassLoader classLoader = new JobClassLoader(List.of(), new URL[0], getClass().getClassLoader());
JobContext jobContext = new JobContext(classLoader, ignored -> {});
@@ -195,9 +197,16 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
willCompleteSuccessfully()
);
- computeExecutor = new ComputeExecutorImpl(ignite, new InMemoryComputeStateMachine(computeConfiguration), computeConfiguration);
+ InMemoryComputeStateMachine stateMachine = new InMemoryComputeStateMachine(computeConfiguration, INSTANCE_NAME);
+ computeExecutor = new ComputeExecutorImpl(ignite, stateMachine, computeConfiguration);
- computeComponent = new ComputeComponentImpl(messagingService, jobContextManager, computeExecutor);
+ computeComponent = new ComputeComponentImpl(
+ messagingService,
+ topologyService,
+ jobContextManager,
+ computeExecutor,
+ computeConfiguration
+ );
computeComponent.start();
}
@@ -213,8 +222,8 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
assertThat(execution.resultAsync(), willBe("jobResponse"));
assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
- assertThat(execution.changePriorityAsync(1), willThrow(ComputeException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
+ assertThat(execution.changePriorityAsync(1), willBe(false));
assertThatNoRequestsWereSent();
}
@@ -225,7 +234,7 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
await().until(execution::statusAsync, willBe(jobStatusWithState(EXECUTING)));
- assertThat(execution.cancelAsync(), willCompleteSuccessfully());
+ assertThat(execution.cancelAsync(), willBe(true));
await().until(execution::statusAsync, willBe(jobStatusWithState(CANCELED)));
@@ -254,7 +263,7 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
respondWithExecuteResponseWhenExecuteRequestIsSent(jobId);
respondWithJobResultResponseWhenJobResultRequestIsSent(jobId);
respondWithJobStatusResponseWhenJobStatusRequestIsSent(jobId, COMPLETED);
- respondWithCancellingExceptionWhenJobCancelRequestIsSent(jobId);
+ respondWithJobCancelResponseWhenJobCancelRequestIsSent(jobId, false);
JobExecution<String> execution = computeComponent.executeRemotely(remoteNode, List.of(), SimpleJob.class.getName(), "a", 42);
assertThat(execution.resultAsync(), willBe("remoteResponse"));
@@ -263,10 +272,12 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
assertThat(execution.resultAsync(), willBe("remoteResponse"));
assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
- assertThat(execution.cancelAsync(), willThrow(CancellingException.class));
+ assertThat(execution.cancelAsync(), willBe(false));
assertThatExecuteRequestWasSent(SimpleJob.class.getName(), "a", 42);
assertThatJobResultRequestWasSent(jobId);
+ assertThatJobStatusRequestWasSent(jobId);
+ assertThatJobCancelRequestWasSent(jobId);
}
@Test
@@ -275,13 +286,13 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
respondWithExecuteResponseWhenExecuteRequestIsSent(jobId);
respondWithJobResultResponseWhenJobResultRequestIsSent(jobId);
respondWithJobStatusResponseWhenJobStatusRequestIsSent(jobId, EXECUTING);
- respondWithJobCancelResponseWhenJobCancelRequestIsSent(jobId);
+ respondWithJobCancelResponseWhenJobCancelRequestIsSent(jobId, true);
JobExecution<String> execution = computeComponent.executeRemotely(remoteNode, List.of(), LongJob.class.getName());
assertThat(execution.statusAsync(), willBe(jobStatusWithState(EXECUTING)));
assertThat(execution.resultAsync(), willBe("remoteResponse"));
- assertThat(execution.cancelAsync(), willCompleteSuccessfully());
+ assertThat(execution.cancelAsync(), willBe(true));
assertThatExecuteRequestWasSent(LongJob.class.getName());
assertThatJobResultRequestWasSent(jobId);
@@ -297,7 +308,7 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
JobExecution<String> execution = computeComponent.executeRemotely(remoteNode, List.of(), LongJob.class.getName());
- assertThat(execution.changePriorityAsync(1), willCompleteSuccessfully());
+ assertThat(execution.changePriorityAsync(1), willBe(true));
assertThatJobChangePriorityRequestWasSent(jobId);
}
@@ -326,16 +337,9 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
.thenReturn(completedFuture(jobStatusResponse));
}
- private void respondWithJobCancelResponseWhenJobCancelRequestIsSent(UUID jobId) {
- JobCancelResponse jobCancelResponse = new ComputeMessagesFactory().jobCancelResponse()
- .build();
- when(messagingService.invoke(any(ClusterNode.class), argThat(msg -> jobCancelRequestWithJobId(msg, jobId)), anyLong()))
- .thenReturn(completedFuture(jobCancelResponse));
- }
-
- private void respondWithCancellingExceptionWhenJobCancelRequestIsSent(UUID jobId) {
+ private void respondWithJobCancelResponseWhenJobCancelRequestIsSent(UUID jobId, boolean result) {
JobCancelResponse jobCancelResponse = new ComputeMessagesFactory().jobCancelResponse()
- .throwable(new CancellingException(jobId))
+ .result(result)
.build();
when(messagingService.invoke(any(ClusterNode.class), argThat(msg -> jobCancelRequestWithJobId(msg, jobId)), anyLong()))
.thenReturn(completedFuture(jobCancelResponse));
@@ -343,6 +347,7 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
private void respondWithJobChangePriorityResponseWhenJobChangePriorityRequestIsSent(UUID jobId) {
JobChangePriorityResponse jobChangePriorityResponse = new ComputeMessagesFactory().jobChangePriorityResponse()
+ .result(true)
.build();
when(messagingService.invoke(any(ClusterNode.class), argThat(msg -> jobChangePriorityRequestWithJobId(msg, jobId)), anyLong()))
.thenReturn(completedFuture(jobChangePriorityResponse));
@@ -699,8 +704,10 @@ class ComputeComponentImplTest extends BaseIgniteAbstractTest {
computeComponent = new ComputeComponentImpl(
messagingService,
+ topologyService,
jobContextManager,
- computeExecutor
+ computeExecutor,
+ computeConfiguration
);
computeComponent.start();
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 32c194940c..ffc4f3af71 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
@@ -21,6 +21,7 @@ import static java.util.Collections.singleton;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+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.Mockito.doReturn;
@@ -43,6 +44,7 @@ import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.network.TopologyService;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.table.mapper.Mapper;
+import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -61,7 +63,7 @@ class IgniteComputeImplTest extends BaseIgniteAbstractTest {
private IgniteTablesInternal igniteTables;
@Mock
- private ComputeComponent computeComponent;
+ private ComputeComponentImpl computeComponent;
@InjectMocks
private IgniteComputeImpl compute;
@@ -163,17 +165,17 @@ class IgniteComputeImplTest extends BaseIgniteAbstractTest {
}
@Override
- public CompletableFuture<JobStatus> statusAsync() {
+ public CompletableFuture<@Nullable JobStatus> statusAsync() {
return nullCompletedFuture();
}
@Override
- public CompletableFuture<Void> cancelAsync() {
- return nullCompletedFuture();
+ public CompletableFuture<@Nullable Boolean> cancelAsync() {
+ return trueCompletedFuture();
}
@Override
- public CompletableFuture<Void> changePriorityAsync(int newPriority) {
+ public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) {
return nullCompletedFuture();
}
};
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/executor/ComputeExecutorTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/executor/ComputeExecutorTest.java
index cec00ba307..7da36308e6 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/executor/ComputeExecutorTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/executor/ComputeExecutorTest.java
@@ -58,7 +58,8 @@ class ComputeExecutorTest extends BaseIgniteAbstractTest {
@BeforeEach
void setUp() {
- computeExecutor = new ComputeExecutorImpl(ignite, new InMemoryComputeStateMachine(computeConfiguration), computeConfiguration);
+ InMemoryComputeStateMachine stateMachine = new InMemoryComputeStateMachine(computeConfiguration, "testNode");
+ computeExecutor = new ComputeExecutorImpl(ignite, stateMachine, computeConfiguration);
computeExecutor.start();
}
@@ -75,7 +76,7 @@ class ComputeExecutorTest extends BaseIgniteAbstractTest {
new Object[]{}
);
JobStatus executingStatus = await().until(execution::status, jobStatusWithState(EXECUTING));
- execution.cancel();
+ assertThat(execution.cancel(), is(true));
await().until(
execution::status,
jobStatusWithStateAndCreateTimeStartTime(CANCELED, executingStatus.createTime(), executingStatus.startTime())
@@ -104,7 +105,7 @@ class ComputeExecutorTest extends BaseIgniteAbstractTest {
new Object[]{}
);
JobStatus executingStatus = await().until(execution::status, jobStatusWithState(EXECUTING));
- execution.cancel();
+ assertThat(execution.cancel(), is(true));
await().until(
execution::status,
jobStatusWithStateAndCreateTimeStartTime(CANCELED, executingStatus.createTime(), executingStatus.startTime())
@@ -213,4 +214,23 @@ class ComputeExecutorTest extends BaseIgniteAbstractTest {
}
+ @Test
+ void cancelCompletedJob() {
+ JobExecutionInternal<Integer> execution = computeExecutor.executeJob(
+ ExecutionOptions.DEFAULT,
+ SimpleJob.class,
+ new Object[]{}
+ );
+
+ await().until(execution::status, jobStatusWithState(COMPLETED));
+
+ assertThat(execution.cancel(), is(false));
+ }
+
+ private static class SimpleJob implements ComputeJob<Integer> {
+ @Override
+ public Integer execute(JobExecutionContext context, Object... args) {
+ return 0;
+ }
+ }
}
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/queue/PriorityQueueExecutorTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/queue/PriorityQueueExecutorTest.java
index 9b9a289fa2..a062e0055a 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/queue/PriorityQueueExecutorTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/queue/PriorityQueueExecutorTest.java
@@ -30,14 +30,12 @@ import static org.apache.ignite.internal.testframework.matchers.JobStatusMatcher
import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
-import static org.junit.jupiter.api.Assertions.assertThrows;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.ignite.compute.ComputeException;
import org.apache.ignite.compute.JobStatus;
import org.apache.ignite.internal.compute.configuration.ComputeConfiguration;
import org.apache.ignite.internal.compute.state.InMemoryComputeStateMachine;
@@ -238,7 +236,7 @@ public class PriorityQueueExecutorTest extends BaseIgniteAbstractTest {
JobStatus executingStatus = await().until(execution::status, jobStatusWithState(EXECUTING));
- execution.cancel();
+ assertThat(execution.cancel(), is(true));
await().until(
execution::status,
@@ -258,7 +256,7 @@ public class PriorityQueueExecutorTest extends BaseIgniteAbstractTest {
JobStatus executingStatus = await().until(execution::status, jobStatusWithState(EXECUTING));
- execution.cancel();
+ assertThat(execution.cancel(), is(true));
await().until(
execution::status,
@@ -274,7 +272,7 @@ public class PriorityQueueExecutorTest extends BaseIgniteAbstractTest {
await().until(execution::status, jobStatusWithState(COMPLETED));
- assertThrows(CancellingException.class, execution::cancel);
+ assertThat(execution.cancel(), is(false));
assertThat(execution.status().state(), is(COMPLETED));
}
@@ -297,7 +295,7 @@ public class PriorityQueueExecutorTest extends BaseIgniteAbstractTest {
await().until(execution::status, jobStatusWithState(QUEUED));
// Cancel the task
- execution.cancel();
+ assertThat(execution.cancel(), is(true));
assertThat(execution.status(), jobStatusWithState(CANCELED));
// Finish the running task
@@ -392,7 +390,7 @@ public class PriorityQueueExecutorTest extends BaseIgniteAbstractTest {
assertThat(task3.isDone(), is(false));
//Change priority on task3, it should be executed before task2
- runningExecution3.changePriority(20);
+ assertThat(runningExecution3.changePriority(20), is(true));
//Task 1 should be completed
latch1.countDown();
@@ -457,7 +455,7 @@ public class PriorityQueueExecutorTest extends BaseIgniteAbstractTest {
assertThat(task4.isDone(), is(false));
//Change priority on task3, it should be executed before task2 and task4
- runningExecution.changePriority(20);
+ assertThat(runningExecution.changePriority(20), is(true));
//Current executing task is 3 because we changed priority
latch2.countDown();
@@ -502,8 +500,8 @@ public class PriorityQueueExecutorTest extends BaseIgniteAbstractTest {
assertThat(task1.isDone(), is(false));
assertThat(task2.isDone(), is(false));
- //Change priority on task1, it is already in executing stage, exception should be thrown
- assertThrows(ComputeException.class, () -> runningExecution1.changePriority(20));
+ //Change priority on task1, it is already in executing stage, should return false
+ assertThat(runningExecution1.changePriority(20), is(false));
//Task 1 should not be completed because change priority failed and task2 has higher priority
latch1.countDown();
@@ -540,7 +538,7 @@ public class PriorityQueueExecutorTest extends BaseIgniteAbstractTest {
assertThat(task2.isDone(), is(false));
//Change priority on task3, it should be executed before task2
- assertThrows(ComputeException.class, () -> runningExecution.changePriority(2));
+ assertThat(runningExecution.changePriority(2), is(false));
//Task 1 should not be completed because of changed priority of task3
latch1.countDown();
@@ -563,10 +561,11 @@ public class PriorityQueueExecutorTest extends BaseIgniteAbstractTest {
willCompleteSuccessfully()
);
+ String nodeName = "testNode";
priorityQueueExecutor = new PriorityQueueExecutor(
configuration,
- new NamedThreadFactory(NamedThreadFactory.threadPrefix("testNode", "compute"), LOG),
- new InMemoryComputeStateMachine(configuration)
+ NamedThreadFactory.create(nodeName, "compute", LOG),
+ new InMemoryComputeStateMachine(configuration, nodeName)
);
}
diff --git a/modules/compute/src/test/java/org/apache/ignite/internal/compute/state/InMemoryComputeStateMachineTest.java b/modules/compute/src/test/java/org/apache/ignite/internal/compute/state/InMemoryComputeStateMachineTest.java
index e514a2b398..a23792189f 100644
--- a/modules/compute/src/test/java/org/apache/ignite/internal/compute/state/InMemoryComputeStateMachineTest.java
+++ b/modules/compute/src/test/java/org/apache/ignite/internal/compute/state/InMemoryComputeStateMachineTest.java
@@ -60,7 +60,7 @@ public class InMemoryComputeStateMachineTest extends BaseIgniteAbstractTest {
@BeforeEach
public void setup() {
- stateMachine = new InMemoryComputeStateMachine(configuration);
+ stateMachine = new InMemoryComputeStateMachine(configuration, "testNode");
stateMachine.start();
jobId = stateMachine.initJob();
}
@@ -160,7 +160,7 @@ public class InMemoryComputeStateMachineTest extends BaseIgniteAbstractTest {
public void testCleanStates() throws InterruptedException {
assertThat(configuration.change(change -> change.changeStatesLifetimeMillis(100)), willCompleteSuccessfully());
- stateMachine = new InMemoryComputeStateMachine(configuration);
+ stateMachine = new InMemoryComputeStateMachine(configuration, "testNode");
stateMachine.start();
jobId = stateMachine.initJob();
diff --git a/modules/platforms/cpp/ignite/common/error_codes.h b/modules/platforms/cpp/ignite/common/error_codes.h
index f2911b85b5..6d88e17f2a 100644
--- a/modules/platforms/cpp/ignite/common/error_codes.h
+++ b/modules/platforms/cpp/ignite/common/error_codes.h
@@ -187,8 +187,7 @@ enum class code : underlying_t {
RESULT_NOT_FOUND = 0x100007,
FAIL_TO_GET_JOB_STATUS = 0x100008,
COMPUTE_JOB_FAILED = 0x100009,
- CHANGE_JOB_PRIORITY_NO_JOB = 0x10000a,
- CHANGE_JOB_PRIORITY_JOB_EXECUTING = 0x10000b,
+ CHANGE_JOB_PRIORITY = 0x10000a,
// 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 f6eba7ee3d..7912a01138 100644
--- a/modules/platforms/cpp/ignite/odbc/common_types.cpp
+++ b/modules/platforms/cpp/ignite/odbc/common_types.cpp
@@ -276,8 +276,7 @@ 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::CHANGE_JOB_PRIORITY_JOB_EXECUTING:
+ case error::code::CHANGE_JOB_PRIORITY:
return sql_state::SHY000_GENERAL_ERROR;
// Catalog group. Group code: 17
diff --git a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
index a8642e2b81..ce697ccd18 100644
--- a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
@@ -507,11 +507,8 @@ 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> ChangeJobPriority error. </summary>
+ public const int ChangeJobPriority = (GroupCode << 16) | (10 & 0xFFFF);
}
/// <summary> Catalog errors. </summary>
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
index 08c412743a..4c689300b8 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
@@ -81,6 +81,7 @@ public abstract class ItAbstractThinClientTest extends BaseIgniteAbstractTest {
+ " network.nodeFinder.netClusterNodes: [ \"localhost:3344\", \"localhost:3345\", \"localhost:3346\" ]\n"
+ " clientConnector.port: 10800,\n"
+ " rest.port: 10300\n"
+ + " compute.threadPoolSize: 1\n"
+ "}"
);
@@ -93,6 +94,7 @@ public abstract class ItAbstractThinClientTest extends BaseIgniteAbstractTest {
+ " clientConnector.metricsEnabled: true\n"
+ " clientConnector.port: 10801,\n"
+ " rest.port: 10301\n"
+ + " compute.threadPoolSize: 1\n"
+ "}"
);
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientComputeTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientComputeTest.java
index 9ba3737d5a..6473062580 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientComputeTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientComputeTest.java
@@ -17,13 +17,21 @@
package org.apache.ignite.internal.runner.app.client;
+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.compute.JobState.QUEUED;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrowFast;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.will;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.JobStatusMatcher.jobStatusWithState;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import static org.apache.ignite.lang.ErrorGroups.Table.COLUMN_ALREADY_EXISTS_ERR;
+import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.oneOf;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -100,14 +108,87 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest {
@Test
void testExecuteOnSpecificNodeAsync() {
- assertThat(
- client().compute().<String>executeAsync(Set.of(node(0)), List.of(), NodeNameJob.class.getName()).resultAsync(),
- willBe("itcct_n_3344")
- );
- assertThat(
- client().compute().<String>executeAsync(Set.of(node(1)), List.of(), NodeNameJob.class.getName()).resultAsync(),
- willBe("itcct_n_3345")
- );
+ JobExecution<String> execution1 = client().compute().executeAsync(Set.of(node(0)), List.of(), NodeNameJob.class.getName());
+ JobExecution<String> execution2 = client().compute().executeAsync(Set.of(node(1)), List.of(), NodeNameJob.class.getName());
+
+ assertThat(execution1.resultAsync(), willBe("itcct_n_3344"));
+ assertThat(execution2.resultAsync(), willBe("itcct_n_3345"));
+
+ assertThat(execution1.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ assertThat(execution2.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ }
+
+ @Test
+ void testCancellingCompletedJob() {
+ JobExecution<String> execution = client().compute().executeAsync(Set.of(node(0)), List.of(), NodeNameJob.class.getName());
+
+ assertThat(execution.resultAsync(), willBe("itcct_n_3344"));
+
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+
+ assertThat(execution.cancelAsync(), willBe(false));
+ }
+
+ @Test
+ void testChangingPriorityCompletedJob() {
+ JobExecution<String> execution = client().compute().executeAsync(Set.of(node(0)), List.of(), NodeNameJob.class.getName());
+
+ assertThat(execution.resultAsync(), willBe("itcct_n_3344"));
+
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+
+ assertThat(execution.changePriorityAsync(0), willBe(false));
+ }
+
+ @Test
+ void testCancelOnSpecificNodeAsync() {
+ int sleepMs = 1_000_000;
+ JobExecution<String> execution1 = client().compute().executeAsync(Set.of(node(0)), List.of(), SleepJob.class.getName(), sleepMs);
+ JobExecution<String> execution2 = client().compute().executeAsync(Set.of(node(1)), List.of(), SleepJob.class.getName(), sleepMs);
+
+ await().until(execution1::statusAsync, willBe(jobStatusWithState(EXECUTING)));
+ await().until(execution2::statusAsync, willBe(jobStatusWithState(EXECUTING)));
+
+ assertThat(execution1.cancelAsync(), willBe(true));
+ assertThat(execution2.cancelAsync(), willBe(true));
+
+ await().until(execution1::statusAsync, willBe(jobStatusWithState(CANCELED)));
+ await().until(execution2::statusAsync, willBe(jobStatusWithState(CANCELED)));
+ }
+
+ @Test
+ void changeJobPriority() {
+ int sleepMs = 1_000_000;
+ // Start 1 task in executor with 1 thread
+ JobExecution<String> execution1 = client().compute().executeAsync(Set.of(node(0)), List.of(), SleepJob.class.getName(), sleepMs);
+ await().until(execution1::statusAsync, willBe(jobStatusWithState(EXECUTING)));
+
+ // Start one more long lasting task
+ JobExecution<String> execution2 = client().compute().executeAsync(Set.of(node(0)), List.of(), SleepJob.class.getName(), sleepMs);
+ await().until(execution2::statusAsync, willBe(jobStatusWithState(QUEUED)));
+
+ // Start third task
+ JobExecution<String> execution3 = client().compute().executeAsync(Set.of(node(0)), List.of(), SleepJob.class.getName(), sleepMs);
+ await().until(execution3::statusAsync, willBe(jobStatusWithState(QUEUED)));
+
+ // Task 2 and 3 are not completed, in queue state
+ assertThat(execution2.resultAsync().isDone(), is(false));
+ assertThat(execution3.resultAsync().isDone(), is(false));
+
+ // Change priority of task 3, so it should be executed before task 2
+ assertThat(execution3.changePriorityAsync(2), willBe(true));
+
+ // Cancel task 1, task 3 should start executing
+ assertThat(execution1.cancelAsync(), willBe(true));
+ await().until(execution1::statusAsync, willBe(jobStatusWithState(CANCELED)));
+ await().until(execution3::statusAsync, willBe(jobStatusWithState(EXECUTING)));
+
+ // Task 2 is still queued
+ assertThat(execution2.statusAsync(), willBe(jobStatusWithState(QUEUED)));
+
+ // Cleanup
+ assertThat(execution2.cancelAsync(), willBe(true));
+ assertThat(execution3.cancelAsync(), willBe(true));
}
@Test
@@ -119,10 +200,14 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest {
@Test
void testExecuteOnRandomNodeAsync() {
+ JobExecution<String> execution = client().compute()
+ .executeAsync(new HashSet<>(sortedNodes()), List.of(), NodeNameJob.class.getName());
+
assertThat(
- client().compute().<String>executeAsync(new HashSet<>(sortedNodes()), List.of(), NodeNameJob.class.getName()).resultAsync(),
+ execution.resultAsync(),
will(oneOf("itcct_n_3344", "itcct_n_3345"))
);
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
}
@Test
@@ -136,7 +221,10 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest {
assertEquals(1, futuresPerNode.size());
- assertThat(futuresPerNode.get(node(1)).resultAsync(), willBe("itcct_n_3345__123"));
+ JobExecution<String> execution = futuresPerNode.get(node(1));
+
+ assertThat(execution.resultAsync(), willBe("itcct_n_3345__123"));
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
}
@Test
@@ -150,17 +238,48 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest {
assertEquals(2, futuresPerNode.size());
- assertThat(futuresPerNode.get(node(0)).resultAsync(), willBe("itcct_n_3344__123"));
- assertThat(futuresPerNode.get(node(1)).resultAsync(), willBe("itcct_n_3345__123"));
+ JobExecution<String> execution1 = futuresPerNode.get(node(0));
+ JobExecution<String> execution2 = futuresPerNode.get(node(1));
+
+ assertThat(execution1.resultAsync(), willBe("itcct_n_3344__123"));
+ assertThat(execution2.resultAsync(), willBe("itcct_n_3345__123"));
+
+ assertThat(execution1.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ assertThat(execution2.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ }
+
+ @Test
+ void testCancelBroadcastAllNodes() {
+ int sleepMs = 1_000_000;
+ Map<ClusterNode, JobExecution<String>> futuresPerNode = client().compute().broadcastAsync(
+ new HashSet<>(sortedNodes()),
+ List.of(),
+ SleepJob.class.getName(),
+ sleepMs
+ );
+
+ assertEquals(2, futuresPerNode.size());
+
+ JobExecution<String> execution1 = futuresPerNode.get(node(0));
+ JobExecution<String> execution2 = futuresPerNode.get(node(1));
+
+ await().until(execution1::statusAsync, willBe(jobStatusWithState(EXECUTING)));
+ await().until(execution2::statusAsync, willBe(jobStatusWithState(EXECUTING)));
+
+ assertThat(execution1.cancelAsync(), willBe(true));
+ assertThat(execution2.cancelAsync(), willBe(true));
+
+ await().until(execution1::statusAsync, willBe(jobStatusWithState(CANCELED)));
+ await().until(execution2::statusAsync, willBe(jobStatusWithState(CANCELED)));
}
@Test
void testExecuteWithArgs() {
var nodes = new HashSet<>(client().clusterNodes());
- assertThat(
- client().compute().<String>executeAsync(nodes, List.of(), ConcatJob.class.getName(), 1, "2", 3.3).resultAsync(),
- willBe("1_2_3.3")
- );
+ JobExecution<String> execution = client().compute().executeAsync(nodes, List.of(), ConcatJob.class.getName(), 1, "2", 3.3);
+
+ assertThat(execution.resultAsync(), willBe("1_2_3.3"));
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
}
@ParameterizedTest
@@ -169,12 +288,16 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest {
IgniteException cause;
if (async) {
+ JobExecution<String> execution = client().compute()
+ .executeAsync(Set.of(node(0)), List.of(), IgniteExceptionJob.class.getName());
+
CompletionException ex = assertThrows(
CompletionException.class,
- () -> client().compute().<String>executeAsync(Set.of(node(0)), List.of(), IgniteExceptionJob.class.getName())
- .resultAsync().join()
+ () -> execution.resultAsync().join()
);
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(FAILED)));
+
cause = (IgniteException) ex.getCause();
} else {
IgniteException ex = assertThrows(
@@ -197,12 +320,15 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest {
IgniteException cause;
if (async) {
+ JobExecution<String> execution = client().compute().executeAsync(Set.of(node(0)), List.of(), ExceptionJob.class.getName());
+
CompletionException ex = assertThrows(
CompletionException.class,
- () -> client().compute().<String>executeAsync(Set.of(node(0)), List.of(), ExceptionJob.class.getName())
- .resultAsync().join()
+ () -> execution.resultAsync().join()
);
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(FAILED)));
+
cause = (IgniteException) ex.getCause();
} else {
IgniteException ex = assertThrows(
@@ -225,12 +351,15 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest {
IgniteException cause;
if (async) {
+ JobExecution<String> execution = client().compute().executeAsync(Set.of(node(1)), List.of(), ExceptionJob.class.getName());
+
CompletionException ex = assertThrows(
CompletionException.class,
- () -> client().compute().executeAsync(Set.of(node(1)), List.of(), ExceptionJob.class.getName())
- .resultAsync().join()
+ () -> execution.resultAsync().join()
);
+ assertThat(execution.statusAsync(), willBe(jobStatusWithState(FAILED)));
+
cause = (IgniteException) ex.getCause();
} else {
IgniteException ex = assertThrows(
@@ -253,29 +382,82 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest {
@ParameterizedTest
@CsvSource({"1,3344", "2,3345", "3,3345", "10,3344"})
- void testExecuteColocatedRunsComputeJobOnKeyNode(int key, int port) {
- var table = TABLE_NAME;
+ void testExecuteColocatedTupleRunsComputeJobOnKeyNode(int key, int port) {
var keyTuple = Tuple.create().set(COLUMN_KEY, key);
- var keyPojo = new TestPojo(key);
- CompletableFuture<String> tupleRes = client().compute().<String>executeColocatedAsync(
- table,
+ JobExecution<String> tupleExecution = client().compute().executeColocatedAsync(
+ TABLE_NAME,
keyTuple,
List.of(),
NodeNameJob.class.getName()
- ).resultAsync();
+ );
+
+ String expectedNode = "itcct_n_" + port;
+ assertThat(tupleExecution.resultAsync(), willBe(expectedNode));
- CompletableFuture<String> pojoRes = client().compute().<TestPojo, String>executeColocatedAsync(
- table,
+ assertThat(tupleExecution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"1,3344", "2,3345", "3,3345", "10,3344"})
+ void testExecuteColocatedPojoRunsComputeJobOnKeyNode(int key, int port) {
+ var keyPojo = new TestPojo(key);
+
+ JobExecution<String> pojoExecution = client().compute().executeColocatedAsync(
+ TABLE_NAME,
keyPojo,
Mapper.of(TestPojo.class),
List.of(),
NodeNameJob.class.getName()
- ).resultAsync();
+ );
String expectedNode = "itcct_n_" + port;
- assertThat(tupleRes, willBe(expectedNode));
- assertThat(pojoRes, willBe(expectedNode));
+ assertThat(pojoExecution.resultAsync(), willBe(expectedNode));
+
+ assertThat(pojoExecution.statusAsync(), willBe(jobStatusWithState(COMPLETED)));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"1,3344", "2,3345", "3,3345", "10,3344"})
+ void testCancelColocatedTuple(int key, int port) {
+ var keyTuple = Tuple.create().set(COLUMN_KEY, key);
+ int sleepMs = 1_000_000;
+
+ JobExecution<String> tupleExecution = client().compute().executeColocatedAsync(
+ TABLE_NAME,
+ keyTuple,
+ List.of(),
+ SleepJob.class.getName(),
+ sleepMs
+ );
+
+ await().until(tupleExecution::statusAsync, willBe(jobStatusWithState(EXECUTING)));
+
+ assertThat(tupleExecution.cancelAsync(), willBe(true));
+
+ await().until(tupleExecution::statusAsync, willBe(jobStatusWithState(CANCELED)));
+ }
+
+ @ParameterizedTest
+ @CsvSource({"1,3344", "2,3345", "3,3345", "10,3344"})
+ void testCancelColocatedPojo(int key, int port) {
+ var keyPojo = new TestPojo(key);
+ int sleepMs = 1_000_000;
+
+ JobExecution<String> pojoExecution = client().compute().executeColocatedAsync(
+ TABLE_NAME,
+ keyPojo,
+ Mapper.of(TestPojo.class),
+ List.of(),
+ SleepJob.class.getName(),
+ sleepMs
+ );
+
+ await().until(pojoExecution::statusAsync, willBe(jobStatusWithState(EXECUTING)));
+
+ assertThat(pojoExecution.cancelAsync(), willBe(true));
+
+ await().until(pojoExecution::statusAsync, willBe(jobStatusWithState(CANCELED)));
}
@Test
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 d8611e60dc..0ba11bee44 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
@@ -67,6 +67,7 @@ import org.apache.ignite.internal.components.LongJvmPauseDetector;
import org.apache.ignite.internal.compute.ComputeComponent;
import org.apache.ignite.internal.compute.ComputeComponentImpl;
import org.apache.ignite.internal.compute.IgniteComputeImpl;
+import org.apache.ignite.internal.compute.IgniteComputeInternal;
import org.apache.ignite.internal.compute.configuration.ComputeConfiguration;
import org.apache.ignite.internal.compute.executor.ComputeExecutorImpl;
import org.apache.ignite.internal.compute.loader.JobClassLoaderFactory;
@@ -280,7 +281,7 @@ public class IgniteImpl implements Ignite {
private final ConfigurationStorage cfgStorage;
/** Compute. */
- private final IgniteCompute compute;
+ private final IgniteComputeInternal compute;
/** JVM pause detector. */
private final LongJvmPauseDetector longJvmPauseDetector;
@@ -657,11 +658,13 @@ public class IgniteImpl implements Ignite {
deploymentManager = deploymentManagerImpl;
ComputeConfiguration computeCfg = nodeConfigRegistry.getConfiguration(ComputeConfiguration.KEY);
- InMemoryComputeStateMachine stateMachine = new InMemoryComputeStateMachine(computeCfg);
+ InMemoryComputeStateMachine stateMachine = new InMemoryComputeStateMachine(computeCfg, name);
computeComponent = new ComputeComponentImpl(
clusterSvc.messagingService(),
+ clusterSvc.topologyService(),
new JobContextManager(deploymentManagerImpl, deploymentManagerImpl.deploymentUnitAccessor(), new JobClassLoaderFactory()),
- new ComputeExecutorImpl(this, stateMachine, computeCfg)
+ new ComputeExecutorImpl(this, stateMachine, computeCfg),
+ computeCfg
);
compute = new IgniteComputeImpl(clusterSvc.topologyService(), distributedTblMgr, computeComponent);