From bfb5ee410dc4ada0c56d2a5cfbe27e6f0dd5a7d5 Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Thu, 16 Jul 2026 20:09:58 +0500 Subject: [PATCH 1/5] 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/5] 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. From bbc1f449bc8b18cbfd3a8d4563a9f081a74bb8fd Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Wed, 19 Aug 2026 10:58:24 +0300 Subject: [PATCH 3/5] IGNITE-28900 Control-utility: Add mass query cancellation command --- docs/_docs/tools/control-script.adoc | 6 +- .../util/KilAlllCommandsControlShTest.java | 90 ++++++++++++++++++- .../management/kill/KillAllCommand.java | 2 +- .../management/kill/KillAllCommandArg.java | 5 ++ .../internal/management/kill/KillAllTask.java | 52 ++++++++++- ...mmandHandlerClusterByClassTest_help.output | 5 +- ...ndlerClusterByClassWithSSLTest_help.output | 5 +- 7 files changed, 155 insertions(+), 10 deletions(-) diff --git a/docs/_docs/tools/control-script.adoc b/docs/_docs/tools/control-script.adoc index 75d00c170f821..fd96397f26197 100644 --- a/docs/_docs/tools/control-script.adoc +++ b/docs/_docs/tools/control-script.adoc @@ -455,7 +455,7 @@ 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 control script allows you to mass cancel SQL queries, scan queries, index queries and continuous queries that match specific criteria. The syntax for the mass cancellation command is as follows: @@ -480,6 +480,7 @@ The following target types are supported: |Target | Description |`sql`| SQL queries |`scan`| Scan queries +|`index`| Index queries |`continuous`| Continuous queries |=== @@ -502,6 +503,9 @@ 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 index queries on a specific node +control.sh --kill all index --nodeId + # Cancel continuous queries control.sh --kill all continuous ---- 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 index f2ceca8ca1515..774f17bcd8518 100644 --- 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 @@ -19,12 +19,16 @@ import java.util.ArrayList; import java.util.List; +import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; import java.util.function.ToIntFunction; +import javax.cache.Cache; import org.apache.ignite.IgniteCache; import org.apache.ignite.cache.query.ContinuousQuery; +import org.apache.ignite.cache.query.FieldsQueryCursor; +import org.apache.ignite.cache.query.IndexQuery; import org.apache.ignite.cache.query.Query; import org.apache.ignite.cache.query.QueryCursor; import org.apache.ignite.cache.query.ScanQuery; @@ -33,12 +37,14 @@ 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.processors.cache.query.GridCacheQueryType; 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.events.EventType.EVT_CACHE_OBJECT_PUT; 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; @@ -51,6 +57,9 @@ public class KilAlllCommandsControlShTest extends GridCommandHandlerClusterByCla /** Operations timeout. */ public static final int TIMEOUT = 10_000; + /** */ + private static final int ENTRIES_CNT = 1_000; + /** */ private static CountDownLatch latch; @@ -63,7 +72,7 @@ public class KilAlllCommandsControlShTest extends GridCommandHandlerClusterByCla .setIndexedTypes(Integer.class, Integer.class) .setSqlFunctionClasses(SqlTestFunctions.class)); - for (int i = 0; i < 1000; i++) + for (int i = 0; i < ENTRIES_CNT; i++) cache.put(i, i); } @@ -102,6 +111,21 @@ public void testKillAllScan() { }), KilAlllCommandsControlShTest::scanQueriesCnt); } + /** */ + @Test + public void testKillAllIndex() { + checkKillAll("index", () -> new IndexQuery<>(Integer.class).setFilter((k, v) -> { + try { + latch.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + + return true; + }).setPageSize(1), KilAlllCommandsControlShTest::indexQueriesCnt); + } + /** */ @Test public void testKillAllContinuous() { @@ -133,6 +157,60 @@ public void testKillAllContinuous() { assertEquals(0, grid(1).context().continuous().remoteRoutineInfos().size()); } + /** */ + @Test + public void testKillUnrelated() { + try ( + QueryCursor> cur = client.cache(DEFAULT_CACHE_NAME) + .query(new IndexQuery(Integer.class).setPageSize(1)) + ) { + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "scan")); + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "sql")); + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "continuous")); + assertEquals(ENTRIES_CNT, cur.getAll().size()); + } + + try ( + QueryCursor> cur = client.cache(DEFAULT_CACHE_NAME) + .query(new ScanQuery().setPageSize(1)) + ) { + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "index")); + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "sql")); + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "continuous")); + assertEquals(ENTRIES_CNT, cur.getAll().size()); + } + + try ( + FieldsQueryCursor cur = client.cache(DEFAULT_CACHE_NAME) + .query(new SqlFieldsQuery("SELECT * FROM Integer").setPageSize(1)) + ) { + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "scan")); + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "index")); + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "continuous")); + assertEquals(ENTRIES_CNT, cur.getAll().size()); + } + } + + /** */ + @Test + public void testRemoteListen() { + UUID evtLsnrId = client.events().remoteListen((nodeId, evt) -> true, evt -> true, EVT_CACHE_OBJECT_PUT); + UUID msgLsnrId = client.message().remoteListen("topic", (nodeId, msg) -> true); + + try { + assertEquals(EXIT_CODE_OK, execute("--kill", "all", "continuous")); + + boolean evtLsnrAlive = client.context().continuous().localRoutineInfos().containsKey(evtLsnrId); + boolean msgLsnrAlive = client.context().continuous().localRoutineInfos().containsKey(msgLsnrId); + + assertTrue("Listeners were stopped", evtLsnrAlive && msgLsnrAlive); + } + finally { + client.events().stopRemoteListen(evtLsnrId); + client.message().stopRemoteListen(msgLsnrId); + } + } + /** */ public void checkKillAll(String target, Supplier> qryFactory, ToIntFunction qryCntProvider) { try { @@ -227,8 +305,14 @@ private static int sqlQueriesCnt(IgniteEx ignite) { /** */ private static int scanQueriesCnt(IgniteEx ignite) { - return ((GridCacheDistributedQueryManager)ignite.cachex(DEFAULT_CACHE_NAME).context().queries()) - .distributedQueryFutures().size(); + return (int)((GridCacheDistributedQueryManager)ignite.cachex(DEFAULT_CACHE_NAME).context().queries()) + .distributedQueryFutures().stream().filter(f -> f.query().query().type() == GridCacheQueryType.SCAN).count(); + } + + /** */ + private static int indexQueriesCnt(IgniteEx ignite) { + return (int)((GridCacheDistributedQueryManager)ignite.cachex(DEFAULT_CACHE_NAME).context().queries()) + .distributedQueryFutures().stream().filter(f -> f.query().query().type() == GridCacheQueryType.INDEX).count(); } /** */ 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 index da4c83710e562..853dd8a4a52a1 100644 --- 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 @@ -30,7 +30,7 @@ public class KillAllCommand implements ComputeCommand> { /** {@inheritDoc} */ @Override public String description() { - return "Kill all SQL/scan/continuous queries matching specified criteria"; + return "Kill all SQL/scan/index/continuous queries matching specified criteria"; } /** {@inheritDoc} */ 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 9ac79732bcb2f..83f3ca425df91 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 @@ -40,11 +40,13 @@ public class KillAllCommandArg extends IgniteDataTransferObject { names = { "SQL", "SCAN", + "INDEX", "CONTUNUOUS" }, descriptions = { "SQL queries", "SCAN queries", + "INDEX queries", "CONTUNUOUS queries" } ) @@ -70,6 +72,9 @@ public enum TargetType { /** */ SCAN, + /** */ + INDEX, + /** */ CONTINUOUS } 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 index 6b78a280f0a02..693acccfc0211 100644 --- 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 @@ -31,7 +31,9 @@ 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.GridCacheQueryType; import org.apache.ignite.internal.processors.cache.query.ScanQueryIterator; +import org.apache.ignite.internal.processors.continuous.ContinousRoutineLocalInfo; 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; @@ -102,6 +104,9 @@ protected KillAllJob(KillAllCommandArg arg, boolean debug) { case SCAN: return cancelScanQueries(arg); + case INDEX: + return cancelIndexQueries(arg); + case CONTINUOUS: return cancelContinuousQueries(arg); @@ -175,6 +180,9 @@ private KillAllTaskResult cancelScanQueries(KillAllCommandArg arg) { // Kill remote part of distributed scans. for (GridCacheDistributedQueryFuture fut : mgr.distributedQueryFutures()) { + if (fut.query().query().type() != GridCacheQueryType.SCAN) + continue; + if (ts > 0 && fut.startTime() >= ts) continue; @@ -195,6 +203,45 @@ private KillAllTaskResult cancelScanQueries(KillAllCommandArg arg) { return new KillAllTaskResult(killed, failed); } + /** + * Cancel index queries matching criteria. + * + * @param arg Command argument. + * @return Result. + */ + private KillAllTaskResult cancelIndexQueries(KillAllCommandArg arg) { + long ts = arg.minDuration() == null ? 0 : U.currentTimeMillis() - SECONDS.toMillis(arg.minDuration()); + int killed = 0; + int failed = 0; + + // Distributed Index queries are registered in distributedQueryFutures structure (both local and remote part). + // But local-only queries are not registered in any structure at all, so local queries are not killable. + for (GridCacheContext cctx : ignite.context().cache().context().cacheContexts()) { + GridCacheDistributedQueryManager mgr = (GridCacheDistributedQueryManager)cctx.queries(); + + for (GridCacheDistributedQueryFuture fut : mgr.distributedQueryFutures()) { + if (fut.query().query().type() != GridCacheQueryType.INDEX) + continue; + + if (ts > 0 && fut.startTime() >= ts) + continue; + + try { + fut.cancel(); + + killed++; + } + catch (IgniteCheckedException e) { + log.warning("Failed to cancel distributed query future for index query", e); + + failed++; + } + } + } + + return new KillAllTaskResult(killed, failed); + } + /** * Cancel continuous queries matching criteria. * @@ -206,7 +253,10 @@ private KillAllTaskResult cancelContinuousQueries(KillAllCommandArg arg) { List> futs = new ArrayList<>(); - for (Map.Entry e : proc.localRoutineInfos().entrySet()) { + for (Map.Entry e : proc.localRoutineInfos().entrySet()) { + if (!e.getValue().handler().isQuery()) + continue; + if (arg.nodeId == null || arg.nodeId.equals(e.getValue().nodeId())) futs.add(proc.stopRoutine(e.getKey())); } 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 ffd0d935e0def..8552880d3664d 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,12 +162,13 @@ 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] + Kill all SQL/scan/index/continuous queries matching specified criteria: + control.(sh|bat) --kill all SQL|SCAN|INDEX|CONTINUOUS [--node-id node_id] [--min-duration 60] Parameters: SQL - SQL queries. SCAN - SCAN queries. + INDEX - INDEX 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 eac5595246a60..f2c67d82a2a64 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,12 +162,13 @@ 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] + Kill all SQL/scan/index/continuous queries matching specified criteria: + control.(sh|bat) --kill all SQL|SCAN|INDEX|CONTINUOUS [--node-id node_id] [--min-duration 60] Parameters: SQL - SQL queries. SCAN - SCAN queries. + INDEX - INDEX queries. CONTUNUOUS - CONTUNUOUS queries. --node-id node_id - Originating node ID to filter targets. --min-duration 60 - Minimum duration in seconds. From be131ab012902fc37adabfe504144beadcf378a0 Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Wed, 19 Aug 2026 11:13:46 +0300 Subject: [PATCH 4/5] IGNITE-28900 Control-utility: Add mass query cancellation command --- .../ignite/testsuites/IgniteControlUtilityTestSuite.java | 4 ++-- ...trolShTest.java => KillAlllCommandsControlShTest.java} | 8 ++++---- .../ignite/internal/management/kill/KillAllCommand.java | 1 - .../internal/management/kill/KillAllCommandArg.java | 6 ++++-- .../GridCommandHandlerClusterByClassTest_help.output | 4 ++-- ...ridCommandHandlerClusterByClassWithSSLTest_help.output | 4 ++-- 6 files changed, 14 insertions(+), 13 deletions(-) rename modules/control-utility/src/test/java/org/apache/ignite/util/{KilAlllCommandsControlShTest.java => KillAlllCommandsControlShTest.java} (97%) 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 5d9805dc14e92..e643252c4443d 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,7 +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.KillAlllCommandsControlShTest; import org.apache.ignite.util.KillCommandsControlShTest; import org.junit.runner.RunWith; import org.junit.runners.Suite; @@ -77,7 +77,7 @@ GridCommandHandlerLegacyClientTest.class, KillCommandsControlShTest.class, - KilAlllCommandsControlShTest.class, + KillAlllCommandsControlShTest.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/KillAlllCommandsControlShTest.java similarity index 97% rename from modules/control-utility/src/test/java/org/apache/ignite/util/KilAlllCommandsControlShTest.java rename to modules/control-utility/src/test/java/org/apache/ignite/util/KillAlllCommandsControlShTest.java index 774f17bcd8518..678262c030091 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/util/KilAlllCommandsControlShTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/util/KillAlllCommandsControlShTest.java @@ -53,7 +53,7 @@ /** * Test for mass queries cancellation. */ -public class KilAlllCommandsControlShTest extends GridCommandHandlerClusterByClassAbstractTest { +public class KillAlllCommandsControlShTest extends GridCommandHandlerClusterByClassAbstractTest { /** Operations timeout. */ public static final int TIMEOUT = 10_000; @@ -93,7 +93,7 @@ public class KilAlllCommandsControlShTest extends GridCommandHandlerClusterByCla public void testKillAllSql() { String sql = "SELECT * FROM Integer WHERE latch()"; - checkKillAll("sql", () -> new SqlFieldsQuery(sql), KilAlllCommandsControlShTest::sqlQueriesCnt); + checkKillAll("sql", () -> new SqlFieldsQuery(sql), KillAlllCommandsControlShTest::sqlQueriesCnt); } /** */ @@ -108,7 +108,7 @@ public void testKillAllScan() { } return true; - }), KilAlllCommandsControlShTest::scanQueriesCnt); + }), KillAlllCommandsControlShTest::scanQueriesCnt); } /** */ @@ -123,7 +123,7 @@ public void testKillAllIndex() { } return true; - }).setPageSize(1), KilAlllCommandsControlShTest::indexQueriesCnt); + }).setPageSize(1), KillAlllCommandsControlShTest::indexQueriesCnt); } /** */ 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 index 853dd8a4a52a1..a3b590b28a0d4 100644 --- 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 @@ -93,5 +93,4 @@ public class KillAllCommand implements ComputeCommand Date: Wed, 19 Aug 2026 11:26:08 +0300 Subject: [PATCH 5/5] IGNITE-28900 Control-utility: Add mass query cancellation command --- .../ignite/testsuites/IgniteControlUtilityTestSuite.java | 4 ++-- ...ntrolShTest.java => KillAllCommandsControlShTest.java} | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) rename modules/control-utility/src/test/java/org/apache/ignite/util/{KillAlllCommandsControlShTest.java => KillAllCommandsControlShTest.java} (98%) 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 e643252c4443d..6528d919f6105 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,7 +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.KillAlllCommandsControlShTest; +import org.apache.ignite.util.KillAllCommandsControlShTest; import org.apache.ignite.util.KillCommandsControlShTest; import org.junit.runner.RunWith; import org.junit.runners.Suite; @@ -77,7 +77,7 @@ GridCommandHandlerLegacyClientTest.class, KillCommandsControlShTest.class, - KillAlllCommandsControlShTest.class, + KillAllCommandsControlShTest.class, BaselineEventsLocalTest.class, BaselineEventsRemoteTest.class, diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/KillAlllCommandsControlShTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/KillAllCommandsControlShTest.java similarity index 98% rename from modules/control-utility/src/test/java/org/apache/ignite/util/KillAlllCommandsControlShTest.java rename to modules/control-utility/src/test/java/org/apache/ignite/util/KillAllCommandsControlShTest.java index 678262c030091..607b97bcf0b55 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/util/KillAlllCommandsControlShTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/util/KillAllCommandsControlShTest.java @@ -53,7 +53,7 @@ /** * Test for mass queries cancellation. */ -public class KillAlllCommandsControlShTest extends GridCommandHandlerClusterByClassAbstractTest { +public class KillAllCommandsControlShTest extends GridCommandHandlerClusterByClassAbstractTest { /** Operations timeout. */ public static final int TIMEOUT = 10_000; @@ -93,7 +93,7 @@ public class KillAlllCommandsControlShTest extends GridCommandHandlerClusterByCl public void testKillAllSql() { String sql = "SELECT * FROM Integer WHERE latch()"; - checkKillAll("sql", () -> new SqlFieldsQuery(sql), KillAlllCommandsControlShTest::sqlQueriesCnt); + checkKillAll("sql", () -> new SqlFieldsQuery(sql), KillAllCommandsControlShTest::sqlQueriesCnt); } /** */ @@ -108,7 +108,7 @@ public void testKillAllScan() { } return true; - }), KillAlllCommandsControlShTest::scanQueriesCnt); + }), KillAllCommandsControlShTest::scanQueriesCnt); } /** */ @@ -123,7 +123,7 @@ public void testKillAllIndex() { } return true; - }).setPageSize(1), KillAlllCommandsControlShTest::indexQueriesCnt); + }).setPageSize(1), KillAllCommandsControlShTest::indexQueriesCnt); } /** */