This is an automated email from the ASF dual-hosted git repository. ptupitsyn 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 c56f077040 IGNITE-22124 Java thin: Implement Compute MapReduce API (#3744) c56f077040 is described below commit c56f077040e736b087ad239c688f43897cc096f7 Author: Vadim Kolodin <vkolo...@gridgain.com> AuthorDate: Mon May 20 16:09:56 2024 +0300 IGNITE-22124 Java thin: Implement Compute MapReduce API (#3744) --- .../internal/client/proto/ClientMessagePacker.java | 13 ++ .../client/proto/ClientMessageUnpacker.java | 10 ++ .../ignite/internal/client/proto/ClientOp.java | 5 + .../handler/ClientInboundMessageHandler.java | 4 + .../ClientComputeExecuteMapReduceRequest.java | 100 +++++++++++++++ .../apache/ignite/client/ClientOperationType.java | 5 + .../org/apache/ignite/client/RetryReadPolicy.java | 1 + .../apache/ignite/internal/client/ClientUtils.java | 3 + .../internal/client/compute/ClientCompute.java | 65 ++++++++-- .../client/compute/ClientJobExecution.java | 14 +-- .../client/compute/ClientTaskExecution.java | 140 +++++++++++++++++++++ .../internal/client/compute/SubmitTaskResult.java | 41 ++++++ .../client/compute/task/ClientTaskExecution.java | 57 --------- .../apache/ignite/client/ClientComputeTest.java | 47 +++++++ .../apache/ignite/client/fakes/FakeCompute.java | 58 ++++++++- .../runner/app/client/ItThinClientComputeTest.java | 122 ++++++++++++++++++ 16 files changed, 612 insertions(+), 73 deletions(-) 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 1fc36b24ce..1f07669394 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 @@ -520,6 +520,19 @@ public class ClientMessagePacker implements AutoCloseable { buf.writeLongLE(val.getLeastSignificantBits()); } + /** + * Writes a UUID. + * + * @param val UUID value. + */ + public void packUuidNullable(@Nullable UUID val) { + if (val == null) { + packNil(); + } else { + packUuid(val); + } + } + /** * Writes a bit set. * 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 cba6481c9c..ec8086fb11 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 @@ -672,6 +672,16 @@ public class ClientMessageUnpacker implements AutoCloseable { return new UUID(buf.readLongLE(), buf.readLongLE()); } + /** + * Reads a nullable UUID. + * + * @return UUID or null. + * @throws MessageTypeException when type is not UUID. + */ + public @Nullable UUID unpackUuidNullable() { + return tryUnpackNil() ? null : unpackUuid(); + } + /** * Reads a bit set. * 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 0b83a4869f..bb371faaed 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 @@ -170,4 +170,9 @@ public class ClientOp { /** Execute SQL query with the parameters batch. */ public static final int SQL_EXEC_BATCH = 63; + + /** + * Execute MapReduce task. + */ + public static final int COMPUTE_EXECUTE_MAPREDUCE = 64; } 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 5669f4d285..a9a75b8405 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 @@ -45,6 +45,7 @@ import org.apache.ignite.client.handler.requests.cluster.ClientClusterGetNodesRe 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.ClientComputeExecuteMapReduceRequest; 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; @@ -735,6 +736,9 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im notificationSender(requestId) ); + case ClientOp.COMPUTE_EXECUTE_MAPREDUCE: + return ClientComputeExecuteMapReduceRequest.process(in, out, compute, notificationSender(requestId)); + case ClientOp.COMPUTE_GET_STATUS: return ClientComputeGetStatusRequest.process(in, out, compute); diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteMapReduceRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteMapReduceRequest.java new file mode 100644 index 0000000000..15ea62f430 --- /dev/null +++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteMapReduceRequest.java @@ -0,0 +1,100 @@ +/* + * 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 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.compute.ClientComputeGetStatusRequest.packJobStatus; +import static org.apache.ignite.internal.util.IgniteUtils.firstNotNull; + +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import org.apache.ignite.client.handler.NotificationSender; +import org.apache.ignite.compute.DeploymentUnit; +import org.apache.ignite.compute.JobStatus; +import org.apache.ignite.compute.TaskExecution; +import org.apache.ignite.internal.client.proto.ClientMessagePacker; +import org.apache.ignite.internal.client.proto.ClientMessageUnpacker; +import org.apache.ignite.internal.compute.IgniteComputeInternal; + +/** + * Compute MapReduce request. + */ +public class ClientComputeExecuteMapReduceRequest { + /** + * Processes the request. + * + * @param in Unpacker. + * @param out Packer. + * @param compute Compute. + * @param notificationSender Notification sender. + * @return Future. + */ + public static CompletableFuture<Void> process( + ClientMessageUnpacker in, + ClientMessagePacker out, + IgniteComputeInternal compute, + NotificationSender notificationSender) { + List<DeploymentUnit> deploymentUnits = unpackDeploymentUnits(in); + String taskClassName = in.unpackString(); + Object[] args = unpackArgs(in); + + TaskExecution<Object> execution = compute.submitMapReduce(deploymentUnits, taskClassName, args); + sendTaskResult(execution, notificationSender); + + var idsAsync = execution.idsAsync() + .handle((ids, ex) -> { + // empty ids in case of split exception to properly respond with task id and failed status + return ex == null ? ids : Collections.<UUID>emptyList(); + }); + + return execution.idAsync() + .thenAcceptBoth(idsAsync, (id, ids) -> { + out.packUuid(id); + packJobIds(out, ids); + }); + } + + static void packJobIds(ClientMessagePacker out, List<UUID> ids) { + out.packInt(ids.size()); + for (var uuid : ids) { + out.packUuid(uuid); + } + } + + static CompletableFuture<Object> sendTaskResult(TaskExecution<Object> execution, NotificationSender notificationSender) { + return execution.resultAsync().whenComplete((val, err) -> + execution.statusAsync().whenComplete((status, errStatus) -> + execution.statusesAsync().whenComplete((statuses, errStatuses) -> + notificationSender.sendNotification(w -> { + w.packObjectAsBinaryTuple(val); + packJobStatus(w, status); + packJobStatuses(w, statuses); + }, firstNotNull(err, errStatus, errStatuses))) + )); + } + + static void packJobStatuses(ClientMessagePacker w, List<JobStatus> statuses) { + w.packInt(statuses.size()); + for (JobStatus status : statuses) { + packJobStatus(w, status); + } + } +} 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 f62cecfc90..820e111de4 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 @@ -132,6 +132,11 @@ public enum ClientOperationType { */ COMPUTE_EXECUTE, + /** + * Compute Execute MapReduce ({@link org.apache.ignite.compute.IgniteCompute#submitMapReduce(List, String, Object...)}). + */ + COMPUTE_EXECUTE_MAPREDUCE, + /** * Get compute job status ({@link org.apache.ignite.compute.JobExecution#statusAsync()}). */ 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 482bc8a0e9..4912f2a2db 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,7 @@ public class RetryReadPolicy extends RetryLimitPolicy { case TUPLE_UPSERT: case COMPUTE_EXECUTE: + case COMPUTE_EXECUTE_MAPREDUCE: case COMPUTE_GET_STATUS: case COMPUTE_CANCEL: case COMPUTE_CHANGE_PRIORITY: 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 dde688c5ce..bc76849b24 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,9 @@ public class ClientUtils { case ClientOp.COMPUTE_EXECUTE_COLOCATED: return ClientOperationType.COMPUTE_EXECUTE; + case ClientOp.COMPUTE_EXECUTE_MAPREDUCE: + return ClientOperationType.COMPUTE_EXECUTE_MAPREDUCE; + case ClientOp.COMPUTE_GET_STATUS: return ClientOperationType.COMPUTE_GET_STATUS; 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 63507b113a..efac50488b 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 @@ -20,12 +20,14 @@ 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.ArrayList; import java.util.HashMap; import java.util.Iterator; 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.ConcurrentHashMap; @@ -251,8 +253,10 @@ public class ClientCompute implements IgniteCompute { @Override public <R> TaskExecution<R> submitMapReduce(List<DeploymentUnit> units, String taskClassName, Object... args) { - // TODO https://issues.apache.org/jira/browse/IGNITE-22124 - throw new UnsupportedOperationException("Not implemented yet."); + Objects.requireNonNull(units); + Objects.requireNonNull(taskClassName); + + return new ClientTaskExecution<>(ch, doExecuteMapReduceAsync(units, taskClassName, args)); } @Override @@ -260,6 +264,20 @@ public class ClientCompute implements IgniteCompute { return sync(executeMapReduceAsync(units, taskClassName, args)); } + private CompletableFuture<SubmitTaskResult> doExecuteMapReduceAsync( + List<DeploymentUnit> units, + String taskClassName, + Object... args) { + return ch.serviceAsync( + ClientOp.COMPUTE_EXECUTE_MAPREDUCE, + w -> packTask(w.out(), units, taskClassName, args), + ClientCompute::unpackSubmitTaskResult, + null, + null, + true + ); + } + private CompletableFuture<SubmitResult> executeOnNodesAsync( Set<ClusterNode> nodes, List<DeploymentUnit> units, @@ -418,11 +436,7 @@ public class ClientCompute implements IgniteCompute { String jobClassName, JobExecutionOptions options, Object[] args) { - w.packInt(units.size()); - for (DeploymentUnit unit : units) { - w.packString(unit.name()); - w.packString(unit.version().render()); - } + packDeploymentUnits(w, units); w.packString(jobClassName); w.packInt(options.priority()); @@ -430,6 +444,23 @@ public class ClientCompute implements IgniteCompute { w.packObjectArrayAsBinaryTuple(args); } + private static void packTask(ClientMessagePacker w, + List<DeploymentUnit> units, + String taskClassName, + Object[] args) { + packDeploymentUnits(w, units); + w.packString(taskClassName); + w.packObjectArrayAsBinaryTuple(args); + } + + private static void packDeploymentUnits(ClientMessagePacker w, List<DeploymentUnit> units) { + w.packInt(units.size()); + for (DeploymentUnit unit : units) { + w.packString(unit.name()); + w.packString(unit.version().render()); + } + } + /** * Unpacks job id from channel and gets notification future. This is needed because we need to unpack message response in the payload * reader because the unpacker will be closed after the response is processed. @@ -441,6 +472,26 @@ public class ClientCompute implements IgniteCompute { return new SubmitResult(ch.in().unpackUuid(), ch.notificationFuture()); } + /** + * Unpacks coordination job id and jobs ids which are executing under this task from channel and gets notification future. + * This is needed because we need to unpack message response in the payload + * reader because the unpacker will be closed after the response is processed. + * + * @param ch Payload channel. + * @return Result of the task submission. + */ + private static SubmitTaskResult unpackSubmitTaskResult(PayloadInputChannel ch) { + var jobId = ch.in().unpackUuid(); + + var size = ch.in().unpackInt(); + List<UUID> jobIds = new ArrayList<>(size); + for (int i = 0; i < size; i++) { + jobIds.add(ch.in().unpackUuid()); + } + + return new SubmitTaskResult(jobId, jobIds, ch.notificationFuture()); + } + private static <R> R sync(CompletableFuture<R> future) { try { return future.join(); 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 c95bd5e8ec..f4535aec1b 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 @@ -75,7 +75,7 @@ class ClientJobExecution<R> implements JobExecution<R> { if (statusFuture.isDone()) { return statusFuture; } - return jobIdFuture.thenCompose(this::getJobStatus); + return jobIdFuture.thenCompose(jobId -> getJobStatus(ch, jobId)); } @Override @@ -83,7 +83,7 @@ class ClientJobExecution<R> implements JobExecution<R> { if (statusFuture.isDone()) { return falseCompletedFuture(); } - return jobIdFuture.thenCompose(this::cancelJob); + return jobIdFuture.thenCompose(jobId -> cancelJob(ch, jobId)); } @Override @@ -91,10 +91,10 @@ class ClientJobExecution<R> implements JobExecution<R> { if (statusFuture.isDone()) { return falseCompletedFuture(); } - return jobIdFuture.thenCompose(jobId -> changePriority(jobId, newPriority)); + return jobIdFuture.thenCompose(jobId -> changePriority(ch, jobId, newPriority)); } - private CompletableFuture<@Nullable JobStatus> getJobStatus(UUID jobId) { + static CompletableFuture<@Nullable JobStatus> getJobStatus(ReliableChannel ch, 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( @@ -107,7 +107,7 @@ class ClientJobExecution<R> implements JobExecution<R> { ); } - private CompletableFuture<@Nullable Boolean> cancelJob(UUID jobId) { + static CompletableFuture<@Nullable Boolean> cancelJob(ReliableChannel ch, 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( @@ -120,7 +120,7 @@ class ClientJobExecution<R> implements JobExecution<R> { ); } - private CompletableFuture<@Nullable Boolean> changePriority(UUID jobId, int newPriority) { + static CompletableFuture<@Nullable Boolean> changePriority(ReliableChannel ch, 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( @@ -136,7 +136,7 @@ class ClientJobExecution<R> implements JobExecution<R> { ); } - private static @Nullable JobStatus unpackJobStatus(PayloadInputChannel payloadInputChannel) { + static @Nullable JobStatus unpackJobStatus(PayloadInputChannel payloadInputChannel) { ClientMessageUnpacker unpacker = payloadInputChannel.in(); if (unpacker.tryUnpackNil()) { return null; diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientTaskExecution.java b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientTaskExecution.java new file mode 100644 index 0000000000..9df7d73b6a --- /dev/null +++ b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientTaskExecution.java @@ -0,0 +1,140 @@ +/* + * 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.client.compute; + +import static org.apache.ignite.internal.client.compute.ClientJobExecution.cancelJob; +import static org.apache.ignite.internal.client.compute.ClientJobExecution.changePriority; +import static org.apache.ignite.internal.client.compute.ClientJobExecution.getJobStatus; +import static org.apache.ignite.internal.client.compute.ClientJobExecution.unpackJobStatus; +import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.function.Function; +import org.apache.ignite.compute.JobStatus; +import org.apache.ignite.compute.TaskExecution; +import org.apache.ignite.internal.client.PayloadInputChannel; +import org.apache.ignite.internal.client.ReliableChannel; +import org.apache.ignite.internal.util.CompletableFutures; +import org.jetbrains.annotations.Nullable; + +/** + * Client compute task implementation. + * + * @param <R> Task result type. + */ +class ClientTaskExecution<R> implements TaskExecution<R> { + private final ReliableChannel ch; + + private final CompletableFuture<UUID> jobIdFuture; + + private final CompletableFuture<List<@Nullable UUID>> jobIdsFuture; + + private final CompletableFuture<R> resultAsync; + + // Local status cache + private final CompletableFuture<@Nullable JobStatus> statusFuture = new CompletableFuture<>(); + + // Local statuses cache + private final CompletableFuture<List<@Nullable JobStatus>> statusesFutures = new CompletableFuture<>(); + + ClientTaskExecution(ReliableChannel ch, CompletableFuture<SubmitTaskResult> reqFuture) { + this.ch = ch; + + jobIdFuture = reqFuture.thenApply(SubmitTaskResult::jobId); + jobIdsFuture = reqFuture.thenApply(SubmitTaskResult::jobIds); + + resultAsync = reqFuture + .thenCompose(SubmitTaskResult::notificationFuture) + .thenApply(payloadInputChannel -> { + // Notifications require explicit input close. + try (payloadInputChannel) { + R result = (R) payloadInputChannel.in().unpackObjectFromBinaryTuple(); + statusFuture.complete(unpackJobStatus(payloadInputChannel)); + statusesFutures.complete(unpackJobStatuses(payloadInputChannel)); + return result; + } + }); + } + + @Override + public CompletableFuture<R> resultAsync() { + return resultAsync; + } + + @Override + public CompletableFuture<@Nullable JobStatus> statusAsync() { + if (statusFuture.isDone()) { + return statusFuture; + } + return jobIdFuture.thenCompose(jobId -> getJobStatus(ch, jobId)); + } + + @Override + public CompletableFuture<@Nullable Boolean> cancelAsync() { + if (statusFuture.isDone()) { + return falseCompletedFuture(); + } + return jobIdFuture.thenCompose(jobId -> cancelJob(ch, jobId)); + } + + @Override + public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) { + if (statusFuture.isDone()) { + return falseCompletedFuture(); + } + return jobIdFuture.thenCompose(jobId -> changePriority(ch, jobId, newPriority)); + } + + @Override + public CompletableFuture<List<@Nullable JobStatus>> statusesAsync() { + if (statusesFutures.isDone()) { + return statusesFutures; + } + + return jobIdsFuture.thenCompose(ids -> { + @SuppressWarnings("unchecked") + CompletableFuture<@Nullable JobStatus>[] futures = ids.stream() + .map(jobId -> getJobStatus(ch, jobId)) + .toArray(CompletableFuture[]::new); + + return CompletableFutures.allOf(futures) + .thenApply(Function.identity()); + }); + } + + private static List<@Nullable JobStatus> unpackJobStatuses(PayloadInputChannel payloadInputChannel) { + var unpacker = payloadInputChannel.in(); + var size = unpacker.unpackInt(); + + if (size == 0) { + return Collections.emptyList(); + } + + var statuses = new ArrayList<@Nullable JobStatus>(size); + for (int i = 0; i < size; i++) { + var status = unpackJobStatus(payloadInputChannel); + statuses.add(status); + } + + return Collections.unmodifiableList(statuses); + } +} diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/SubmitTaskResult.java b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/SubmitTaskResult.java new file mode 100644 index 0000000000..71ef494201 --- /dev/null +++ b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/SubmitTaskResult.java @@ -0,0 +1,41 @@ +/* + * 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.client.compute; + +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import org.apache.ignite.internal.client.PayloadInputChannel; +import org.jetbrains.annotations.Nullable; + +/** + * Result of the task submission. + * Contains unpacked job id, a collection of ids of the jobs which are executing under this task, and notification future. + */ +class SubmitTaskResult extends SubmitResult { + private final List<UUID> jobIds; + + SubmitTaskResult(UUID jobId, List<UUID> jobIds, CompletableFuture<PayloadInputChannel> notificationFuture) { + super(jobId, notificationFuture); + this.jobIds = jobIds; + } + + List<@Nullable UUID> jobIds() { + return jobIds; + } +} diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/task/ClientTaskExecution.java b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/task/ClientTaskExecution.java deleted file mode 100644 index dd1b766cef..0000000000 --- a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/task/ClientTaskExecution.java +++ /dev/null @@ -1,57 +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.client.compute.task; - -import java.util.List; -import java.util.concurrent.CompletableFuture; -import org.apache.ignite.compute.JobStatus; -import org.apache.ignite.compute.TaskExecution; -import org.jetbrains.annotations.Nullable; - -// TODO https://issues.apache.org/jira/browse/IGNITE-22124 -/** - * Client compute task implementation. - * - * @param <R> Task result type. - */ -public class ClientTaskExecution<R> implements TaskExecution<R> { - @Override - public CompletableFuture<R> resultAsync() { - throw new UnsupportedOperationException("Not implemented yet."); - } - - @Override - public CompletableFuture<@Nullable JobStatus> statusAsync() { - throw new UnsupportedOperationException("Not implemented yet."); - } - - @Override - public CompletableFuture<@Nullable Boolean> cancelAsync() { - throw new UnsupportedOperationException("Not implemented yet."); - } - - @Override - public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) { - throw new UnsupportedOperationException("Not implemented yet."); - } - - @Override - public CompletableFuture<List<@Nullable JobStatus>> statusesAsync() { - throw new UnsupportedOperationException("Not implemented yet."); - } -} 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 6a58de7d9f..6fb03e47da 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 @@ -28,7 +28,10 @@ import static org.apache.ignite.internal.util.IgniteUtils.closeAll; import static org.apache.ignite.lang.ErrorGroups.Table.TABLE_NOT_FOUND_ERR; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.everyItem; +import static org.hamcrest.Matchers.hasItem; import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.not; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -45,6 +48,7 @@ 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.TaskExecution; import org.apache.ignite.compute.version.Version; import org.apache.ignite.internal.client.table.ClientTable; import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest; @@ -216,6 +220,49 @@ public class ClientComputeTest extends BaseIgniteAbstractTest { } } + @Test + void testMapReduceExecute() throws Exception { + initServers(reqId -> false); + + try (var client = getClient(server1)) { + Object[] args = {"arg1", 2}; + String res1 = client.compute().executeMapReduce(List.of(), "job", args); + assertEquals("s1", res1); + } + } + + @Test + void testMapReduceSubmit() throws Exception { + initServers(reqId -> false); + + try (var client = getClient(server1)) { + TaskExecution<Object> task = client.compute().submitMapReduce(List.of(), "job"); + + assertThat(task.resultAsync(), willBe("s1")); + + assertThat(task.statusAsync(), willBe(jobStatusWithState(COMPLETED))); + assertThat(task.statusesAsync(), willBe(everyItem(jobStatusWithState(COMPLETED)))); + + assertThat("compute task and sub tasks ids must be different", + task.idsAsync(), willBe(not(hasItem(task.idAsync().get())))); + } + } + + @Test + void testMapReduceException() throws Exception { + initServers(reqId -> false); + + try (var client = getClient(server1)) { + FakeCompute.future = CompletableFuture.failedFuture(new RuntimeException("job failed")); + + TaskExecution<Object> execution = client.compute().submitMapReduce(List.of(), "job"); + + assertThat(execution.resultAsync(), willThrowFast(IgniteException.class)); + assertThat(execution.statusAsync(), willBe(jobStatusWithState(FAILED))); + assertThat(execution.statusesAsync(), willBe(everyItem(jobStatusWithState(FAILED)))); + } + } + @Test void testUnitsPropagation() throws Exception { initServers(reqId -> false); 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 3d5c9d79d0..6dc1c72a07 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 @@ -34,6 +34,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; +import java.util.function.BiFunction; import java.util.stream.Collectors; import org.apache.ignite.compute.DeploymentUnit; import org.apache.ignite.compute.IgniteCompute; @@ -190,7 +191,7 @@ public class FakeCompute implements IgniteComputeInternal { @Override public <R> TaskExecution<R> submitMapReduce(List<DeploymentUnit> units, String taskClassName, Object... args) { - return null; + return taskExecution(future != null ? future : completedFuture((R) nodeName)); } @Override @@ -215,7 +216,7 @@ public class FakeCompute implements IgniteComputeInternal { result.whenComplete((r, throwable) -> { JobState state = throwable != null ? FAILED : COMPLETED; - JobStatus newStatus = status.toBuilder().state(state).finishTime(Instant.now()).build(); + JobStatus newStatus = status.toBuilder().id(jobId).state(state).finishTime(Instant.now()).build(); statuses.put(jobId, newStatus); }); return new JobExecution<>() { @@ -241,6 +242,59 @@ public class FakeCompute implements IgniteComputeInternal { }; } + private <R> TaskExecution<R> taskExecution(CompletableFuture<R> result) { + BiFunction<UUID, JobState, JobStatus> toStatus = (id, jobState) -> + JobStatus.builder() + .id(id) + .state(jobState) + .createTime(Instant.now()) + .startTime(Instant.now()) + .build(); + + UUID jobId = UUID.randomUUID(); + UUID subJobId1 = UUID.randomUUID(); + UUID subJobId2 = UUID.randomUUID(); + + statuses.put(jobId, toStatus.apply(jobId, EXECUTING)); + statuses.put(subJobId1, toStatus.apply(subJobId1, EXECUTING)); + statuses.put(subJobId2, toStatus.apply(subJobId2, EXECUTING)); + + result.whenComplete((r, throwable) -> { + JobState state = throwable != null ? FAILED : COMPLETED; + + statuses.put(jobId, toStatus.apply(jobId, state)); + statuses.put(subJobId1, toStatus.apply(subJobId1, state)); + statuses.put(subJobId2, toStatus.apply(subJobId2, state)); + }); + + return new TaskExecution<>() { + @Override + public CompletableFuture<R> resultAsync() { + return result; + } + + @Override + public CompletableFuture<@Nullable JobStatus> statusAsync() { + return completedFuture(statuses.get(jobId)); + } + + @Override + public CompletableFuture<List<@Nullable JobStatus>> statusesAsync() { + return completedFuture(List.of(statuses.get(subJobId1), statuses.get(subJobId2))); + } + + @Override + public CompletableFuture<@Nullable Boolean> cancelAsync() { + return trueCompletedFuture(); + } + + @Override + public CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority) { + return trueCompletedFuture(); + } + }; + } + @Override public CompletableFuture<Collection<JobStatus>> statusesAsync() { return completedFuture(statuses.values()); 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 1e7bb01e2e..ca2e22cf2d 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 @@ -32,8 +32,12 @@ import static org.apache.ignite.lang.ErrorGroups.Compute.COMPUTE_JOB_FAILED_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.allOf; import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.everyItem; +import static org.hamcrest.Matchers.hasItem; import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.not; import static org.hamcrest.Matchers.oneOf; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -70,13 +74,20 @@ import org.apache.ignite.compute.ComputeJob; import org.apache.ignite.compute.DeploymentUnit; import org.apache.ignite.compute.JobExecution; import org.apache.ignite.compute.JobExecutionContext; +import org.apache.ignite.compute.TaskExecution; +import org.apache.ignite.compute.task.ComputeJobRunner; +import org.apache.ignite.compute.task.MapReduceTask; +import org.apache.ignite.compute.task.TaskExecutionContext; import org.apache.ignite.lang.IgniteException; import org.apache.ignite.network.ClusterNode; import org.apache.ignite.table.Tuple; import org.apache.ignite.table.mapper.Mapper; +import org.hamcrest.Matcher; +import org.hamcrest.Matchers; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; /** * Thin client compute integration test. @@ -668,6 +679,46 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest { assertEquals(expected, res); } + @Test + void testExecuteMapReduce() throws Exception { + TaskExecution<String> execution = client().compute().submitMapReduce(List.of(), MapReduceNodeNameTask.class.getName()); + + List<Matcher<? super String>> nodeNames = sortedNodes().stream() + .map(ClusterNode::name) + .map(Matchers::containsString) + .collect(Collectors.toList()); + assertThat(execution.resultAsync(), willBe(allOf(nodeNames))); + + assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED))); + assertThat(execution.statusesAsync(), willBe(everyItem(jobStatusWithState(COMPLETED)))); + + assertThat("compute task and sub tasks ids must be different", + execution.idsAsync(), willBe(not(hasItem(execution.idAsync().get())))); + } + + @Test + void testExecuteMapReduceWithArgs() { + TaskExecution<String> execution = client().compute() + .submitMapReduce(List.of(), MapReduceArgsTask.class.getName(), 1, "2", 3.3); + + assertThat(execution.resultAsync(), willBe(containsString("1_2_3.3"))); + assertThat(execution.statusAsync(), willBe(jobStatusWithState(COMPLETED))); + } + + @ParameterizedTest + @ValueSource(classes = {MapReduceExceptionOnSplitTask.class, MapReduceExceptionOnReduceTask.class}) + void testExecuteMapReduceExceptionPropagation(Class<?> taskClass) { + IgniteException cause = getExceptionInJobExecutionAsync( + client().compute().submitMapReduce(List.of(), taskClass.getName()) + ); + + assertThat(cause.getMessage(), containsString("Custom job error")); + assertEquals(TRACE_ID, cause.traceId()); + assertEquals(COLUMN_ALREADY_EXISTS_ERR, cause.code()); + assertInstanceOf(CustomException.class, cause); + assertNull(cause.getCause()); // No stack trace by default. + } + private void testEchoArg(Object arg) { Object res = client().compute().execute(Set.of(node(0)), List.of(), EchoJob.class.getName(), arg, arg.toString()); @@ -755,6 +806,77 @@ public class ItThinClientComputeTest extends ItAbstractThinClientTest { } } + private static class MapReduceNodeNameTask implements MapReduceTask<String> { + @Override + public List<ComputeJobRunner> split(TaskExecutionContext context, Object... args) { + return context.ignite().clusterNodes().stream() + .map(node -> ComputeJobRunner.builder() + .jobClassName(NodeNameJob.class.getName()) + .nodes(Set.of(node)) + .args(args) + .build()) + .collect(Collectors.toList()); + } + + @Override + public String reduce(Map<UUID, ?> results) { + return results.values().stream() + .map(String.class::cast) + .collect(Collectors.joining(",")); + } + } + + private static class MapReduceArgsTask implements MapReduceTask<String> { + @Override + public List<ComputeJobRunner> split(TaskExecutionContext context, Object... args) { + return context.ignite().clusterNodes().stream() + .map(node -> ComputeJobRunner.builder() + .jobClassName(ConcatJob.class.getName()) + .nodes(Set.of(node)) + .args(args) + .build()) + .collect(Collectors.toList()); + } + + @Override + public String reduce(Map<UUID, ?> results) { + return results.values().stream() + .map(String.class::cast) + .collect(Collectors.joining(",")); + } + } + + private static class MapReduceExceptionOnSplitTask implements MapReduceTask<String> { + @Override + public List<ComputeJobRunner> split(TaskExecutionContext context, Object... args) { + throw new CustomException(TRACE_ID, COLUMN_ALREADY_EXISTS_ERR, "Custom job error", null); + } + + @Override + public String reduce(Map<UUID, ?> results) { + return "expected split exception"; + } + } + + private static class MapReduceExceptionOnReduceTask implements MapReduceTask<String> { + + @Override + public List<ComputeJobRunner> split(TaskExecutionContext context, Object... args) { + return context.ignite().clusterNodes().stream() + .map(node -> ComputeJobRunner.builder() + .jobClassName(NodeNameJob.class.getName()) + .nodes(Set.of(node)) + .args(args) + .build()) + .collect(Collectors.toList()); + } + + @Override + public String reduce(Map<UUID, ?> results) { + throw new CustomException(TRACE_ID, COLUMN_ALREADY_EXISTS_ERR, "Custom job error", null); + } + } + /** * Custom public exception class. */