IGNITE-23499 Integrate cancellation of compute job with sql kill hand…

…ler (#4898)
apache · Dec 17, 2024 · 977f6c4 · 977f6c4
1 parent cd021ff
commit 977f6c4
Show file tree

Hide file tree

Showing 7 changed files with 248 additions and 47 deletions.
diff --git a/modules/compute/build.gradle b/modules/compute/build.gradle
@@ -33,6 +33,7 @@ dependencies {
     implementation project(':ignite-placement-driver-api')
     implementation project(':ignite-binary-tuple')
     implementation project(':ignite-client-common')
+    implementation project(':ignite-sql-engine-api')
     implementation project(':ignite-system-view-api')
     implementation libs.jetbrains.annotations
     implementation libs.fastutil.core

diff --git a/modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeImpl.java b/modules/compute/src/main/java/org/apache/ignite/internal/compute/IgniteComputeImpl.java
@@ -62,6 +62,8 @@
 import org.apache.ignite.internal.placementdriver.PlacementDriver;
 import org.apache.ignite.internal.replicator.TablePartitionId;
 import org.apache.ignite.internal.sql.SqlCommon;
+import org.apache.ignite.internal.sql.engine.api.kill.CancellableOperationType;
+import org.apache.ignite.internal.sql.engine.api.kill.OperationKillHandler;
 import org.apache.ignite.internal.table.IgniteTablesInternal;
 import org.apache.ignite.internal.table.StreamerReceiverRunner;
 import org.apache.ignite.internal.table.TableViewInternal;
@@ -452,6 +454,29 @@ public CompletableFuture<byte[]> runReceiverAsync(byte[] payload, ClusterNode no
                 });
     }
 
+    /** Returns a {@link OperationKillHandler kill handler} for the compute job. */
+    public OperationKillHandler killHandler() {
+        return new OperationKillHandler() {
+            @Override
+            public CompletableFuture<Boolean> cancelAsync(String operationId) {
+                UUID jobId = UUID.fromString(operationId);
+
+                return IgniteComputeImpl.this.cancelAsync(jobId)
+                        .thenApply(res -> res != null ? res : Boolean.FALSE);
+            }
+
+            @Override
+            public boolean local() {
+                return false;
+            }
+
+            @Override
+            public CancellableOperationType type() {
+                return CancellableOperationType.COMPUTE;
+            }
+        };
+    }
+
     @TestOnly
     ComputeComponent computeComponent() {
         return computeComponent;

diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -948,6 +948,8 @@ public class IgniteImpl implements Ignite {
         metaStorageMgr.addElectionListener(catalogCompactionRunner::updateCoordinator);
         this.catalogCompactionRunner = catalogCompactionRunner;
 
+        KillCommandHandler killCommandHandler = new KillCommandHandler(name, logicalTopologyService, clusterSvc.messagingService());
+
         lowWatermark.listen(LowWatermarkEvent.LOW_WATERMARK_CHANGED,
                 params -> catalogCompactionRunner.onLowWatermarkChanged(((ChangeLowWatermarkEventParameters) params).newLowWatermark()));
 
@@ -1085,7 +1087,7 @@ public class IgniteImpl implements Ignite {
                 txManager,
                 lowWatermark,
                 threadPoolsManager.commonScheduler(),
-                new KillCommandHandler(name, logicalTopologyService, clusterSvc.messagingService())
+                killCommandHandler
         );
 
         systemViewManager.register(qryEngine);
@@ -1125,6 +1127,8 @@ public class IgniteImpl implements Ignite {
                 clock
         );
 
+        killCommandHandler.register(((IgniteComputeImpl) compute).killHandler());
+
         authenticationManager = createAuthenticationManager();
 
         ClientConnectorConfiguration clientConnectorConfiguration = nodeConfigRegistry

diff --git a/...pi/src/main/java/org/apache/ignite/internal/sql/engine/api/kill/OperationKillHandler.java b/...pi/src/main/java/org/apache/ignite/internal/sql/engine/api/kill/OperationKillHandler.java
@@ -43,7 +43,7 @@ public interface OperationKillHandler {
      * be performed by the SQL engine.
      *
      * @return {@code True} if the handler can abort operations only on local node,
-     *         {@code false} if the handler can abort operations across the entire cluster (in such a case .
+     *         {@code false} if the handler can abort operations across the entire cluster.
      */
     boolean local();
 

diff --git a/...c/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java b/...c/integrationTest/java/org/apache/ignite/internal/sql/engine/ItComputeSystemViewTest.java
@@ -247,7 +247,8 @@ public CompletableFuture<Void> reduceAsync(TaskExecutionContext taskContext, Map
         }
     }
 
-    private static class InfiniteJob implements ComputeJob<Void, Void> {
+    /** Infinite job. */
+    public static class InfiniteJob implements ComputeJob<Void, Void> {
         @Override
         public @Nullable CompletableFuture<Void> executeAsync(JobExecutionContext context, @Nullable Void arg) {
             while (true) {