This is an automated email from the ASF dual-hosted git repository.
alex-plekhanov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 174bde616a1 IGNITE-28900 Control-utility: Add mass query cancellation
command (#13420)
174bde616a1 is described below
commit 174bde616a1be5af556582ef2f67b34cf6d1be85
Author: Aleksey Plekhanov <[email protected]>
AuthorDate: Thu Aug 20 17:10:06 2026 +0300
IGNITE-28900 Control-utility: Add mass query cancellation command (#13420)
---
docs/_docs/tools/control-script.adoc | 59 ++++
.../commandline/CommandHandlerParsingTest.java | 19 ++
.../testsuites/IgniteControlUtilityTestSuite.java | 2 +
.../ignite/util/KillAllCommandsControlShTest.java | 333 +++++++++++++++++++++
.../internal/management/kill/KillAllCommand.java | 96 ++++++
.../management/kill/KillAllCommandArg.java | 129 ++++++++
.../internal/management/kill/KillAllTask.java | 287 ++++++++++++++++++
.../management/kill/KillAllTaskResult.java | 65 ++++
.../internal/management/kill/KillCommand.java | 1 +
.../query/GridCacheDistributedQueryManager.java | 5 +
.../cache/query/GridCacheQueryFutureAdapter.java | 20 +-
.../continuous/CacheContinuousQueryHandler.java | 7 +
.../continuous/GridContinuousProcessor.java | 4 +-
.../ignite/internal/util/GridTestClockTimer.java | 2 +
...ridCommandHandlerClusterByClassTest_help.output | 11 +
...andHandlerClusterByClassWithSSLTest_help.output | 11 +
16 files changed, 1046 insertions(+), 5 deletions(-)
diff --git a/docs/_docs/tools/control-script.adoc
b/docs/_docs/tools/control-script.adoc
index a4820d20a8c..fd96397f261 100644
--- a/docs/_docs/tools/control-script.adoc
+++ b/docs/_docs/tools/control-script.adoc
@@ -453,6 +453,65 @@ 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, index
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 <target> [--node-id <nodeId>] [--min-duration <seconds>]
+----
+tab:Windows[]
+[source,shell,subs="verbatim,quotes"]
+----
+control.bat --kill all <target> [--node-id <nodeId>] [--min-duration <seconds>]
+----
+--
+
+The following target types are supported:
+
+[cols="1,3",opts="header"]
+|===
+|Target | Description
+|`sql`| SQL queries
+|`scan`| Scan queries
+|`index`| Index queries
+|`continuous`| Continuous queries
+|===
+
+Parameters:
+
+[cols="2,5",opts="header"]
+|===
+|Parameter | Description
+|--nodeId <nodeId>| Optional. UUID of the originator node to filter targets.
+|--min-duration <seconds>| 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 <uuid> --min-duration 30
+
+# Cancel index queries on a specific node
+control.sh --kill all index --nodeId <uuid>
+
+# 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 c6b6c5b8a80..0fc350ab7c8 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
@@ -71,6 +71,8 @@ import
org.apache.ignite.internal.management.encryption.EncryptionChangeMasterKe
import org.apache.ignite.internal.management.encryption.EncryptionCommand;
import org.apache.ignite.internal.management.event.EventCommand;
import org.apache.ignite.internal.management.io.IoTestCommand;
+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;
@@ -479,6 +481,13 @@ public class CommandHandlerParsingTest {
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();
@@ -524,6 +533,8 @@ public class CommandHandlerParsingTest {
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));
@@ -695,6 +706,14 @@ public class CommandHandlerParsingTest {
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 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 1c5701b6ada..6528d919f61 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.GridCommandHandlerWalTest;
import org.apache.ignite.util.GridCommandHandlerWithSslFactoryTest;
import org.apache.ignite.util.GridCommandHandlerWithSslTest;
import org.apache.ignite.util.GridPersistenceCommandsTest;
+import org.apache.ignite.util.KillAllCommandsControlShTest;
import org.apache.ignite.util.KillCommandsControlShTest;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
@@ -76,6 +77,7 @@ import org.junit.runners.Suite;
GridCommandHandlerLegacyClientTest.class,
KillCommandsControlShTest.class,
+ KillAllCommandsControlShTest.class,
BaselineEventsLocalTest.class,
BaselineEventsRemoteTest.class,
diff --git
a/modules/control-utility/src/test/java/org/apache/ignite/util/KillAllCommandsControlShTest.java
b/modules/control-utility/src/test/java/org/apache/ignite/util/KillAllCommandsControlShTest.java
new file mode 100644
index 00000000000..607b97bcf0b
--- /dev/null
+++
b/modules/control-utility/src/test/java/org/apache/ignite/util/KillAllCommandsControlShTest.java
@@ -0,0 +1,333 @@
+/*
+ * 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.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;
+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.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;
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+
+/**
+ * Test for mass queries cancellation.
+ */
+public class KillAllCommandsControlShTest extends
GridCommandHandlerClusterByClassAbstractTest {
+ /** Operations timeout. */
+ public static final int TIMEOUT = 10_000;
+
+ /** */
+ private static final int ENTRIES_CNT = 1_000;
+
+ /** */
+ private static CountDownLatch latch;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ IgniteCache<Object, Object> cache = client.getOrCreateCache(
+ new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+ .setIndexedTypes(Integer.class, Integer.class)
+ .setSqlFunctionClasses(SqlTestFunctions.class));
+
+ for (int i = 0; i < ENTRIES_CNT; 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),
KillAllCommandsControlShTest::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;
+ }), KillAllCommandsControlShTest::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), KillAllCommandsControlShTest::indexQueriesCnt);
+ }
+
+ /** */
+ @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());
+ }
+
+ /** */
+ @Test
+ public void testKillUnrelated() {
+ try (
+ QueryCursor<Cache.Entry<Integer, Integer>> cur =
client.cache(DEFAULT_CACHE_NAME)
+ .query(new IndexQuery<Integer,
Integer>(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<Cache.Entry<Integer, Integer>> cur =
client.cache(DEFAULT_CACHE_NAME)
+ .query(new ScanQuery<Integer, Integer>().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<Query<?>> qryFactory,
ToIntFunction<IgniteEx> qryCntProvider) {
+ try {
+ assertTrue(SERVER_NODE_CNT >= 2);
+
+ long ts = U.currentTimeMillis();
+ GridTestClockTimer.timeSupplier(() -> ts);
+
+ List<QueryCursor<?>> 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().<SqlQueryView>view(SQL_QRY_VIEW).iterator(),
v -> !v.mapQuery());
+ }
+
+ /** */
+ private static int scanQueriesCnt(IgniteEx ignite) {
+ 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();
+ }
+
+ /** */
+ 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 00000000000..a3b590b28a0
--- /dev/null
+++
b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommand.java
@@ -0,0 +1,96 @@
+/*
+ * 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<KillAllCommandArg,
Map<ClusterNode, KillAllTaskResult>> {
+ /** {@inheritDoc} */
+ @Override public String description() {
+ return "Kill all SQL/scan/index/continuous queries matching specified
criteria";
+ }
+
+ /** {@inheritDoc} */
+ @Override public Class<KillAllCommandArg> argClass() {
+ return KillAllCommandArg.class;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Class<KillAllTask> taskClass() {
+ return KillAllTask.class;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Collection<ClusterNode> nodes(Collection<ClusterNode>
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<ClusterNode, KillAllTaskResult> res,
+ Consumer<String> printer
+ ) {
+ if (res.isEmpty()) {
+ printer.accept("Nothing found.");
+ return;
+ }
+
+ int totalKilled = 0;
+ int totalFailed = 0;
+
+ for (Map.Entry<ClusterNode, KillAllTaskResult> 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 00000000000..20a818b1713
--- /dev/null
+++
b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllCommandArg.java
@@ -0,0 +1,129 @@
+/*
+ * 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.CliConfirmArgument;
+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;
+
+/**
+ * Argument for --kill all command.
+ */
+@CliConfirmArgument
+public class KillAllCommandArg extends IgniteDataTransferObject {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Target type. */
+ @Order(0)
+ @Positional
+ @Argument()
+ @EnumDescription(
+ names = {
+ "SQL",
+ "SCAN",
+ "INDEX",
+ "CONTINUOUS"
+ },
+ descriptions = {
+ "SQL queries",
+ "SCAN queries",
+ "INDEX queries",
+ "CONTINUOUS queries"
+ }
+ )
+ 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,
+
+ /** */
+ INDEX,
+
+ /** */
+ 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 00000000000..ba610730f72
--- /dev/null
+++
b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillAllTask.java
@@ -0,0 +1,287 @@
+/*
+ * 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.GridCacheQueryType;
+import org.apache.ignite.internal.processors.cache.query.ScanQueryIterator;
+import
org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler;
+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;
+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<KillAllCommandArg,
Map<ClusterNode, KillAllTaskResult>, KillAllTaskResult> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** {@inheritDoc} */
+ @Override protected VisorJob<KillAllCommandArg, KillAllTaskResult>
job(KillAllCommandArg arg) {
+ return new KillAllJob(arg, debug);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Map<ClusterNode, KillAllTaskResult> reduce0(
+ List<ComputeJobResult> results
+ ) throws IgniteException {
+ Map<ClusterNode, KillAllTaskResult> 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<KillAllCommandArg,
KillAllTaskResult> {
+ /** */
+ 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 INDEX:
+ return cancelIndexQueries(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<GridRunningQueryInfo> 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 (arg.minDuration() != null && 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 (fut.query().query().type() != GridCacheQueryType.SCAN)
+ continue;
+
+ if (arg.minDuration() != null && 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 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 (arg.minDuration() != null && 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.
+ *
+ * @param arg Command argument.
+ * @return Result.
+ */
+ private KillAllTaskResult cancelContinuousQueries(KillAllCommandArg
arg) {
+ GridContinuousProcessor proc = ignite.context().continuous();
+
+ List<IgniteInternalFuture<?>> futs = new ArrayList<>();
+
+ for (Map.Entry<UUID, ContinousRoutineLocalInfo> e :
proc.localRoutineInfos().entrySet()) {
+ if (!e.getValue().handler().isQuery())
+ continue;
+
+ if (e.getValue().handler() instanceof
CacheContinuousQueryHandler<?, ?> h && h.internal())
+ continue;
+
+ 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 00000000000..8b61bc509a8
--- /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 5dcbacaa934..0abc330d952 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 583d3006bc4..9ebdcbbc251 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 @@ public class GridCacheDistributedQueryManager<K, V>
extends GridCacheQueryManage
return fut;
}
+ /** */
+ public Collection<GridCacheDistributedQueryFuture<?, ?, ?>>
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 2dcc4aa59da..7cbe82234e9 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.UUID;
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<K, V, R>
extends GridFutureAda
/** */
private final IgniteUuid timeoutId = IgniteUuid.randomUuid();
+ /** */
+ private long startTime;
+
/** */
private long endTime;
@@ -103,7 +107,7 @@ public abstract class GridCacheQueryFutureAdapter<K, V, R>
extends GridFutureAda
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 abstract class GridCacheQueryFutureAdapter<K, V,
R> extends GridFutureAda
* @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 @@ public abstract class GridCacheQueryFutureAdapter<K, V, R>
extends GridFutureAda
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/cache/query/continuous/CacheContinuousQueryHandler.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
index fca3a2b0037..4bb6c6fc754 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
@@ -408,6 +408,13 @@ public final class CacheContinuousQueryHandler<K, V>
implements GridContinuousHa
this.internal = internal;
}
+ /**
+ * @return {@code True} if query is internal.
+ */
+ public boolean internal() {
+ return internal;
+ }
+
/**
* @param notifyExisting Notify existing.
*/
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 7570b94f37e..b1c9618ba28 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
@@ -324,12 +324,12 @@ public class GridContinuousProcessor extends
GridProcessorAdapter {
}
/** */
- Map<UUID, RemoteRoutineInfo> remoteRoutineInfos() {
+ public Map<UUID, RemoteRoutineInfo> remoteRoutineInfos() {
return Collections.unmodifiableMap(rmtInfos);
}
/** */
- Map<UUID, ContinousRoutineLocalInfo> localRoutineInfos() {
+ public Map<UUID, ContinousRoutineLocalInfo> 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 eba1e484200..bdeb7e3065d 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 class GridTestClockTimer implements Runnable {
*/
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 51554d2a892..dd6ff4e43ca 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,17 @@ This utility can do the following commands:
Parameters:
new_limit - Decimal value to change re-encryption rate limit (MB/s).
+ 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] [--yes]
+
+ Parameters:
+ SQL - SQL queries.
+ SCAN - SCAN queries.
+ INDEX - INDEX queries.
+ CONTINUOUS - CONTINUOUS queries.
+ --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 6254a020a6f..76f0f8c7e0e 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,17 @@ This utility can do the following commands:
Parameters:
new_limit - Decimal value to change re-encryption rate limit (MB/s).
+ 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] [--yes]
+
+ Parameters:
+ SQL - SQL queries.
+ SCAN - SCAN queries.
+ INDEX - INDEX queries.
+ CONTINUOUS - CONTINUOUS queries.
+ --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
