IGNITE-22435 Add JobTarget interface

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

@@ -0,0 +1,59 @@
+/*
+ * 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.compute;
+
+import java.util.Objects;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.table.mapper.Mapper;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Colocated job execution target.
+ */
+public class ColocatedExecutionTarget implements JobTarget {
+    private final String tableName;
+
+    private final Object key;
+
+    private final @Nullable Mapper<?> keyMapper;
+
+    ColocatedExecutionTarget(String tableName, Object key, @Nullable Mapper<?> keyMapper) {
+        Objects.requireNonNull(tableName);
+        Objects.requireNonNull(key);
+
+        if (keyMapper == null && !(key instanceof Tuple)) {
+            throw new IllegalArgumentException("Key must be an instance of Tuple when keyMapper is not provided.");
+        }
+
+        this.tableName = tableName;
+        this.key = key;
+        this.keyMapper = keyMapper;
+    }
+
+    public String tableName() {
+        return tableName;
+    }
+
+    public Object key() {
+        return key;
+    }
+
+    public @Nullable Mapper<?> keyMapper() {
+        return keyMapper;
+    }
+}

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

@@ -30,8 +30,6 @@
 import org.apache.ignite.compute.task.MapReduceTask;
 import org.apache.ignite.compute.task.TaskExecution;
 import org.apache.ignite.network.ClusterNode;
-import org.apache.ignite.table.Tuple;
-import org.apache.ignite.table.mapper.Mapper;
 
 /**
  * Provides the ability to execute Compute jobs.
@@ -44,13 +42,13 @@ 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 nodes Candidate nodes; the job will be executed on one of them.
+     * @param target Execution target.
      * @param descriptor Job descriptor.
      * @param args Arguments of the job.
      * @return Job execution object.
      */
     <R> JobExecution<R> submit(
-            Set<ClusterNode> nodes,
+            JobTarget target,
             JobDescriptor descriptor,
             Object... args
     );
