rpuch commented on code in PR #7156:
URL: https://github.com/apache/ignite-3/pull/7156#discussion_r2645414762
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java:
##########
@@ -1000,7 +1005,7 @@ private
CompletableFuture<LocalTablePartitionStateResponse> tableStateForZoneOnN
.catalogVersion(catalogVersion)
.build();
- return messagingService.invoke(node, request,
TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS))
+ return messagingService.invoke(node, request,
TimeUnit.SECONDS.toMillis(DISASTER_RECOVERY_TIMEOUT_MILLIS))
Review Comment:
```suggestion
return messagingService.invoke(node, request,
DISASTER_RECOVERY_TIMEOUT_MILLIS)
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java:
##########
@@ -1197,44 +1202,39 @@ private CompletableFuture<Void>
remoteProcessingFuture(DisasterRecoveryRequest r
MultiNodeDisasterRecoveryRequest multiNodeRequest =
(MultiNodeDisasterRecoveryRequest) request;
- Collection<String> actualNodeNames =
getActualNodeNames(multiNodeRequest.nodeNames());
+ Set<NodeWithAttributes> nodes =
getRequestNodes(multiNodeRequest.nodeNames());
- CompletableFuture<?>[] remoteProcessingFutures = actualNodeNames
+ CompletableFuture<?>[] remoteProcessingFutures = nodes
.stream()
- .map(nodeName -> addMultiNodeOperation(nodeName, operationId))
+ .map(node -> addMultiNodeOperation(node, operationId))
.toArray(CompletableFuture[]::new);
- return allOf(remoteProcessingFutures)
- .whenComplete((ignored, e) -> {
- for (String nodeName : actualNodeNames) {
- operationsByNodeName.compute(nodeName, (node,
operations) -> {
- if (operations != null) {
- operations.remove(operationId);
-
- return operations.isEmpty() ? null :
operations;
- }
-
- return null;
- });
- }
- });
+ return allOf(remoteProcessingFutures);
}
/** If request node names is empty, returns all nodes in the logical
topology. */
- private Collection<String> getActualNodeNames(Set<String>
requestNodeNames) {
- if (requestNodeNames.isEmpty()) {
- return dzManager.logicalTopology().stream()
- .map(NodeWithAttributes::nodeName)
- .collect(toSet());
- } else {
- return requestNodeNames;
- }
+ private Set<NodeWithAttributes> getRequestNodes(Set<String>
requestNodeNames) {
+ return dzManager.logicalTopology().stream()
+ .filter(node -> requestNodeNames.isEmpty() ||
requestNodeNames.contains(node.nodeName()))
+ .collect(toSet());
}
- private CompletableFuture<Void> addMultiNodeOperation(String nodeName,
UUID operationId) {
- CompletableFuture<Void> result = new
CompletableFuture<Void>().orTimeout(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ private CompletableFuture<Void> addMultiNodeOperation(NodeWithAttributes
node, UUID operationId) {
+ CompletableFuture<Void> result = new
CompletableFuture<Void>().orTimeout(DISASTER_RECOVERY_TIMEOUT_MILLIS,
TimeUnit.SECONDS);
Review Comment:
Milliseconds
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java:
##########
@@ -278,10 +277,16 @@ public DisasterRecoveryManager(
nodeLeftListener = new LogicalTopologyEventListener() {
@Override
public void onNodeLeft(LogicalNode leftNode,
LogicalTopologySnapshot newTopology) {
- MultiNodeOperations operations =
operationsByNodeName.get(leftNode.name());
- if (operations != null) {
- operations.completeAllExceptionally(leftNode.name(), new
NodeStoppingException());
- }
+ operationsByNodeId.compute(leftNode.id(), (node, operations)
-> {
+ if (operations != null) {
+ operations.completeAllExceptionally(
+ leftNode.name(),
+ new
NodeNotFoundException(Set.of(leftNode.name()))
Review Comment:
This exception class comes from the compute module, so it seems weird here.
How about adding an exception class specific to disaster recovery, or a common
exception with the same meaning? But it should be not about 'node not found',
but rather 'node left'. Or maybe we already have such an exception class?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java:
##########
@@ -1197,44 +1202,39 @@ private CompletableFuture<Void>
remoteProcessingFuture(DisasterRecoveryRequest r
MultiNodeDisasterRecoveryRequest multiNodeRequest =
(MultiNodeDisasterRecoveryRequest) request;
- Collection<String> actualNodeNames =
getActualNodeNames(multiNodeRequest.nodeNames());
+ Set<NodeWithAttributes> nodes =
getRequestNodes(multiNodeRequest.nodeNames());
- CompletableFuture<?>[] remoteProcessingFutures = actualNodeNames
+ CompletableFuture<?>[] remoteProcessingFutures = nodes
.stream()
- .map(nodeName -> addMultiNodeOperation(nodeName, operationId))
+ .map(node -> addMultiNodeOperation(node, operationId))
.toArray(CompletableFuture[]::new);
- return allOf(remoteProcessingFutures)
- .whenComplete((ignored, e) -> {
- for (String nodeName : actualNodeNames) {
- operationsByNodeName.compute(nodeName, (node,
operations) -> {
- if (operations != null) {
- operations.remove(operationId);
-
- return operations.isEmpty() ? null :
operations;
- }
-
- return null;
- });
- }
- });
+ return allOf(remoteProcessingFutures);
}
/** If request node names is empty, returns all nodes in the logical
topology. */
- private Collection<String> getActualNodeNames(Set<String>
requestNodeNames) {
- if (requestNodeNames.isEmpty()) {
- return dzManager.logicalTopology().stream()
- .map(NodeWithAttributes::nodeName)
- .collect(toSet());
- } else {
- return requestNodeNames;
- }
+ private Set<NodeWithAttributes> getRequestNodes(Set<String>
requestNodeNames) {
+ return dzManager.logicalTopology().stream()
+ .filter(node -> requestNodeNames.isEmpty() ||
requestNodeNames.contains(node.nodeName()))
+ .collect(toSet());
}
- private CompletableFuture<Void> addMultiNodeOperation(String nodeName,
UUID operationId) {
- CompletableFuture<Void> result = new
CompletableFuture<Void>().orTimeout(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ private CompletableFuture<Void> addMultiNodeOperation(NodeWithAttributes
node, UUID operationId) {
+ CompletableFuture<Void> result = new
CompletableFuture<Void>().orTimeout(DISASTER_RECOVERY_TIMEOUT_MILLIS,
TimeUnit.SECONDS);
+
+ operationsByNodeId.compute(node.nodeId(), (nodeId, operations) -> {
+ Set<UUID> nodes = dzManager.logicalTopology().stream()
+ .map(NodeWithAttributes::nodeId)
+ .collect(toSet());
+
+ if (!nodes.contains(nodeId)) {
+ result.completeExceptionally(new IllegalStateException(
Review Comment:
Should it be completed with the same exception class you use on 'node left'
handling?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java:
##########
@@ -1162,7 +1167,7 @@ private CompletableFuture<Void>
processNewRequest(DisasterRecoveryRequest reques
UUID operationId = request.operationId();
- CompletableFuture<Void> operationFuture = new
CompletableFuture<Void>().orTimeout(TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ CompletableFuture<Void> operationFuture = new
CompletableFuture<Void>().orTimeout(DISASTER_RECOVERY_TIMEOUT_MILLIS,
TimeUnit.SECONDS);
Review Comment:
Milliseconds, not seconds
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java:
##########
@@ -1300,7 +1310,7 @@ private CompletableFuture<Void>
forwardDisasterRecoveryRequest(
.revision(revision)
.build();
- return messagingService.invoke(targetNodeName, message,
TimeUnit.SECONDS.toMillis(TIMEOUT_SECONDS))
+ return messagingService.invoke(targetNodeName, message,
TimeUnit.SECONDS.toMillis(DISASTER_RECOVERY_TIMEOUT_MILLIS))
Review Comment:
It's already in millis
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
