From bfb5ee410dc4ada0c56d2a5cfbe27e6f0dd5a7d5 Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Thu, 16 Jul 2026 20:09:58 +0500 Subject: [PATCH 1/2] IGNITE-28900 Control-utility: Add mass query cancellation command --- docs/_docs/tools/control-script.adoc | 55 ++++ .../CommandHandlerParsingTest.java | 8 + .../IgniteControlUtilityTestSuite.java | 2 + .../util/KilAlllCommandsControlShTest.java | 249 ++++++++++++++++++ .../management/kill/KillAllCommand.java | 97 +++++++ .../management/kill/KillAllCommandArg.java | 109 ++++++++ .../internal/management/kill/KillAllTask.java | 233 ++++++++++++++++ .../management/kill/KillAllTaskResult.java | 65 +++++ .../internal/management/kill/KillCommand.java | 1 + .../GridCacheDistributedQueryManager.java | 5 + .../query/GridCacheQueryFutureAdapter.java | 20 +- .../continuous/GridContinuousProcessor.java | 4 +- .../internal/util/GridTestClockTimer.java | 2 + ...mmandHandlerClusterByClassTest_help.output | 8 + ...ndlerClusterByClassWithSSLTest_help.output | 8 + 15 files changed, 861 insertions(+), 5 deletions(-) create mode 100644 modules/control-utility/src/test/java/org/apache/ignite/util/KilAlllCommandsControlShTest.java create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommand.java create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommandArg.java create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllTask.java create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllTaskResult.java diff --git a/docs/_docs/tools/control-script.adoc b/docs/_docs/tools/control-script.adoc index 2c0ed77f4d8b4..9013d6016e3cc 100644 --- a/docs/_docs/tools/control-script.adoc +++ b/docs/_docs/tools/control-script.adoc @@ -453,6 +453,61 @@ For example, to cancel the transactions that have been running for more than 100 control.sh --tx --min-duration 100 --kill ---- +== Mass Cancellation of Queries + +The control script allows you to mass cancel SQL queries, scan queries, and continuous queries that match specific criteria. + +The syntax for the mass cancellation command is as follows: + +[tabs] +-- +tab:Unix[] +[source,shell,subs="verbatim,quotes"] +---- +control.sh --kill all [--node-id ] [--min-duration ] +---- +tab:Windows[] +[source,shell,subs="verbatim,quotes"] +---- +control.bat --kill all [--node-id ] [--min-duration ] +---- +-- + +The following target types are supported: + +[cols="1,3",opts="header"] +|=== +|Target | Description +|`sql`| SQL queries +|`scan`| Scan queries +|`continuous`| Continuous queries +|=== + +Parameters: + +[cols="2,5",opts="header"] +|=== +|Parameter | Description +|--nodeId | Optional. UUID of the originator node to filter targets. +|--min-duration | Optional. Minimum duration in seconds. Only objects that have been running longer than the specified value will be cancelled. +|=== + +Example commands: + +[source, shell] +---- +# Cancel all SQL queries that have been running for more than 60 seconds +control.sh --kill all sql --min-duration 60 + +# Cancel scan queries that have been running for more than 30 seconds on a specific node +control.sh --kill all scan --nodeId --min-duration 30 + +# Cancel continuous queries +control.sh --kill all continuous +---- + +To cancel specific individual objects (without mass filtering) see link:sql-reference/operational-commands[Operational Commands] section. + == Contention Detection in Transactions The `contention` command detects when multiple transactions are in contention to create a lock for the same key. The command is useful if you have long-running or hanging transactions. diff --git a/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java b/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java index 6ae26db4359d8..53cb16046eed3 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java @@ -694,6 +694,14 @@ public void testKillArguments() { assertParseArgsThrows("String representation of \"java.util.UUID\" is exepected", IllegalArgumentException.class, "--kill", "continuous", UUID.randomUUID().toString(), "not_a_uuid"); + + // Kill all command format errors. + assertParseArgsThrows("Argument target_type required.", "--kill", "all"); + assertParseArgsThrows("Can't parse value 'unknown'", "--kill", "all", "unknown"); + assertParseArgsThrows("Argument is invalid: --min-duration", "--kill", "all", "sql", "--min-duration", "-1"); + assertParseArgsThrows("Argument is invalid: --min-duration", "--kill", "all", "sql", "--min-duration", "0"); + assertParseArgsThrows("Argument is invalid: --minDuration is not supported for CONTINUOUS queries", + "--kill", "all", "continuous", "--min-duration", "60"); } /** diff --git a/modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite.java b/modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite.java index 1c5701b6adaf9..5d9805dc14e92 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite.java @@ -42,6 +42,7 @@ import org.apache.ignite.util.GridCommandHandlerWithSslFactoryTest; import org.apache.ignite.util.GridCommandHandlerWithSslTest; import org.apache.ignite.util.GridPersistenceCommandsTest; +import org.apache.ignite.util.KilAlllCommandsControlShTest; import org.apache.ignite.util.KillCommandsControlShTest; import org.junit.runner.RunWith; import org.junit.runners.Suite; @@ -76,6 +77,7 @@ GridCommandHandlerLegacyClientTest.class, KillCommandsControlShTest.class, + KilAlllCommandsControlShTest.class, BaselineEventsLocalTest.class, BaselineEventsRemoteTest.class, diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/KilAlllCommandsControlShTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/KilAlllCommandsControlShTest.java new file mode 100644 index 0000000000000..f2ceca8ca1515 --- /dev/null +++ b/modules/control-utility/src/test/java/org/apache/ignite/util/KilAlllCommandsControlShTest.java @@ -0,0 +1,249 @@ +/* + * 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.util; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import java.util.function.ToIntFunction; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.cache.query.ContinuousQuery; +import org.apache.ignite.cache.query.Query; +import org.apache.ignite.cache.query.QueryCursor; +import org.apache.ignite.cache.query.ScanQuery; +import org.apache.ignite.cache.query.SqlFieldsQuery; +import org.apache.ignite.cache.query.annotations.QuerySqlFunction; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.processors.cache.query.GridCacheDistributedQueryManager; +import org.apache.ignite.internal.util.GridTestClockTimer; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.spi.systemview.view.SqlQueryView; +import org.junit.Test; + +import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_OK; +import static org.apache.ignite.internal.processors.query.running.RunningQueryManager.SQL_QRY_VIEW; +import static org.apache.ignite.testframework.GridTestUtils.assertContains; +import static org.apache.ignite.testframework.GridTestUtils.assertThrows; + +/** + * Test for mass queries cancellation. + */ +public class KilAlllCommandsControlShTest extends GridCommandHandlerClusterByClassAbstractTest { + /** Operations timeout. */ + public static final int TIMEOUT = 10_000; + + /** */ + private static CountDownLatch latch; + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + super.beforeTestsStarted(); + + IgniteCache cache = client.getOrCreateCache( + new CacheConfiguration<>(DEFAULT_CACHE_NAME) + .setIndexedTypes(Integer.class, Integer.class) + .setSqlFunctionClasses(SqlTestFunctions.class)); + + for (int i = 0; i < 1000; i++) + cache.put(i, i); + } + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + latch = new CountDownLatch(1); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + latch.countDown(); + } + + /** */ + @Test + public void testKillAllSql() { + String sql = "SELECT * FROM Integer WHERE latch()"; + + checkKillAll("sql", () -> new SqlFieldsQuery(sql), KilAlllCommandsControlShTest::sqlQueriesCnt); + } + + /** */ + @Test + public void testKillAllScan() { + checkKillAll("scan", () -> new ScanQuery<>().setPageSize(1).setFilter((k, v) -> { + try { + latch.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + + return true; + }), KilAlllCommandsControlShTest::scanQueriesCnt); + } + + /** */ + @Test + public void testKillAllContinuous() { + assertTrue(SERVER_NODE_CNT >= 2); + + client.cache(DEFAULT_CACHE_NAME).query(new ContinuousQuery<>().setLocalListener(evts -> {})); + grid(0).cache(DEFAULT_CACHE_NAME).query(new ContinuousQuery<>().setLocalListener(evts -> {})); + grid(1).cache(DEFAULT_CACHE_NAME).query(new ContinuousQuery<>().setLocalListener(evts -> {})); + + // Kill all queries using --node-id argument. + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "continuous", + "--node-id", grid(0).context().localNodeId().toString())); + + assertEquals(1, client.context().continuous().localRoutineInfos().size()); + assertEquals(0, grid(0).context().continuous().localRoutineInfos().size()); + assertEquals(1, grid(1).context().continuous().localRoutineInfos().size()); + assertEquals(0, client.context().continuous().remoteRoutineInfos().size()); + assertEquals(2, grid(0).context().continuous().remoteRoutineInfos().size()); + assertEquals(1, grid(1).context().continuous().remoteRoutineInfos().size()); + + // Kill all queries without arguments. + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "continuous")); + + assertEquals(0, client.context().continuous().localRoutineInfos().size()); + assertEquals(0, grid(0).context().continuous().localRoutineInfos().size()); + assertEquals(0, grid(1).context().continuous().localRoutineInfos().size()); + assertEquals(0, client.context().continuous().remoteRoutineInfos().size()); + assertEquals(0, grid(0).context().continuous().remoteRoutineInfos().size()); + assertEquals(0, grid(1).context().continuous().remoteRoutineInfos().size()); + } + + /** */ + public void checkKillAll(String target, Supplier> qryFactory, ToIntFunction qryCntProvider) { + try { + assertTrue(SERVER_NODE_CNT >= 2); + + long ts = U.currentTimeMillis(); + GridTestClockTimer.timeSupplier(() -> ts); + + List> curs = new ArrayList<>(); + + curs.add(client.cache(DEFAULT_CACHE_NAME).query(qryFactory.get())); + + for (int i = 0; i < 2; i++) + curs.add(grid(i).cache(DEFAULT_CACHE_NAME).query(qryFactory.get())); + + GridTestClockTimer.timeSupplier(() -> ts + 1001L); + + curs.add(client.cache(DEFAULT_CACHE_NAME).query(qryFactory.get())); + + for (int i = 0; i < 2; i++) + curs.add(grid(i).cache(DEFAULT_CACHE_NAME).query(qryFactory.get())); + + injectTestSystemOut(); + + // Kill all queries using both --min-duration and --node-id arguments. + assertEquals(EXIT_CODE_OK, execute("--kill", "all", target, "--min-duration", "1", + "--node-id", client.context().localNodeId().toString())); + + assertContains(log, testOut.toString(), "Node ID: " + client.context().localNodeId() + " Killed: 1"); + assertContains(log, testOut.toString(), "Total killed: 1"); + + assertEquals(1, qryCntProvider.applyAsInt(client)); + assertEquals(2, qryCntProvider.applyAsInt(grid(0))); + assertEquals(2, qryCntProvider.applyAsInt(grid(1))); + assertThrows(log, () -> curs.get(0).getAll(), Exception.class, ""); + + testOut.reset(); + + // Kill all queries using --min-duration argument. + assertEquals(EXIT_CODE_OK, execute("--kill", "all", target, "--min-duration", "1")); + assertContains(log, testOut.toString(), "Node ID: " + grid(0).context().localNodeId() + " Killed: 1"); + assertContains(log, testOut.toString(), "Node ID: " + grid(1).context().localNodeId() + " Killed: 1"); + assertContains(log, testOut.toString(), "Total killed: 2"); + + assertEquals(1, qryCntProvider.applyAsInt(client)); + assertEquals(1, qryCntProvider.applyAsInt(grid(0))); + assertEquals(1, qryCntProvider.applyAsInt(grid(1))); + + assertThrows(log, () -> curs.get(1).getAll(), Exception.class, ""); + assertThrows(log, () -> curs.get(2).getAll(), Exception.class, ""); + + testOut.reset(); + + // Kill all queries using --node-id argument. + assertEquals(EXIT_CODE_OK, execute("--kill", "all", target, "--node-id", + grid(0).context().localNodeId().toString())); + + assertContains(log, testOut.toString(), "Node ID: " + grid(0).context().localNodeId() + " Killed: 1"); + assertContains(log, testOut.toString(), "Total killed: 1"); + + assertEquals(1, qryCntProvider.applyAsInt(client)); + assertEquals(0, qryCntProvider.applyAsInt(grid(0))); + assertEquals(1, qryCntProvider.applyAsInt(grid(1))); + + assertThrows(log, () -> curs.get(4).getAll(), Exception.class, ""); + + testOut.reset(); + + // Kill all queries without arguments. + assertEquals(EXIT_CODE_OK, execute("--kill", "all", target)); + + assertContains(log, testOut.toString(), "Node ID: " + client.context().localNodeId() + " Killed: 1"); + assertContains(log, testOut.toString(), "Node ID: " + grid(1).context().localNodeId() + " Killed: 1"); + assertContains(log, testOut.toString(), "Total killed: 2"); + + assertEquals(0, qryCntProvider.applyAsInt(client)); + assertEquals(0, qryCntProvider.applyAsInt(grid(0))); + assertEquals(0, qryCntProvider.applyAsInt(grid(1))); + + assertThrows(log, () -> curs.get(3).getAll(), Exception.class, ""); + assertThrows(log, () -> curs.get(5).getAll(), Exception.class, ""); + } + finally { + GridTestClockTimer.timeSupplier(GridTestClockTimer.DFLT_TIME_SUPPLIER); + } + } + + /** */ + private static int sqlQueriesCnt(IgniteEx ignite) { + return F.size(ignite.context().systemView().view(SQL_QRY_VIEW).iterator(), v -> !v.mapQuery()); + } + + /** */ + private static int scanQueriesCnt(IgniteEx ignite) { + return ((GridCacheDistributedQueryManager)ignite.cachex(DEFAULT_CACHE_NAME).context().queries()) + .distributedQueryFutures().size(); + } + + /** */ + public static class SqlTestFunctions { + /** */ + @QuerySqlFunction + public static boolean latch() { + try { + latch.await(TIMEOUT, TimeUnit.MILLISECONDS); + } + catch (InterruptedException ignored) { + return false; + } + + return true; + } + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommand.java new file mode 100644 index 0000000000000..da4c83710e562 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommand.java @@ -0,0 +1,97 @@ +/* + * 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.internal.management.kill; + +import java.util.Collection; +import java.util.Map; +import java.util.function.Consumer; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.management.api.CommandUtils; +import org.apache.ignite.internal.management.api.ComputeCommand; + +/** + * Kill all command for mass cancellation of queries. + */ +public class KillAllCommand implements ComputeCommand> { + /** {@inheritDoc} */ + @Override public String description() { + return "Kill all SQL/scan/continuous queries matching specified criteria"; + } + + /** {@inheritDoc} */ + @Override public Class argClass() { + return KillAllCommandArg.class; + } + + /** {@inheritDoc} */ + @Override public Class taskClass() { + return KillAllTask.class; + } + + /** {@inheritDoc} */ + @Override public Collection nodes(Collection nodes, KillAllCommandArg arg) { + return CommandUtils.nodeOrAll(arg.nodeId(), nodes); + } + + /** {@inheritDoc} */ + @Override public String confirmationPrompt(KillAllCommandArg arg) { + StringBuilder sb = new StringBuilder("Warning: the command will kill all "); + + sb.append(arg.target().toString().toLowerCase()).append(" queries"); + + if (arg.minDuration() != null) + sb.append(" with duration > ").append(arg.minDuration()).append(" seconds"); + + if (arg.nodeId() != null) + sb.append(" on node ").append(arg.nodeId()); + + sb.append("."); + + return sb.toString(); + } + + /** {@inheritDoc} */ + @Override public void printResult( + KillAllCommandArg arg, + Map res, + Consumer printer + ) { + if (res.isEmpty()) { + printer.accept("Nothing found."); + return; + } + + int totalKilled = 0; + int totalFailed = 0; + + for (Map.Entry entry : res.entrySet()) { + ClusterNode node = entry.getKey(); + KillAllTaskResult result = entry.getValue(); + + totalKilled += result.killed(); + totalFailed += result.failed(); + + if (result.killed() > 0 || result.failed() > 0) + printer.accept("Node ID: " + node.id() + " Killed: " + result.killed() + " Failed: " + result.failed()); + } + + printer.accept("\nTotal killed: " + totalKilled + ", failed to kill: " + totalFailed + " " + + arg.target().toString().toLowerCase() + " queries"); + } + +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommandArg.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommandArg.java new file mode 100644 index 0000000000000..60d34a11fc42c --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommandArg.java @@ -0,0 +1,109 @@ +/* + * 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.internal.management.kill; + +import java.util.UUID; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.dto.IgniteDataTransferObject; +import org.apache.ignite.internal.management.api.Argument; +import org.apache.ignite.internal.management.api.Positional; +import org.apache.ignite.internal.util.typedef.internal.A; + +/** + * Argument for --kill all command. + */ +public class KillAllCommandArg extends IgniteDataTransferObject { + /** */ + private static final long serialVersionUID = 0L; + + /** Target type. */ + @Order(0) + @Positional + @Argument(description = "Target type: SQL, SCAN, CONTINUOUS") + TargetType target; + + /** Node ID to filter targets. */ + @Order(1) + @Argument(description = "Originating node ID to filter targets", optional = true) + UUID nodeId; + + /** Minimum duration in seconds. */ + @Order(2) + @Argument(description = "Minimum duration in seconds", example = "60", optional = true) + Long minDuration; + + /** + * Target type enum. + */ + public enum TargetType { + /** */ + SQL, + + /** */ + SCAN, + + /** */ + CONTINUOUS + } + + /** + * @return Target type. + */ + public TargetType target() { + return target; + } + + /** + * @param target Target type. + */ + public void target(TargetType target) { + this.target = target; + } + + /** + * @return Node ID. + */ + public UUID nodeId() { + return nodeId; + } + + /** + * @param nodeId Node ID. + */ + public void nodeId(UUID nodeId) { + this.nodeId = nodeId; + } + + /** + * @return Minimum duration in seconds. + */ + public Long minDuration() { + return minDuration; + } + + /** + * @param minDuration Minimum duration in seconds. + */ + public void minDuration(Long minDuration) { + A.ensure(minDuration == null || minDuration > 0, "--min-duration"); + A.ensure(minDuration == null || target != TargetType.CONTINUOUS, + "--minDuration is not supported for CONTINUOUS queries"); + + this.minDuration = minDuration; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllTask.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllTask.java new file mode 100644 index 0000000000000..6b78a280f0a02 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllTask.java @@ -0,0 +1,233 @@ +/* + * 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.internal.management.kill; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.compute.ComputeJobResult; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.processors.cache.GridCacheContext; +import org.apache.ignite.internal.processors.cache.query.GridCacheDistributedQueryFuture; +import org.apache.ignite.internal.processors.cache.query.GridCacheDistributedQueryManager; +import org.apache.ignite.internal.processors.cache.query.ScanQueryIterator; +import org.apache.ignite.internal.processors.continuous.GridContinuousProcessor; +import org.apache.ignite.internal.processors.query.running.GridRunningQueryInfo; +import org.apache.ignite.internal.processors.task.GridInternal; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.internal.visor.VisorJob; +import org.apache.ignite.internal.visor.VisorMultiNodeTask; +import org.apache.ignite.resources.LoggerResource; + +import static java.util.concurrent.TimeUnit.SECONDS; + +/** + * Task to cancel multiple SQL queries, scan queries, continuous queries based on specified criteria. + */ +@GridInternal +public class KillAllTask extends VisorMultiNodeTask, KillAllTaskResult> { + /** */ + private static final long serialVersionUID = 0L; + + /** {@inheritDoc} */ + @Override protected VisorJob job(KillAllCommandArg arg) { + return new KillAllJob(arg, debug); + } + + /** {@inheritDoc} */ + @Override protected Map reduce0( + List results + ) throws IgniteException { + Map mapRes = new HashMap<>(); + + for (ComputeJobResult result : results) { + if (result.getException() != null) + throw result.getException(); + + KillAllTaskResult data = result.getData(); + + if (data != null && (data.killed() > 0 || data.failed() > 0)) + mapRes.put(result.getNode(), data); + } + + return mapRes; + } + + /** + * Job to cancel multiple targets on a node. + */ + private static class KillAllJob extends VisorJob { + /** */ + private static final long serialVersionUID = 0L; + + /** Injected logger. */ + @LoggerResource + private IgniteLogger log; + + /** + * @param arg Job argument. + * @param debug Debug flag. + */ + protected KillAllJob(KillAllCommandArg arg, boolean debug) { + super(arg, debug); + } + + /** {@inheritDoc} */ + @Override protected KillAllTaskResult run(KillAllCommandArg arg) throws IgniteException { + switch (arg.target()) { + case SQL: + return cancelSqlQueries(arg); + + case SCAN: + return cancelScanQueries(arg); + + case CONTINUOUS: + return cancelContinuousQueries(arg); + + default: + throw new IgniteException("Unknown target type: " + arg.target()); + } + } + + /** + * Cancel SQL queries matching criteria. + * + * @param arg Command argument. + * @return Result. + */ + private KillAllTaskResult cancelSqlQueries(KillAllCommandArg arg) { + List qrys = ignite.context().query().runningQueryManager().runningSqlQueries(); + + qrys.removeIf(qry -> qry.mapQuery() || (arg.minDuration() != null + && U.currentTimeMillis() - qry.startTime() <= SECONDS.toMillis(arg.minDuration()))); + + for (GridRunningQueryInfo qry : qrys) + ignite.context().query().runningQueryManager().cancelLocalQuery(qry.id()); + + return new KillAllTaskResult(qrys.size(), 0); + } + + /** + * Cancel scan queries matching criteria. + * + * @param arg Command argument. + * @return Result. + */ + private KillAllTaskResult cancelScanQueries(KillAllCommandArg arg) { + long ts = arg.minDuration() == null ? 0 : U.currentTimeMillis() - SECONDS.toMillis(arg.minDuration()); + int killed = 0; + int failed = 0; + + for (GridCacheContext cctx : ignite.context().cache().context().cacheContexts()) { + // Scan queries can be registered in multiple structures. There is no single registry for all scan + // queries. To properly cancel a scan query, all relevant structures must be analyzed: + // - cctx.queries().localQueryIterators() - iterators for local-only scans and local parts of + // distributed scans (on initiator node). If initiator is not affinity node, there will be no + // local iterator for distributed scan. + // - cctx.queries().distributedQueryFutures() - futures for distributed scans (on initiator node). + // For local-only data (REPLICATED cache or local scans), there will be no distributed future. + // - cctx.queries().queryIterators() - remote iterators for distributed scans (on affinity nodes). + // Keyed by originator node ID, each entry contains map of requests to iterators. + // Correct way to kill scan (see also GridCacheDistributedQueryManager.scanQueryDistributed -> + // new GridCloseableIteratorAdapter.onClose) + // - Close local iterator (removes iterator from localQueryIterators()) + // - Cancel distributed future (removes future from distrivuted future list, completes future, + // sends cancel to remote nodes, removes iterator from queryIterrators() on remote nodes) + GridCacheDistributedQueryManager mgr = (GridCacheDistributedQueryManager)cctx.queries(); + + // Kill local-only scans and local part of distributed scans. + for (ScanQueryIterator locIter : mgr.localQueryIterators()) { + if (ts > 0 && locIter.startTime() >= ts) + continue; + + try { + locIter.close(); + + killed++; + } + catch (IgniteCheckedException e) { + log.warning("Failed to close local iterator for scan query", e); + + failed++; + } + } + + // Kill remote part of distributed scans. + for (GridCacheDistributedQueryFuture fut : mgr.distributedQueryFutures()) { + if (ts > 0 && fut.startTime() >= ts) + continue; + + try { + fut.cancel(); + + if (!cctx.affinityNode()) // For affinity nodes killed count is already incremented by locIter. + killed++; + } + catch (IgniteCheckedException e) { + log.warning("Failed to cancel distributed query future for scan query", e); + + failed++; + } + } + } + + return new KillAllTaskResult(killed, failed); + } + + /** + * Cancel continuous queries matching criteria. + * + * @param arg Command argument. + * @return Result. + */ + private KillAllTaskResult cancelContinuousQueries(KillAllCommandArg arg) { + GridContinuousProcessor proc = ignite.context().continuous(); + + List> futs = new ArrayList<>(); + + for (Map.Entry e : proc.localRoutineInfos().entrySet()) { + if (arg.nodeId == null || arg.nodeId.equals(e.getValue().nodeId())) + futs.add(proc.stopRoutine(e.getKey())); + } + + int killed = 0; + int failed = 0; + + for (IgniteInternalFuture fut : futs) { + try { + fut.get(); + + killed++; + } + catch (IgniteCheckedException e) { + log.warning("Failed to stop continuous query routine", e); + + failed++; + } + } + + return new KillAllTaskResult(killed, failed); + } + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllTaskResult.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllTaskResult.java new file mode 100644 index 0000000000000..8b61bc509a8cf --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllTaskResult.java @@ -0,0 +1,65 @@ +/* + * 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.internal.management.kill; + +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.dto.IgniteDataTransferObject; + +/** + * Task result. + */ +public class KillAllTaskResult extends IgniteDataTransferObject { + /** */ + private static final long serialVersionUID = 0L; + + /** Number of killed targets. */ + @Order(0) + int killed; + + /** Number of failures. */ + @Order(1) + int failed; + + /** */ + public KillAllTaskResult() { + // No-op. + } + + /** + * @param killed Number of killed targets. + * @param failed Number of failures. + */ + public KillAllTaskResult(int killed, int failed) { + this.killed = killed; + this.failed = failed; + } + + /** + * @return Number of killed targets. + */ + public int killed() { + return killed; + } + + /** + * @return Number of failures. + */ + public int failed() { + return failed; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillCommand.java index 5dcbacaa93420..0abc330d95295 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillCommand.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillCommand.java @@ -24,6 +24,7 @@ public class KillCommand extends CommandRegistryImpl { /** */ public KillCommand() { super( + new KillAllCommand(), new KillComputeCommand(), new KillServiceCommand(), new KillTransactionCommand(), diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java index 583d3006bc4cf..9ebdcbbc25114 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryManager.java @@ -532,6 +532,11 @@ private CacheQueryFuture queryDistributed(GridCacheQueryBean qry, final Colle return fut; } + /** */ + public Collection> distributedQueryFutures() { + return futs.values(); + } + /** {@inheritDoc} */ @SuppressWarnings({"unchecked"}) @Override public GridCloseableIterator scanQueryDistributed(final CacheQuery qry, diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java index 2dcc4aa59da05..7cbe82234e9f3 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryFutureAdapter.java @@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteLogger; +import org.apache.ignite.cache.query.QueryCancelledException; import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException; import org.apache.ignite.internal.cache.query.index.IndexQueryResultMeta; import org.apache.ignite.internal.processors.cache.CacheObjectUtils; @@ -76,6 +77,9 @@ public abstract class GridCacheQueryFutureAdapter extends GridFutureAda /** */ private final IgniteUuid timeoutId = IgniteUuid.randomUuid(); + /** */ + private long startTime; + /** */ private long endTime; @@ -103,7 +107,7 @@ protected GridCacheQueryFutureAdapter(GridCacheContext cctx, GridCacheQuer if (log == null) log = U.logger(cctx.kernalContext(), logRef, GridCacheQueryFutureAdapter.class); - long startTime = U.currentTimeMillis(); + startTime = U.currentTimeMillis(); long timeout = qry.query().timeout(); capacity = query().query().limit(); @@ -188,10 +192,15 @@ public GridCacheContext cacheContext() { * @throws IgniteCheckedException If future is done with an error. */ private void checkError() throws IgniteCheckedException { - if (error() != null) { + Throwable err = error(); + + if (err == null && isCancelled()) + err = new QueryCancelledException("Query was cancelled"); + + if (err != null) { clear(); - throw U.cast(error()); + throw U.cast(err); } } @@ -386,6 +395,11 @@ void clear() { return timeoutId; } + /** Query start time. */ + public long startTime() { + return startTime; + } + /** {@inheritDoc} */ @Override public long endTime() { return endTime; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java index fecd065ff7551..fb8a55b73d8d6 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java @@ -331,12 +331,12 @@ private void cancelFutures(IgniteCheckedException e) { } /** */ - Map remoteRoutineInfos() { + public Map remoteRoutineInfos() { return Collections.unmodifiableMap(rmtInfos); } /** */ - Map localRoutineInfos() { + public Map localRoutineInfos() { return Collections.unmodifiableMap(locInfos); } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/GridTestClockTimer.java b/modules/core/src/test/java/org/apache/ignite/internal/util/GridTestClockTimer.java index eba1e48420097..bdeb7e3065d30 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/util/GridTestClockTimer.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/util/GridTestClockTimer.java @@ -59,6 +59,8 @@ public static boolean startTestTimer() { */ public static void timeSupplier(LongSupplier timeSupplier) { GridTestClockTimer.timeSupplier = timeSupplier; + + update(); } /** diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output index 82d7b734fc049..ae200a1fb37a3 100644 --- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output +++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output @@ -162,6 +162,14 @@ This utility can do the following commands: Parameters: new_limit - Decimal value to change re-encryption rate limit (MB/s). + Kill all SQL/scan/continuous queries matching specified criteria: + control.(sh|bat) --kill all SQL|SCAN|CONTINUOUS [--node-id node_id] [--min-duration 60] + + Parameters: + target - Target type: SQL, SCAN, CONTINUOUS. + --node-id node_id - Originating node ID to filter targets. + --min-duration 60 - Minimum duration in seconds. + Kill compute task by session id: control.(sh|bat) --kill compute session_id diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output index d8b764eb4e6b5..c54da9e384084 100644 --- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output +++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output @@ -162,6 +162,14 @@ This utility can do the following commands: Parameters: new_limit - Decimal value to change re-encryption rate limit (MB/s). + Kill all SQL/scan/continuous queries matching specified criteria: + control.(sh|bat) --kill all SQL|SCAN|CONTINUOUS [--node-id node_id] [--min-duration 60] + + Parameters: + target - Target type: SQL, SCAN, CONTINUOUS. + --node-id node_id - Originating node ID to filter targets. + --min-duration 60 - Minimum duration in seconds. + Kill compute task by session id: control.(sh|bat) --kill compute session_id From c6344c25d4a43f54d7a986360a31682df2846183 Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Fri, 31 Jul 2026 11:55:24 +0300 Subject: [PATCH 2/2] IGNITE-28900 Control-utility: Add mass query cancellation command --- .../commandline/CommandHandlerParsingTest.java | 13 ++++++++++++- .../management/kill/KillAllCommandArg.java | 15 ++++++++++++++- ...idCommandHandlerClusterByClassTest_help.output | 4 +++- ...ndHandlerClusterByClassWithSSLTest_help.output | 4 +++- 4 files changed, 32 insertions(+), 4 deletions(-) diff --git a/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java b/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java index 53cb16046eed3..a369d930a2913 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java @@ -70,6 +70,8 @@ import org.apache.ignite.internal.management.encryption.EncryptionChangeMasterKeyCommand; import org.apache.ignite.internal.management.encryption.EncryptionCommand; import org.apache.ignite.internal.management.event.EventCommand; +import org.apache.ignite.internal.management.kill.KillAllCommand; +import org.apache.ignite.internal.management.kill.KillAllCommandArg; import org.apache.ignite.internal.management.kill.KillCommand; import org.apache.ignite.internal.management.meta.MetaCommand; import org.apache.ignite.internal.management.meta.MetaRemoveCommand; @@ -478,6 +480,13 @@ else if (cmd.getClass() == CacheClearCommand.class) { arg = (A)a; } + else if (cmd.getClass() == KillAllCommand.class) { + KillAllCommandArg a = new KillAllCommandArg(); + + a.target(KillAllCommandArg.TargetType.SQL); + + arg = (A)a; + } else arg = cmd.argClass().newInstance(); @@ -523,6 +532,8 @@ else if (cmd.getClass() == MetaUpdateCommand.class) return; else if (cmd.getClass() == MetaRemoveCommand.class) cmdText = F.concat(cmdText, "--typeId", "1"); + else if (cmd.getClass() == KillAllCommand.class) + cmdText = F.concat(cmdText, "SQL"); args = parseArgs(asList(cmdText)); @@ -696,7 +707,7 @@ public void testKillArguments() { "--kill", "continuous", UUID.randomUUID().toString(), "not_a_uuid"); // Kill all command format errors. - assertParseArgsThrows("Argument target_type required.", "--kill", "all"); + assertParseArgsThrows("Argument target required.", "--kill", "all"); assertParseArgsThrows("Can't parse value 'unknown'", "--kill", "all", "unknown"); assertParseArgsThrows("Argument is invalid: --min-duration", "--kill", "all", "sql", "--min-duration", "-1"); assertParseArgsThrows("Argument is invalid: --min-duration", "--kill", "all", "sql", "--min-duration", "0"); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommandArg.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommandArg.java index 60d34a11fc42c..9ac79732bcb2f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommandArg.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommandArg.java @@ -21,6 +21,7 @@ import org.apache.ignite.internal.Order; import org.apache.ignite.internal.dto.IgniteDataTransferObject; import org.apache.ignite.internal.management.api.Argument; +import org.apache.ignite.internal.management.api.EnumDescription; import org.apache.ignite.internal.management.api.Positional; import org.apache.ignite.internal.util.typedef.internal.A; @@ -34,7 +35,19 @@ public class KillAllCommandArg extends IgniteDataTransferObject { /** Target type. */ @Order(0) @Positional - @Argument(description = "Target type: SQL, SCAN, CONTINUOUS") + @Argument() + @EnumDescription( + names = { + "SQL", + "SCAN", + "CONTUNUOUS" + }, + descriptions = { + "SQL queries", + "SCAN queries", + "CONTUNUOUS queries" + } + ) TargetType target; /** Node ID to filter targets. */ diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output index ae200a1fb37a3..86f715387860a 100644 --- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output +++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output @@ -166,7 +166,9 @@ This utility can do the following commands: control.(sh|bat) --kill all SQL|SCAN|CONTINUOUS [--node-id node_id] [--min-duration 60] Parameters: - target - Target type: SQL, SCAN, CONTINUOUS. + SQL - SQL queries. + SCAN - SCAN queries. + CONTUNUOUS - CONTUNUOUS queries. --node-id node_id - Originating node ID to filter targets. --min-duration 60 - Minimum duration in seconds. diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output index c54da9e384084..bcc1ffd002d6b 100644 --- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output +++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output @@ -166,7 +166,9 @@ This utility can do the following commands: control.(sh|bat) --kill all SQL|SCAN|CONTINUOUS [--node-id node_id] [--min-duration 60] Parameters: - target - Target type: SQL, SCAN, CONTINUOUS. + SQL - SQL queries. + SCAN - SCAN queries. + CONTUNUOUS - CONTUNUOUS queries. --node-id node_id - Originating node ID to filter targets. --min-duration 60 - Minimum duration in seconds.