@@ -60,151 +58,35 @@ <R> JobExecution<R> submit(
      * {@code submit(...).resultAsync()}.
      *
      * @param <R> Job result type.
-     * @param nodes Candidate nodes; the job will be executed on one of them.
+     * @param target Execution target.
      * @param descriptor Job descriptor.
      * @param args Arguments of the job.
      * @return Job result future.
      */
     default <R> CompletableFuture<R> executeAsync(
-            Set<ClusterNode> nodes,
+            JobTarget target,
             JobDescriptor descriptor,
             Object... args
     ) {
-        return this.<R>submit(nodes, descriptor, args).resultAsync();
+        return this.<R>submit(target, descriptor, args).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 nodes Candidate nodes; the job will be executed on one of them.
+     * @param target Execution target.
      * @param descriptor Job descriptor.
      * @param args Arguments of the job.
      * @return Job result.
      * @throws ComputeException If there is any problem executing the job.
      */
     <R> R execute(
-            Set<ClusterNode> nodes,
-            JobDescriptor descriptor,
-            Object... args
-    );
-
-    /**
-     * Submits a job of the given class for the execution on the node where the given key is located. The node is a leader of the
-     * corresponding RAFT group.
-     *
-     * @param tableName Name of the table whose key is used to determine the node to execute the job on.
-     * @param key Key that identifies the node to execute the job on.
-     * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
-     * @param <R> Job result type.
-     * @return Job execution object.
-     */
-    <R> JobExecution<R> submitColocated(
-            String tableName,
-            Tuple key,
+            JobTarget target,
             JobDescriptor descriptor,
             Object... args
     );
 
-    /**
-     * Submits a job of the given class for the execution on the node where the given key is located. The node is a leader of the
-     * corresponding RAFT group.
-     *
-     * @param tableName Name of the table whose key is used to determine the node to execute the job on.
-     * @param key Key that identifies the node to execute the job on.
-     * @param keyMapper Mapper used to map the key to a binary representation.
-     * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
-     * @param <R> Job result type.
-     * @return Job execution object.
-     */
-    <K, R> JobExecution<R> submitColocated(
-            String tableName,
-            K key,
-            Mapper<K> keyMapper,
-            JobDescriptor descriptor,
-            Object... args
-    );
-
-    /**
-     * Submits a job of the given class for the execution on the node where the given key is located. The node is a leader of the
-     * corresponding RAFT group. A shortcut for {@code submitColocated(...).resultAsync()}.
-     *
-     * @param tableName Name of the table whose key is used to determine the node to execute the job on.
-     * @param key Key that identifies the node to execute the job on.
-     * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
-     * @param <R> Job result type.
-     * @return Job result future.
-     */
-    default <R> CompletableFuture<R> executeColocatedAsync(
-            String tableName,
-            Tuple key,
-            JobDescriptor descriptor,
-            Object... args
-    ) {
-        return this.<R>submitColocated(tableName, key, descriptor, args).resultAsync();
-    }
-
-    /**
-     * Submits a job of the given class for the execution on the node where the given key is located. The node is a leader of the
-     * corresponding RAFT group. A shortcut for {@code submitColocated(...).resultAsync()}.
-     *
-     * @param tableName Name of the table whose key is used to determine the node to execute the job on.
-     * @param key Key that identifies the node to execute the job on.
-     * @param keyMapper Mapper used to map the key to a binary representation.
-     * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
-     * @param <R> Job result type.
-     * @return Job result future.
-     */
-    default <K, R> CompletableFuture<R> executeColocatedAsync(
-            String tableName,
-            K key,
-            Mapper<K> keyMapper,
-            JobDescriptor descriptor,
-            Object... args
-    ) {
-        return this.<K, R>submitColocated(tableName, key, keyMapper, descriptor, args).resultAsync();
-    }
-
-    /**
-     * Executes a job of the given class on the node where the given key is located. The node is a leader of the corresponding RAFT group.
-     *
-     * @param <R> Job result type.
-     * @param tableName Name of the table whose key is used to determine the node to execute the job on.
-     * @param key Key that identifies the node to execute the job on.
-     * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
-     * @return Job result.
-     * @throws ComputeException If there is any problem executing the job.
-     */
-    <R> R executeColocated(
-            String tableName,
-            Tuple key,
-            JobDescriptor descriptor,
-            Object... args);
-
-    /**
-     * Executes a job of the given class on the node where the given key is located. The node is a leader of the corresponding RAFT group.
-     *
-     * @param <R> Job result type.
-     * @param tableName Name of the table whose key is used to determine the node to execute the job on.
-     * @param key Key that identifies the node to execute the job on.
-     * @param keyMapper Mapper used to map the key to a binary representation.
-     * @param descriptor Job descriptor.
-     * @param args Arguments of the job.
-     * @return Job result.
-     * @throws ComputeException If there is any problem executing the job.
-     */
-    <K, R> R executeColocated(
-            String tableName,
-            K key,
-            Mapper<K> keyMapper,
-            JobDescriptor descriptor,
-            Object... args);
-
     /**
      * Submits a {@link ComputeJob} of the given class for an execution on all nodes in the given node set.
      *
@@ -235,7 +117,7 @@ default <R> CompletableFuture<Map<ClusterNode, R>> executeBroadcastAsync(
             Object... args
     ) {
         Map<ClusterNode, CompletableFuture<R>> futures = nodes.stream()
-                .collect(toMap(identity(), node -> this.executeAsync(Set.of(node), descriptor, args)));
+                .collect(toMap(identity(), node -> this.executeAsync(JobTarget.node(node), descriptor, args)));
 
         return allOf(futures.values().toArray(CompletableFuture[]::new))
                 .thenApply(ignored -> {
@@ -268,7 +150,7 @@ default <R> Map<ClusterNode, R> executeBroadcast(
         Map<ClusterNode, R> map = new HashMap<>();
 
         for (ClusterNode node : nodes) {
-            map.put(node, execute(Set.of(node), descriptor, args));
+            map.put(node, execute(JobTarget.node(node), descriptor, args));
         }
 
         return map;

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

@@ -0,0 +1,92 @@
+/*
+ * 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.compute;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.table.mapper.Mapper;
+
+/**
+ * Job execution target.
+ */
+public interface JobTarget {
+    /**
+     * Creates a job target for a specific node.
+     *
+     * @param node Node.
+     * @return Job target.
+     */
+    static JobTarget node(ClusterNode node) {
+        return new NodesJobTarget(Set.of(node));
+    }
+
+    /**
+     * Creates a job target for any node from the provided collection.
+     *
+     * @param nodes Collection of nodes.
+     * @return Job target.
+     */
+    static JobTarget anyNode(ClusterNode... nodes) {
+        return new NodesJobTarget(Set.of(nodes));
+    }
+
+    /**
+     * Creates a job target for any node from the provided collection.
+     *
+     * @param nodes Collection of nodes.
+     * @return Job target.
+     */
+    static JobTarget anyNode(Collection<ClusterNode> nodes) {
+        return new NodesJobTarget(new HashSet<>(nodes));
+    }
+
+    /**
+     * Creates a job target for any node from the provided collection.
+     *
+     * @param nodes Collection of nodes.
+     * @return Job target.
+     */
+    static JobTarget anyNode(Set<ClusterNode> nodes) {
+        return new NodesJobTarget(nodes);
+    }
+
+    /**
+     * Creates a colocated job target for a specific table and key.
+     *
+     * @param tableName Table name.
+     * @param key Key.
+     * @return Job target.
+     */
+    static JobTarget colocated(String tableName, Tuple key) {
+        return new ColocatedExecutionTarget(tableName, key, null);
+    }
+
+    /**
+     * Creates a colocated job target for a specific table and key with mapper.
+     *
+     * @param tableName Table name.
+     * @param key Key.
+     * @return Job target.
+     */
+    static <K> JobTarget colocated(String tableName, K key, Mapper<K> keyMapper) {
+        return new ColocatedExecutionTarget(tableName, key, keyMapper);
+    }
+}

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

@@ -0,0 +1,43 @@
+/*
+ * 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.compute;
+
+import java.util.Objects;
+import java.util.Set;
+import org.apache.ignite.network.ClusterNode;
+
+/**
+ * Nodes-based job execution target.
+ */
+public class NodesJobTarget implements JobTarget {
+    private final Set<ClusterNode> nodes;
+
+    NodesJobTarget(Set<ClusterNode> nodes) {
+        Objects.requireNonNull(nodes);
+
+        if (nodes.isEmpty()) {
+            throw new IllegalArgumentException("Nodes collection must not be empty.");
+        }
+
+        this.nodes = nodes;
+    }
+
+    public Set<ClusterNode> nodes() {
+        return nodes;
+    }
+}