IGNITE-28483 Sql. Simplify close method in DistributedQueryManager (#7951)

Author: korlov42

modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java

@@ -1143,7 +1143,17 @@ private void handleError(Throwable ex, String initiatorNode, long fragmentId) {
             } catch (Exception e) {
                 LOG.info("Unable to send error message", e);
 
-                close(CancellationReason.CANCEL);
+                try {
+                    messageService.send(
+                            initiatorNode,
+                            FACTORY.queryCloseMessage()
+                                    .queryId(executionId.queryId())
+                                    .executionToken(executionId.executionToken())
+                                    .build()
+                    );
+                } finally {
+                    close(CancellationReason.CANCEL);
+                }
             }
         }
 
@@ -1327,19 +1337,13 @@ private CompletableFuture<Void> close(CancellationReason reason) {
                 return cancelFut;
             }
 
-            CompletableFuture<Void> start = new CompletableFuture<>();
-
             CompletableFuture<Void> stage;
 
             if (coordinator) {
-                stage = start.thenCompose(ignored -> closeRootNode(reason))
+                stage = closeRootNode(reason)
                         .thenCompose(ignored -> awaitFragmentInitialisationAndClose());
             } else {
-                stage = start.thenCompose(ignored -> messageService.send(coordinatorNodeName, FACTORY.queryCloseMessage()
-                                .queryId(executionId.queryId())
-                                .executionToken(executionId.executionToken())
-                                .build()).exceptionally(ignore -> null))
-                        .thenCompose(ignored -> closeLocalFragments());
+                stage = closeLocalFragments();
             }
 
             stage.whenComplete((r, e) -> {
@@ -1354,8 +1358,6 @@ private CompletableFuture<Void> close(CancellationReason reason) {
                 cancelFut.complete(null);
             }).thenRun(() -> localFragments.forEach(f -> f.context().cancel()));
 
-            start.completeAsync(() -> null, taskExecutor);
-
             return cancelFut;
         }
 

modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java

@@ -1235,48 +1235,6 @@ public void ddlExecutionUpdatesObservableTime() {
         assertThat(txCtx.observableTime(), equalTo(expectedCatalogActivationTimestamp));
     }
 
-    @Test
-    public void coordinatorIgnoresRemoteCloseErrorFromNodeOnCoordinator() throws InterruptedException {
-        ExecutionService execService = executionServices.get(0);
-
-        nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
-
-        var expectedEx = new RuntimeException("Test error");
-        var queryClosed = new CountDownLatch(nodeNames.size() - 1);
-
-        String coordinatorNode = nodeNames.get(0);
-        testCluster.node(coordinatorNode).interceptor((senderNode, msg, original) -> {
-            if (msg instanceof QueryStartRequest) {
-                QueryStartRequest queryStart = (QueryStartRequest) msg;
-
-                String nodeName = senderNode.name();
-                testCluster.node(coordinatorNode).messageService().send(nodeName, new SqlQueryMessagesFactory().queryStartResponse()
-                        .queryId(queryStart.queryId())
-                        .fragmentId(queryStart.fragmentId())
-                        .error(expectedEx)
-                        .build()
-                );
-            } else {
-                original.onMessage(senderNode, msg);
-            }
-
-            if (msg instanceof QueryCloseMessage) {
-                queryClosed.countDown();
-                return CompletableFuture.failedFuture(new RuntimeException("Test exception: failed to close"));
-            } else {
-                return nullCompletedFuture();
-            }
-        });
-
-        SqlOperationContext ctx = createContext();
-        QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
-
-        RuntimeException actualException = assertWillThrow(execService.executePlan(plan, ctx), RuntimeException.class);
-        assertEquals(expectedEx, actualException);
-
-        queryClosed.await();
-    }
-
     @Test
     public void coordinatorIgnoresRemoteCloseErrorOnNode() throws InterruptedException {
         ExecutionService execService = executionServices.get(0);