IGNITE-22478 Use single argument in Compute API

modules/api/src/main/java/org/apache/ignite/compute/ComputeJob.java

@@ -18,21 +18,47 @@
 package org.apache.ignite.compute;
 
 import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.marshaling.ByteArrayMarshaler;
+import org.apache.ignite.marshaling.Marshaler;
 import org.jetbrains.annotations.Nullable;
 
 /**
- * A Compute job that may be executed on a single Ignite node, on several nodes, or on the entire cluster.
+ * Core Ignite Compute Job interface. If you want to define your own job, you should implement this interface and
+ * deploy the job to the cluster with Deployment API. Then, you can execute this job on the cluster by calling
+ * {@link IgniteCompute} APIs.
  *
- * @param <R> Job result type.
+ * <p>If you want to pass/return custom data structures to/from the job, you should also implement {@link Marshaler}
+ * and return it from {@link #inputMarshaler()} and {@link #resultMarshaler()} methods.
+ *
+ * @param <T> Type of the job argument.
+ * @param <R> Type of the job result.
  */
 @SuppressWarnings("InterfaceMayBeAnnotatedFunctional")
-public interface ComputeJob<R> {
+public interface ComputeJob<T, R> {
     /**
      * Executes the job on an Ignite node.
      *
      * @param context The execution context.
-     * @param args Job arguments.
+     * @param arg Job arguments.
      * @return Job future. Can be null if the job is synchronous and does not return any result.
      */
-    @Nullable CompletableFuture<R> executeAsync(JobExecutionContext context, Object... args);
+    @Nullable CompletableFuture<R> executeAsync(JobExecutionContext context, @Nullable T arg);
+
+    /**
+     * Marshaler for the input argument. Default is {@link ByteArrayMarshaler}.
+     *
+     * @return Input marshaler.
+     */
+    default @Nullable Marshaler<T, byte[]> inputMarshaler() {
+        return null;
+    }
+
+    /**
+     * Marshaler for the job result. Default is {@link ByteArrayMarshaler}.
+     *
+     * @return Result marshaler.
+     */
+    default @Nullable Marshaler<R, byte[]> resultMarshaler() {
+        return null;
+    }
 }

modules/api/src/main/java/org/apache/ignite/compute/IgniteCompute.java

@@ -30,6 +30,7 @@
 import org.apache.ignite.compute.task.MapReduceTask;
 import org.apache.ignite.compute.task.TaskExecution;
 import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * Provides the ability to execute Compute jobs.
@@ -41,83 +42,88 @@ public interface IgniteCompute {
     /**
      * Submits a {@link ComputeJob} of the given class for an execution on a single node from a set of candidate nodes.
      *
-     * @param <R> Job result type.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
      * @param target Execution target.
      * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
+     * @param arg Arguments of the job.
      * @return Job execution object.
      */
-    <R> JobExecution<R> submit(
+    <T, R> JobExecution<R> submit(
             JobTarget target,
             JobDescriptor descriptor,
-            Object... args
+            @Nullable T arg
     );
 
     /**
      * Submits a {@link ComputeJob} of the given class for an execution on a single node from a set of candidate nodes. A shortcut for
      * {@code submit(...).resultAsync()}.
      *
-     * @param <R> Job result type.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
      * @param target Execution target.
      * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
+     * @param arg Arguments of the job.
      * @return Job result future.
      */
-    default <R> CompletableFuture<R> executeAsync(
+    default <T, R> CompletableFuture<R> executeAsync(
             JobTarget target,
             JobDescriptor descriptor,
-            Object... args
+            @Nullable T arg
     ) {
-        return this.<R>submit(target, descriptor, args).resultAsync();
+        return this.<T, R>submit(target, descriptor, arg).resultAsync();
     }
 
     /**
      * Executes a {@link ComputeJob} of the given class on a single node from a set of candidate nodes.
      *
-     * @param <R> Job result type
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
      * @param target Execution target.
      * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
+     * @param arg Arguments of the job.
      * @return Job result.
      * @throws ComputeException If there is any problem executing the job.
      */
-    <R> R execute(
+    <T, R> R execute(
             JobTarget target,
             JobDescriptor descriptor,
-            Object... args
+            @Nullable T arg
     );
 
     /**
      * Submits a {@link ComputeJob} of the given class for an execution on all nodes in the given node set.
      *
-     * @param <R> Job result type.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
      * @param nodes Nodes to execute the job on.
      * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
+     * @param arg Arguments of the job.
      * @return Map from node to job execution object.
      */
-    <R> Map<ClusterNode, JobExecution<R>> submitBroadcast(
+    <T, R> Map<ClusterNode, JobExecution<R>> submitBroadcast(
             Set<ClusterNode> nodes,
             JobDescriptor descriptor,
-            Object... args
+            @Nullable T arg
     );
 
     /**
      * Executes a {@link ComputeJob} of the given class on all nodes in the given node set.
      *
-     * @param <R> Job result type.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
      * @param nodes Nodes to execute the job on.
      * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
+     * @param arg Arguments of the job.
      * @return Map from node to job result.
      */
-    default <R> CompletableFuture<Map<ClusterNode, R>> executeBroadcastAsync(
+    default <T, R> CompletableFuture<Map<ClusterNode, R>> executeBroadcastAsync(
             Set<ClusterNode> nodes,
             JobDescriptor descriptor,
-            Object... args
+            @Nullable T arg
     ) {
         Map<ClusterNode, CompletableFuture<R>> futures = nodes.stream()
-                .collect(toMap(identity(), node -> this.executeAsync(JobTarget.node(node), descriptor, args)));
+                .collect(toMap(identity(), node -> this.executeAsync(JobTarget.node(node), descriptor, arg)));
 
         return allOf(futures.values().toArray(CompletableFuture[]::new))
                 .thenApply(ignored -> {
@@ -135,22 +141,23 @@ default <R> CompletableFuture<Map<ClusterNode, R>> executeBroadcastAsync(
     /**
      * Executes a {@link ComputeJob} of the given class on all nodes in the given node set.
      *
-     * @param <R> Job result type.
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
      * @param nodes Nodes to execute the job on.
      * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
+     * @param arg Arguments of the job.
      * @return Map from node to job result.
      * @throws ComputeException If there is any problem executing the job.
      */
-    default <R> Map<ClusterNode, R> executeBroadcast(
+    default <T, R> Map<ClusterNode, R> executeBroadcast(
             Set<ClusterNode> nodes,
             JobDescriptor descriptor,
-            Object... args
+            @Nullable T arg
     ) {
         Map<ClusterNode, R> map = new HashMap<>();
 
         for (ClusterNode node : nodes) {
-            map.put(node, execute(JobTarget.node(node), descriptor, args));
+            map.put(node, execute(JobTarget.node(node), descriptor, arg));
         }
 
         return map;
@@ -159,36 +166,39 @@ default <R> Map<ClusterNode, R> executeBroadcast(
     /**
      * Submits a {@link MapReduceTask} of the given class for an execution.
      *
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
      * @param units Deployment units.
      * @param taskClassName Map reduce task class name.
-     * @param args Task arguments.
-     * @param <R> Task result type.
+     * @param arg Task arguments.
      * @return Task execution interface.
      */
-    <R> TaskExecution<R> submitMapReduce(List<DeploymentUnit> units, String taskClassName, Object... args);
+    <T, R> TaskExecution<R> submitMapReduce(List<DeploymentUnit> units, String taskClassName, @Nullable T arg);
 
     /**
      * Submits a {@link MapReduceTask} of the given class for an execution. A shortcut for {@code submitMapReduce(...).resultAsync()}.
      *
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
      * @param units Deployment units.
      * @param taskClassName Map reduce task class name.
-     * @param args Task arguments.
-     * @param <R> Task result type.
+     * @param input Task arguments.
      * @return Task result future.
      */
-    default <R> CompletableFuture<R> executeMapReduceAsync(List<DeploymentUnit> units, String taskClassName, Object... args) {
-        return this.<R>submitMapReduce(units, taskClassName, args).resultAsync();
+    default <T, R> CompletableFuture<R> executeMapReduceAsync(List<DeploymentUnit> units, String taskClassName, @Nullable T input) {
+        return this.<T, R>submitMapReduce(units, taskClassName, input).resultAsync();
     }
 
     /**
      * Executes a {@link MapReduceTask} of the given class.
      *
+     * @param <T> Job argument (T)ype.
+     * @param <R> Job (R)esult type.
      * @param units Deployment units.
      * @param taskClassName Map reduce task class name.
-     * @param args Task arguments.
-     * @param <R> Task result type.
+     * @param arg Task arguments.
      * @return Task result.
      * @throws ComputeException If there is any problem executing the task.
      */
-    <R> R executeMapReduce(List<DeploymentUnit> units, String taskClassName, Object... args);
+    <T, R> R executeMapReduce(List<DeploymentUnit> units, String taskClassName, @Nullable T arg);
 }

modules/api/src/main/java/org/apache/ignite/compute/JobDescriptor.java

@@ -19,21 +19,34 @@
 
 import java.util.List;
 import java.util.Objects;
+import org.apache.ignite.marshaling.Marshaler;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * Compute job descriptor.
  */
-public class JobDescriptor {
+public class JobDescriptor<T, R> {
     private final String jobClassName;
 
     private final List<DeploymentUnit> units;
 
     private final JobExecutionOptions options;
 
-    private JobDescriptor(String jobClassName, List<DeploymentUnit> units, JobExecutionOptions options) {
+    private final @Nullable Marshaler<R, byte[]> resultMarshaller;
+
+    private final @Nullable Marshaler<T, byte[]> argumentMarshaler;
+
+    private JobDescriptor(
+            String jobClassName,
+            List<DeploymentUnit> units,
+            JobExecutionOptions options,
+            @Nullable Marshaler<R, byte[]> resultMarshaller,
+            @Nullable Marshaler<T, byte[]> argumentMarshaller) {
         this.jobClassName = jobClassName;
         this.units = units;
         this.options = options;
+        this.resultMarshaller = resultMarshaller;
+        this.argumentMarshaler = argumentMarshaller;
     }
 
     /**
@@ -79,19 +92,29 @@ public static Builder builder(String jobClassName) {
      *
      * @return Job descriptor builder.
      */
-    public static Builder builder(Class<? extends ComputeJob<?>> jobClass) {
+    public static Builder builder(Class<? extends ComputeJob<?, ?>> jobClass) {
         Objects.requireNonNull(jobClass);
 
         return new Builder(jobClass.getName());
     }
 
+    public Marshaler<R, byte[]> resultMarshaller() {
+        return resultMarshaller;
+    }
+
+    public Marshaler<T, byte[]> argumentMarshaler() {
+        return argumentMarshaler;
+    }
+
     /**
      * Builder.
      */
     public static class Builder {
         private final String jobClassName;
         private List<DeploymentUnit> units;
         private JobExecutionOptions options;
+        private Marshaler<?, ?> resultMarshaller;
+        private Marshaler<?, ?> argumentMarshaller;
 
         private Builder(String jobClassName) {
             Objects.requireNonNull(jobClassName);
@@ -132,16 +155,43 @@ public Builder options(JobExecutionOptions options) {
             return this;
         }
 
+        /**
+         * Sets the result marshaller.
+         *
+         * @param marshaller Result marshaller.
+         *
+         * @return This builder.
+         */
+        public Builder resultMarshaller(Marshaler<?, ?> marshaller) {
+            this.resultMarshaller = marshaller;
+            return this;
+        }
+
+        /**
+         * Sets the argument marshaller.
+         *
+         * @param marshaller Argument marshaller.
+         *
+         * @return This builder.
+         */
+        public Builder argumentMarshaller(Marshaler<?, ?> marshaller) {
+            this.argumentMarshaller = marshaller;
+            return this;
+        }
+
         /**
          * Builds the job descriptor.
          *
          * @return Job descriptor.
          */
-        public JobDescriptor build() {
+        public <T, R> JobDescriptor build() {
             return new JobDescriptor(
                     jobClassName,
                     units == null ? List.of() : units,
-                    options == null ? JobExecutionOptions.DEFAULT : options);
+                    options == null ? JobExecutionOptions.DEFAULT : options,
+                    resultMarshaller,
+                    argumentMarshaller
+            );
         }
     }
 }

modules/api/src/main/java/org/apache/ignite/compute/JobExecution.java

@@ -19,6 +19,7 @@
 
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.marshaling.Marshaler;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -70,4 +71,6 @@ public interface JobExecution<R> {
      *         retention time limit.
      */
     CompletableFuture<@Nullable Boolean> changePriorityAsync(int newPriority);
+
+    Marshaler<R, byte[]> resultMarshaler();
 }