Pochatkin commented on code in PR #3104:
URL: https://github.com/apache/ignite-3/pull/3104#discussion_r1470700748
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteRequest.java:
##########
@@ -50,29 +53,40 @@ public class ClientComputeExecuteRequest {
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteCompute compute,
+ IgniteComputeInternal compute,
ClusterService cluster,
- NotificationSender notificationSender) {
- var nodeName = in.tryUnpackNil() ? null : in.unpackString();
-
- var node = nodeName == null
- ? cluster.topologyService().localMember()
- : cluster.topologyService().getByConsistentId(nodeName);
-
- if (node == null) {
- throw new IgniteException("Specified node is not present in the
cluster: " + nodeName);
- }
+ NotificationSender notificationSender
+ ) {
+ Set<ClusterNode> candidates = unpackCandidateNodes(in, cluster);
List<DeploymentUnit> deploymentUnits = unpackDeploymentUnits(in);
String jobClassName = in.unpackString();
JobExecutionOptions options =
JobExecutionOptions.builder().priority(in.unpackInt()).maxRetries(in.unpackInt()).build();
Object[] args = unpackArgs(in);
- JobExecution<Object> execution = compute.executeAsync(Set.of(node),
deploymentUnits, jobClassName, options, args);
+ JobExecution<Object> execution =
compute.executeAsyncWithFailover(candidates, deploymentUnits, jobClassName,
options, args);
sendResultAndStatus(execution, notificationSender);
return execution.idAsync().thenAccept(out::packUuid);
}
+ private static Set<ClusterNode> unpackCandidateNodes(ClientMessageUnpacker
in, ClusterService cluster) {
Review Comment:
Do we need Set here? Mby reduce to Collection?
##########
modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/compute/ClientComputeExecuteRequest.java:
##########
@@ -50,29 +56,82 @@ public class ClientComputeExecuteRequest {
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientMessagePacker out,
- IgniteCompute compute,
+ IgniteComputeInternal compute,
ClusterService cluster,
- NotificationSender notificationSender) {
- var nodeName = in.tryUnpackNil() ? null : in.unpackString();
-
- var node = nodeName == null
- ? cluster.topologyService().localMember()
- : cluster.topologyService().getByConsistentId(nodeName);
-
- if (node == null) {
- throw new IgniteException("Specified node is not present in the
cluster: " + nodeName);
- }
+ NotificationSender notificationSender
+ ) {
+ Set<ClusterNode> candidates = unpackCandidateNodes(in, cluster);
List<DeploymentUnit> deploymentUnits = unpackDeploymentUnits(in);
String jobClassName = in.unpackString();
JobExecutionOptions options =
JobExecutionOptions.builder().priority(in.unpackInt()).maxRetries(in.unpackInt()).build();
Object[] args = unpackArgs(in);
- JobExecution<Object> execution = compute.executeAsync(Set.of(node),
deploymentUnits, jobClassName, options, args);
+ ClusterNode localNode = cluster.topologyService().localMember();
+ ClusterNode targetNode = selectTargetNode(candidates, localNode);
+ candidates.remove(targetNode);
+
+ JobExecution<Object> execution =
compute.executeAsyncWithFailover(targetNode, candidates,
+ deploymentUnits, jobClassName, options, args);
sendResultAndStatus(execution, notificationSender);
return execution.idAsync().thenAccept(out::packUuid);
}
+ private static Set<ClusterNode> unpackCandidateNodes(ClientMessageUnpacker
in, ClusterService cluster) {
+ Set<String> candidateIds =
Arrays.stream(in.unpackObjectArrayFromBinaryTuple())
+ .map(String.class::cast)
+ .collect(Collectors.toSet());
+
+ Set<ClusterNode> candidates = new HashSet<>();
+ List<String> notFoundNodes = new ArrayList<>();
+ candidateIds.forEach(id -> {
+ ClusterNode node = cluster.topologyService().getByConsistentId(id);
+ if (node == null) {
+ notFoundNodes.add(id);
+ } else {
+ candidates.add(node);
+ }
+ });
+
+ if (!notFoundNodes.isEmpty()) {
+ throw new IgniteException("Specified nodes are not present in the
cluster: " + notFoundNodes);
+ }
+ return candidates;
+ }
+
+ /**
+ * Selects a random node from the set of candidates, preferably not a
local node.
Review Comment:
@ptupitsyn If we will choose local node, then our failover feature just put
in the trash. In IEP we have strict conclusion that Compute Job has not recover
mechanism in case when Coordinator node is die, our failover support only
Worker node die. Your approach here is use the Coordinator node as prefered for
Worker node choosing and as result our failover just doesn't work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
