You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@lucene.apache.org by GitBox <gi...@apache.org> on 2021/02/19 05:30:38 UTC
[GitHub] [lucene-solr] atris opened a new pull request #2403: SOLR-15164: Implement Task Management Interface
atris opened a new pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403
This commit introduces the concept of task management in Solr. Tasks can be marked as cancellable (thus, trackable) and can then be listed (ps -all), cancelled and specific task's status be queried.
This commit also implements SOLR-15165 and LUCENE-9789.
Please refer to the JIRA for more information.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r589950273
##########
File path: lucene/core/src/java/org/apache/lucene/search/CancellableCollector.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, CancellableTask {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
+
+ private Collector collector;
+ private AtomicBoolean isQueryCancelled;
+
+ public CancellableCollector(Collector collector) {
+ Objects.requireNonNull(collector, "Internal collector not provided but wrapper collector accessed");
+
+ this.collector = collector;
+ this.isQueryCancelled = new AtomicBoolean();
+ }
+
+ @Override
+ public LeafCollector getLeafCollector(LeafReaderContext context) throws IOException {
+
+ if (isQueryCancelled.compareAndSet(true, false)) {
Review comment:
No -- the idea was that once the cancellation is processed, you "reset" the flag for sanity, but I guess it serves no special purpose. Changed, thanks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r595401010
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,66 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface supports the following types of operations:
+
+1. List all currently running cancellable tasks.
+2. Cancel a specific task.
+3. Query the status of a specific task.
+
+== Listing All Active Cancellable Tasks
+To list all the active cancellable tasks currently running, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list`
+
+==== Sample Response
+
+`{responseHeader={status=0, QTime=11370}, taskList={0=q=*%3A*&canCancel=true&queryUUID=0&_stateVer_=collection1%3A4&wt=javabin&version=2, 5=q=*%3A*&canCancel=true&queryUUID=5&_stateVer_=collection1%3A4&wt=javabin&version=2, 7=q=*%3A*&canCancel=true&queryUUID=7&_stateVer_=collection1%3A4&wt=javabin&version=2}`
+
+== Cancelling An Active Cancellable Task
+To cancel an active task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/cancel?queryUUID=foobar`
+
+==== Sample Response
+===== If the task UUID was found and successfully cancelled:
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 cancelled successfully}`
+
+===== If the task UUID was not found
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 not found}`
+
+== Check Status of a Specific Task
+To check the status of a specific task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list?taskUUID=foobar`
Review comment:
Not in the current status
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r590061216
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,73 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface (TMI) supports the following types of operations:
+
+1. List all currently running cancellable tasks.
+2. Cancel a specific task.
+3. Query the status of a specific task.
+
+== Listing All Active Cancellable Tasks
+To list all the active cancellable tasks currently running, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list`
+
+==== Sample Response
+
+`{responseHeader={status=0, QTime=11370}, taskList={0=q=*%3A*&canCancel=true&queryUUID=0&_stateVer_=collection1%3A4&wt=javabin&version=2, 5=q=*%3A*&canCancel=true&queryUUID=5&_stateVer_=collection1%3A4&wt=javabin&version=2, 7=q=*%3A*&canCancel=true&queryUUID=7&_stateVer_=collection1%3A4&wt=javabin&version=2}`
+
+== Cancelling An Active Cancellable Task
+To cancel an active task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/cancel?cancelUUID=foobar`
+
+==== cancelUUID Parameter
+This parameter is used to specify the UUID of the task to be cancelled.
+
+==== Sample Response
+===== If the task UUID was found and successfully cancelled:
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 cancelled successfully}`
+
+===== If the task UUID was not found
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 not found}`
+
+===== If the cancellation failed
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 could not be cancelled successfully}`
Review comment:
Honestly, this is just me being paranoid -- there is no known cause why a query will not be cancelled. If there is a SolrServerException or some other kind of runtime issue, it will be propagated appropriately without a need for this response. Removed the cancellation failed messaging.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r589255458
##########
File path: solr/solrj/src/java/org/apache/solr/common/params/CommonParams.java
##########
@@ -160,6 +160,26 @@
*/
String TIME_ALLOWED = "timeAllowed";
+ /**
+ * Is the query cancellable?
+ */
+ String IS_QUERY_CANCELLABLE = "canCancel";
+
+ /**
+ * Custom query UUID if provided.
+ */
+ String CUSTOM_QUERY_UUID = "queryUUID";
+
+ /**
+ * UUID for query to be cancelled
+ */
+ String QUERY_CANCELLATION_UUID = "cancelUUID";
+
+ /**
+ * UUID of the task whose status is to be checked
+ */
+ String TASK_CHECK_UUID = "taskUUID";
Review comment:
We will (I plan to get to the long running collection creation soon)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] muse-dev[bot] commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
muse-dev[bot] commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r589280471
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListHandler.java
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.api.Api;
+import org.apache.solr.api.ApiBag;
+import org.apache.solr.request.SolrQueryRequest;
+import org.apache.solr.request.SolrRequestHandler;
+import org.apache.solr.response.SolrQueryResponse;
+import org.apache.solr.security.AuthorizationContext;
+import org.apache.solr.security.PermissionNameProvider;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.solr.common.params.CommonParams.TASK_CHECK_UUID;
+
+/**
+ * Handles request for listing all active cancellable tasks
+ */
+public class ActiveTasksListHandler extends TaskManagementHandler {
+ // This can be a parent level member but we keep it here to allow future handlers to have
+ // a custom list of components
+ private List<SearchComponent> components;
+
+ @Override
+ public void handleRequestBody(SolrQueryRequest req, SolrQueryResponse rsp) throws Exception {
+ Map<String, String> extraParams = null;
+ ResponseBuilder rb = buildResponseBuilder(req, rsp, getComponentsList());
+
+ String taskStatusCheckUUID = req.getParams().get(TASK_CHECK_UUID, null);
+
+ if (taskStatusCheckUUID != null) {
+ if (rb.isDistrib) {
+ extraParams = new HashMap<>();
+
+ extraParams.put(TASK_CHECK_UUID, taskStatusCheckUUID);
+ }
+
+ rb.setTaskStatusCheckUUID(taskStatusCheckUUID);
+ }
+
+ // Let this be visible to handleResponses in the handling component
+ rb.setTaskListRequest(true);
+
+ processRequest(req, rb, extraParams);
+ }
+
+ @Override
+ public String getDescription() {
+ return "activetaskslist";
+ }
+
+ @Override
+ public Category getCategory() {
+ return Category.ADMIN;
+ }
+
+ @Override
+ public PermissionNameProvider.Name getPermissionName(AuthorizationContext ctx) {
+ return PermissionNameProvider.Name.READ_PERM;
+ }
+
+ @Override
+ public SolrRequestHandler getSubHandler(String path) {
Review comment:
*:* param being equated to Strings using equals
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
##########
File path: solr/core/src/java/org/apache/solr/handler/component/QueryCancellationHandler.java
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.api.Api;
+import org.apache.solr.api.ApiBag;
+import org.apache.solr.request.SolrQueryRequest;
+import org.apache.solr.request.SolrRequestHandler;
+import org.apache.solr.response.SolrQueryResponse;
+import org.apache.solr.security.AuthorizationContext;
+import org.apache.solr.security.PermissionNameProvider;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.solr.common.params.CommonParams.QUERY_UUID;
+
+/**
+ * Handles requests for query cancellation for cancellable queries
+ */
+public class QueryCancellationHandler extends TaskManagementHandler {
+ // This can be a parent level member but we keep it here to allow future handlers to have
+ // a custom list of components
+ private List<SearchComponent> components;
+
+ @Override
+ public void handleRequestBody(SolrQueryRequest req, SolrQueryResponse rsp) throws Exception {
+ ResponseBuilder rb = buildResponseBuilder(req, rsp, getComponentsList());
+ Map<String, String> extraParams = null;
+
+ rb.setCancellation(true);
+
+ String cancellationUUID = req.getParams().get(QUERY_UUID, null);
+
+ if (cancellationUUID == null) {
+ throw new IllegalArgumentException("Query cancellation was requested but no query UUID for cancellation was given");
+ }
+
+ if (rb.isDistrib) {
+ extraParams = new HashMap<>();
+
+ extraParams.put(QUERY_UUID, cancellationUUID);
+ }
+
+ // Let this be visible to handleResponses in the handling component
+ rb.setCancellationUUID(cancellationUUID);
+
+ processRequest(req, rb, extraParams);
+ }
+
+ @Override
+ public String getDescription() {
+ return "Cancel queries";
+ }
+
+ @Override
+ public Category getCategory() {
+ return Category.ADMIN;
+ }
+
+ @Override
+ public PermissionNameProvider.Name getPermissionName(AuthorizationContext ctx) {
+ return PermissionNameProvider.Name.READ_PERM;
+ }
+
+ @Override
+ public SolrRequestHandler getSubHandler(String path) {
Review comment:
*:* param being equated to Strings using equals
(at-me [in a reply](https://docs.muse.dev/docs/talk-to-muse/) with `help` or `ignore`)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#issuecomment-781833858
@sigram Please review
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r592508784
##########
File path: solr/core/src/java/org/apache/solr/core/CancellableQueryTracker.java
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.solr.core;
+
+import org.apache.solr.client.solrj.util.Cancellable;
+import org.apache.solr.search.CancellableCollector;
+import org.apache.solr.request.SolrQueryRequest;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.solr.common.params.CommonParams.QUERY_UUID;
+
+/**
+ * Tracks metadata for active queries and provides methods for access
+ */
+public class CancellableQueryTracker {
+ //TODO: This needs to become a time aware storage model
+ private final Map<String, Cancellable> activeCancellableQueries = new ConcurrentHashMap<>();
+ private final Map<String, String> activeQueriesGenerated = new ConcurrentHashMap<>();
+
+ /** Generates a UUID for the given query or if the user provided a UUID
+ * for this query, uses that.
+ */
+ public String generateQueryID(SolrQueryRequest req) {
+ String queryID;
+ String customQueryUUID = req.getParams().get(QUERY_UUID, null);
+
+ if (customQueryUUID != null) {
+ queryID = customQueryUUID;
+ } else {
+ queryID = UUID.randomUUID().toString();
Review comment:
Ack, will add a todo
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r589952550
##########
File path: solr/solr-ref-guide/src/common-query-parameters.adoc
##########
@@ -84,6 +84,18 @@ You can use the `rows` parameter to paginate results from a query. The parameter
The default value is `10`. That is, by default, Solr returns 10 documents at a time in response to a query.
+== canCancel Parameter
+
+This parameter defines if this query is cancellable i.e. can be cancelled during execution using the
+task management interface.
+
+== queryUUID Parameter
+
+For cancellable queries, this allows specifying a custom UUID to identify the query with. If `canCancel` is specified and `queryUUID` is not set, an auto generated UUID will be assigned to the query.
+
+If `queryUUID` is specified, this UUID will be used for identifying the query. Note that if using `queryUUID`, the responsibility of ensuring uniqueness of the UUID lies with the caller.
Review comment:
Updated the docs
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r591126938
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,73 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface (TMI) supports the following types of operations:
+
+1. List all currently running cancellable tasks.
+2. Cancel a specific task.
+3. Query the status of a specific task.
+
+== Listing All Active Cancellable Tasks
+To list all the active cancellable tasks currently running, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list`
+
+==== Sample Response
+
+`{responseHeader={status=0, QTime=11370}, taskList={0=q=*%3A*&canCancel=true&queryUUID=0&_stateVer_=collection1%3A4&wt=javabin&version=2, 5=q=*%3A*&canCancel=true&queryUUID=5&_stateVer_=collection1%3A4&wt=javabin&version=2, 7=q=*%3A*&canCancel=true&queryUUID=7&_stateVer_=collection1%3A4&wt=javabin&version=2}`
+
+== Cancelling An Active Cancellable Task
+To cancel an active task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/cancel?cancelUUID=foobar`
+
+==== cancelUUID Parameter
+This parameter is used to specify the UUID of the task to be cancelled.
+
+==== Sample Response
+===== If the task UUID was found and successfully cancelled:
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 cancelled successfully}`
+
+===== If the task UUID was not found
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 not found}`
+
+===== If the cancellation failed
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 could not be cancelled successfully}`
+
+== Check Status of a Specific Task
+To check the status of a specific task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list?taskUUID=foobar`
+
+==== taskUUID Parameter
+`taskUUID` parameter can be used to specify a task UUID whose status can be checked.
+
+==== Sample Response
+`{responseHeader={status=0, QTime=6128}, taskStatus=foobar:true}`
Review comment:
Yes. This checks if the query exists in the system or not.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris closed pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris closed pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r592959697
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
Review comment:
Sorry, missed a lot of comments. Fixing now.
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
+ ConcurrentHashSet<Integer> notFoundIdsSet = new ConcurrentHashSet<>();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertTrue(queryIdsSet.size() + notFoundIdsSet.size() == 90);
+ }
+
+ @Test
+ public void testListCancellableQueries() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/list");
+
+ for (int i = 0; i < 50; i++) {
+ executeQueryAsync(Integer.toString(i));
+ }
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ NamedList<String> result = (NamedList<String>) queryResponse.get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ Set<Integer> presentQueryIDs = new HashSet<>();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ presentQueryIDs.add(Integer.parseInt(entry.getKey()));
+ }
+
+ assertTrue(presentQueryIDs.size() > 0 && presentQueryIDs.size() <= 50);
+
+ Iterator<Integer> integerIterator = presentQueryIDs.iterator();
+
+ while (integerIterator.hasNext()) {
+ int value = integerIterator.next();
+
+ assertTrue (value >= 0 && value < 50);
+ }
+ }
+
+ @Test
+ public void testCheckSpecificQueryStatus() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("taskUUID", "25");
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+
+ request.setPath("/tasks/list");
+
+ NamedList<Object> queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ String result = (String) queryResponse.get("taskStatus");
+
+ assertFalse(result.contains("true"));
+ }
+
+ private CompletableFuture<Void> cancelQuery(final String queryID, final int sleepTime, Set<Integer> cancelledQueryIdsSet,
+ Set<Integer> notFoundQueryIdSet) {
+ return CompletableFuture.runAsync(() -> {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", queryID);
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ // Wait for some time to let the query start
+ try {
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+
+ try {
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ String cancellationResult = (String) queryResponse.get("status");
+ if (cancellationResult.contains("cancelled successfully")) {
+ cancelledQueryIdsSet.add(Integer.parseInt(queryID));
+ } else if (cancellationResult.contains("not found")) {
+ notFoundQueryIdSet.add(Integer.parseInt(queryID));
+ }
Review comment:
Changed to returning HTTP codes
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
+ ConcurrentHashSet<Integer> notFoundIdsSet = new ConcurrentHashSet<>();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertTrue(queryIdsSet.size() + notFoundIdsSet.size() == 90);
+ }
+
+ @Test
+ public void testListCancellableQueries() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/list");
+
+ for (int i = 0; i < 50; i++) {
+ executeQueryAsync(Integer.toString(i));
+ }
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ NamedList<String> result = (NamedList<String>) queryResponse.get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ Set<Integer> presentQueryIDs = new HashSet<>();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ presentQueryIDs.add(Integer.parseInt(entry.getKey()));
+ }
+
+ assertTrue(presentQueryIDs.size() > 0 && presentQueryIDs.size() <= 50);
+
+ Iterator<Integer> integerIterator = presentQueryIDs.iterator();
+
+ while (integerIterator.hasNext()) {
+ int value = integerIterator.next();
+
+ assertTrue (value >= 0 && value < 50);
+ }
+ }
+
+ @Test
+ public void testCheckSpecificQueryStatus() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("taskUUID", "25");
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+
+ request.setPath("/tasks/list");
+
+ NamedList<Object> queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ String result = (String) queryResponse.get("taskStatus");
+
+ assertFalse(result.contains("true"));
+ }
+
+ private CompletableFuture<Void> cancelQuery(final String queryID, final int sleepTime, Set<Integer> cancelledQueryIdsSet,
Review comment:
Yeah, to allow future modifications :)
##########
File path: solr/core/src/java/org/apache/solr/core/CancellableQueryTracker.java
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.solr.core;
+
+import org.apache.solr.client.solrj.util.Cancellable;
+import org.apache.solr.search.CancellableCollector;
+import org.apache.solr.request.SolrQueryRequest;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.solr.common.params.CommonParams.QUERY_UUID;
+
+/**
+ * Tracks metadata for active queries and provides methods for access
+ */
+public class CancellableQueryTracker {
+ //TODO: This needs to become a time aware storage model
+ private final Map<String, Cancellable> activeCancellableQueries = new ConcurrentHashMap<>();
+ private final Map<String, String> activeQueriesGenerated = new ConcurrentHashMap<>();
+
+ /** Generates a UUID for the given query or if the user provided a UUID
+ * for this query, uses that.
+ */
+ public String generateQueryID(SolrQueryRequest req) {
+ String queryID;
+ String customQueryUUID = req.getParams().get(QUERY_UUID, null);
+
+ if (customQueryUUID != null) {
+ queryID = customQueryUUID;
+ } else {
+ queryID = UUID.randomUUID().toString();
+ }
+
+ if (activeQueriesGenerated.containsKey(queryID)) {
+ if (customQueryUUID != null) {
+ throw new IllegalArgumentException("Duplicate query UUID given");
Review comment:
Added to docs, thanks
##########
File path: solr/core/src/java/org/apache/solr/handler/component/TaskManagementHandler.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.ShardParams;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.handler.RequestHandlerBase;
+import org.apache.solr.request.SolrQueryRequest;
+import org.apache.solr.response.SolrQueryResponse;
+import org.apache.solr.security.PermissionNameProvider;
+import org.apache.solr.util.plugin.SolrCoreAware;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.solr.common.params.CommonParams.DISTRIB;
+import static org.apache.solr.common.params.CommonParams.PATH;
+
+/**
+ * Abstract class which serves as the root of all task managing handlers
+ */
+public abstract class TaskManagementHandler extends RequestHandlerBase implements SolrCoreAware, PermissionNameProvider {
+ private ShardHandlerFactory shardHandlerFactory;
+
+ @Override
+ public void inform(SolrCore core) {
+ this.shardHandlerFactory = core.getCoreContainer().getShardHandlerFactory();
+ }
+
+ /**
+ * Process the actual request.
+ * extraParams is required for allowing sub handlers to pass in custom parameters to be put in the
+ * outgoing shard request
+ */
+ protected void processRequest(SolrQueryRequest req, ResponseBuilder rb, Map<String, String> extraParams) throws IOException {
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
+ List<SearchComponent> components = rb.components;
+
+ shardHandler.prepDistributed(rb);
+
+ for(SearchComponent c : components) {
+ c.prepare(rb);
+ }
+
+ if (!rb.isDistrib) {
+ for (SearchComponent component : components) {
+ component.process(rb);
+ }
+ } else {
+ ShardRequest sreq = new ShardRequest();
+
+ // Distribute to all shards
+ sreq.shards = rb.shards;
+ sreq.actualShards = sreq.shards;
+
+ sreq.responses = new ArrayList<>(sreq.actualShards.length);
+ rb.finished = new ArrayList<>();
+
+ for (String shard : sreq.actualShards) {
+ ModifiableSolrParams params = new ModifiableSolrParams(sreq.params);
+ String reqPath = (String) req.getContext().get(PATH);
+
+ params.set(CommonParams.QT, reqPath);
+ params.remove(ShardParams.SHARDS); // not a top-level request
+ params.set(DISTRIB, "false"); // not a top-level request
+ params.remove("indent");
+ params.remove(CommonParams.HEADER_ECHO_PARAMS);
+ params.set(ShardParams.IS_SHARD, true); // a sub (shard) request
+ params.set(ShardParams.SHARDS_PURPOSE, sreq.purpose);
+ params.set(ShardParams.SHARD_URL, shard); // so the shard knows what was asked
+ params.set(CommonParams.OMIT_HEADER, false);
Review comment:
Unfortunately, no, without refactoring SearchHandler, which I am hesitant to do
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r589256768
##########
File path: solr/solrj/src/java/org/apache/solr/common/params/ShardParams.java
##########
@@ -28,6 +28,9 @@
public interface ShardParams {
/** the shards to use (distributed configuration) */
String SHARDS = "shards";
+
+ /** UUID of the query */
+ String QUERY_ID = "queryID";
Review comment:
Will that not cause confusion, since this is the generic queryID (generated for every cancellable query) and the other parameter references the queryID the incoming cancellation/list request wants to deal with?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] madrob commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
madrob commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r595245360
##########
File path: solr/core/src/java/org/apache/solr/handler/component/QueryCancellationComponent.java
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.client.solrj.util.Cancellable;
+
+import java.io.IOException;
+
+/** Responsible for handling query cancellation requests */
+public class QueryCancellationComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "querycancellation";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException
+ {
+ if (rb.isCancellation()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ String cancellationUUID = rb.getCancellationUUID();
+
+ if (cancellationUUID == null) {
+ throw new RuntimeException("Null query UUID seen");
+ }
+
+ Cancellable cancellableTask = rb.req.getCore().getCancellableQueryTracker().getCancellableTask(cancellationUUID);
+
+ if (cancellableTask != null) {
+ cancellableTask.cancel();
+ rb.rsp.add("cancellationResult", "success");
+ } else {
+ rb.rsp.add("cancellationResult", "not found");
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void handleResponses(ResponseBuilder rb, ShardRequest sreq) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ boolean queryFound = false;
+
+ for (ShardResponse r : sreq.responses) {
+
+ String cancellationResult = (String) r.getSolrResponse()
+ .getResponse().get("cancellationResult");
+
+ if (cancellationResult.equalsIgnoreCase("success")) {
+ queryFound = true;
+
+ break;
+ }
+ }
+
+ // If any shard sees the query as present, then we mark the query as successfully cancelled. If no shard found
+ // the query, then that can denote that the query was not found. This is important since the query cancellation
+ // request is broadcast to all shards, and the query might have completed on some shards but not on others
+
+ if(queryFound) {
+ rb.rsp.getValues().add("status", "Query with queryID " + rb.getCancellationUUID() +
+ " cancelled successfully");
+ rb.rsp.getValues().add("responseCode", 200 /* HTTP OK */);
+ } else {
+ rb.rsp.getValues().add("status", "Query with queryID " + rb.getCancellationUUID() +
+ " not found");
+ rb.rsp.getValues().add("responseCode", 401 /* HTTP NOT FOUND */);
Review comment:
Do you mean 404?
##########
File path: solr/core/src/java/org/apache/solr/search/CancellableCollector.java
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.solr.search;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.Collector;
+import org.apache.lucene.search.FilterLeafCollector;
+import org.apache.lucene.search.LeafCollector;
+import org.apache.lucene.search.ScoreMode;
+import org.apache.solr.client.solrj.util.Cancellable;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, Cancellable {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
+
+ private final Collector collector;
+ private final AtomicBoolean isQueryCancelled;
+
+ public CancellableCollector(Collector collector) {
+ Objects.requireNonNull(collector, "Internal collector not provided but wrapper collector accessed");
+
+ this.collector = collector;
+ this.isQueryCancelled = new AtomicBoolean();
+ }
+
+ @Override
+ public LeafCollector getLeafCollector(LeafReaderContext context) throws IOException {
+
+ if (isQueryCancelled.compareAndSet(true, true)) {
Review comment:
if it's already true, set it to true...?
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,261 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("queryUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
Review comment:
Should test on status code as well or instead.
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,66 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface supports the following types of operations:
+
+1. List all currently running cancellable tasks.
+2. Cancel a specific task.
+3. Query the status of a specific task.
+
+== Listing All Active Cancellable Tasks
+To list all the active cancellable tasks currently running, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list`
+
+==== Sample Response
+
+`{responseHeader={status=0, QTime=11370}, taskList={0=q=*%3A*&canCancel=true&queryUUID=0&_stateVer_=collection1%3A4&wt=javabin&version=2, 5=q=*%3A*&canCancel=true&queryUUID=5&_stateVer_=collection1%3A4&wt=javabin&version=2, 7=q=*%3A*&canCancel=true&queryUUID=7&_stateVer_=collection1%3A4&wt=javabin&version=2}`
+
+== Cancelling An Active Cancellable Task
+To cancel an active task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/cancel?queryUUID=foobar`
+
+==== Sample Response
+===== If the task UUID was found and successfully cancelled:
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 cancelled successfully}`
+
+===== If the task UUID was not found
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 not found}`
+
+== Check Status of a Specific Task
+To check the status of a specific task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list?taskUUID=foobar`
Review comment:
Can we specify multiple tasks in a single query?
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,73 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface (TMI) supports the following types of operations:
+
+1. List all currently running cancellable tasks.
+2. Cancel a specific task.
+3. Query the status of a specific task.
+
+== Listing All Active Cancellable Tasks
+To list all the active cancellable tasks currently running, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list`
+
+==== Sample Response
+
+`{responseHeader={status=0, QTime=11370}, taskList={0=q=*%3A*&canCancel=true&queryUUID=0&_stateVer_=collection1%3A4&wt=javabin&version=2, 5=q=*%3A*&canCancel=true&queryUUID=5&_stateVer_=collection1%3A4&wt=javabin&version=2, 7=q=*%3A*&canCancel=true&queryUUID=7&_stateVer_=collection1%3A4&wt=javabin&version=2}`
+
+== Cancelling An Active Cancellable Task
+To cancel an active task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/cancel?cancelUUID=foobar`
+
+==== cancelUUID Parameter
+This parameter is used to specify the UUID of the task to be cancelled.
+
+==== Sample Response
+===== If the task UUID was found and successfully cancelled:
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 cancelled successfully}`
+
+===== If the task UUID was not found
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 not found}`
+
+===== If the cancellation failed
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 could not be cancelled successfully}`
+
+== Check Status of a Specific Task
+To check the status of a specific task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list?taskUUID=foobar`
+
+==== taskUUID Parameter
+`taskUUID` parameter can be used to specify a task UUID whose status can be checked.
+
+==== Sample Response
+`{responseHeader={status=0, QTime=6128}, taskStatus=foobar:true}`
Review comment:
Right, and when checking for existence `taskStatus=foobar:true` does not feel meaningful. Maybe something a little more verbose: `taskStatus={id:foobar, status:active}` would be easier to understand.
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,261 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("queryUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() {
+ Set<Integer> queryIdsSet = ConcurrentHashMap.newKeySet();
+ Set<Integer> notFoundIdsSet = ConcurrentHashMap.newKeySet();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertEquals("Total query count did not match the expected value",
+ queryIdsSet.size() + notFoundIdsSet.size(), 90);
+ }
+
+ @Test
+ public void testListCancellableQueries() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/list");
+
+ for (int i = 0; i < 50; i++) {
+ executeQueryAsync(Integer.toString(i));
+ }
+
+ NamedList<Object> queryResponse;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ NamedList<String> result = (NamedList<String>) queryResponse.get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ Set<Integer> presentQueryIDs = new HashSet<>();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ presentQueryIDs.add(Integer.parseInt(entry.getKey()));
+ }
+
+ assertTrue(presentQueryIDs.size() > 0 && presentQueryIDs.size() <= 50);
+
+ for (int value : presentQueryIDs) {
+ assertTrue(value >= 0 && value < 50);
+ }
+ }
+
+ @Test
+ public void testCheckSpecificQueryStatus() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("taskUUID", "25");
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+
+ request.setPath("/tasks/list");
+
+ NamedList<Object> queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ String result = (String) queryResponse.get("taskStatus");
+
+ assertFalse(result.contains("true"));
+ }
+
+ private CompletableFuture<Void> cancelQuery(final String queryID, final int sleepTime, Set<Integer> cancelledQueryIdsSet,
+ Set<Integer> notFoundQueryIdSet) {
+ return CompletableFuture.runAsync(() -> {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("queryUUID", queryID);
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ // Wait for some time to let the query start
+ try {
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+
+ try {
+ NamedList<Object> queryResponse;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ int responseCode = (int) queryResponse.get("responseCode");
+
+ if (responseCode == 200 /* HTTP OK */) {
+ cancelledQueryIdsSet.add(Integer.parseInt(queryID));
+ } else if (responseCode == 401 /* HTTP NOT FOUND */) {
+ notFoundQueryIdSet.add(Integer.parseInt(queryID));
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e.getMessage());
+ }
+ } catch (InterruptedException e) {
Review comment:
restore interrupted status
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,261 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("queryUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() {
+ Set<Integer> queryIdsSet = ConcurrentHashMap.newKeySet();
+ Set<Integer> notFoundIdsSet = ConcurrentHashMap.newKeySet();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertEquals("Total query count did not match the expected value",
+ queryIdsSet.size() + notFoundIdsSet.size(), 90);
+ }
+
+ @Test
+ public void testListCancellableQueries() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/list");
+
+ for (int i = 0; i < 50; i++) {
+ executeQueryAsync(Integer.toString(i));
+ }
+
+ NamedList<Object> queryResponse;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ NamedList<String> result = (NamedList<String>) queryResponse.get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ Set<Integer> presentQueryIDs = new HashSet<>();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ presentQueryIDs.add(Integer.parseInt(entry.getKey()));
+ }
+
+ assertTrue(presentQueryIDs.size() > 0 && presentQueryIDs.size() <= 50);
+
+ for (int value : presentQueryIDs) {
+ assertTrue(value >= 0 && value < 50);
+ }
+ }
+
+ @Test
+ public void testCheckSpecificQueryStatus() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("taskUUID", "25");
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+
+ request.setPath("/tasks/list");
+
+ NamedList<Object> queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ String result = (String) queryResponse.get("taskStatus");
+
+ assertFalse(result.contains("true"));
+ }
+
+ private CompletableFuture<Void> cancelQuery(final String queryID, final int sleepTime, Set<Integer> cancelledQueryIdsSet,
+ Set<Integer> notFoundQueryIdSet) {
+ return CompletableFuture.runAsync(() -> {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("queryUUID", queryID);
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ // Wait for some time to let the query start
+ try {
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+
+ try {
+ NamedList<Object> queryResponse;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ int responseCode = (int) queryResponse.get("responseCode");
+
+ if (responseCode == 200 /* HTTP OK */) {
+ cancelledQueryIdsSet.add(Integer.parseInt(queryID));
+ } else if (responseCode == 401 /* HTTP NOT FOUND */) {
+ notFoundQueryIdSet.add(Integer.parseInt(queryID));
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e.getMessage());
Review comment:
`throw new CompletionException(e);`
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,261 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("queryUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() {
+ Set<Integer> queryIdsSet = ConcurrentHashMap.newKeySet();
+ Set<Integer> notFoundIdsSet = ConcurrentHashMap.newKeySet();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
Review comment:
Why do we start 100 queries, but only cancel 90?
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,261 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
Review comment:
duplicate
##########
File path: solr/core/src/java/org/apache/solr/search/CancellableCollector.java
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.solr.search;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.search.Collector;
+import org.apache.lucene.search.FilterLeafCollector;
+import org.apache.lucene.search.LeafCollector;
+import org.apache.lucene.search.ScoreMode;
+import org.apache.solr.client.solrj.util.Cancellable;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, Cancellable {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
+
+ private final Collector collector;
+ private final AtomicBoolean isQueryCancelled;
+
+ public CancellableCollector(Collector collector) {
+ Objects.requireNonNull(collector, "Internal collector not provided but wrapper collector accessed");
+
+ this.collector = collector;
+ this.isQueryCancelled = new AtomicBoolean();
+ }
+
+ @Override
+ public LeafCollector getLeafCollector(LeafReaderContext context) throws IOException {
+
+ if (isQueryCancelled.compareAndSet(true, true)) {
+ throw new QueryCancelledException();
+ }
+
+ return new FilterLeafCollector(collector.getLeafCollector(context)) {
+
+ @Override
+ public void collect(int doc) throws IOException {
+ if (isQueryCancelled.compareAndSet(true, true)) {
Review comment:
same as above
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#issuecomment-792591442
@sigram Updated per comments, please see and let me know your comments
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#issuecomment-795007935
@sigram @madrob @anshumg Updated the PR, please see and let me know.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r592889141
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,73 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface (TMI) supports the following types of operations:
+
+1. List all currently running cancellable tasks.
+2. Cancel a specific task.
+3. Query the status of a specific task.
+
+== Listing All Active Cancellable Tasks
+To list all the active cancellable tasks currently running, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list`
+
+==== Sample Response
+
+`{responseHeader={status=0, QTime=11370}, taskList={0=q=*%3A*&canCancel=true&queryUUID=0&_stateVer_=collection1%3A4&wt=javabin&version=2, 5=q=*%3A*&canCancel=true&queryUUID=5&_stateVer_=collection1%3A4&wt=javabin&version=2, 7=q=*%3A*&canCancel=true&queryUUID=7&_stateVer_=collection1%3A4&wt=javabin&version=2}`
+
+== Cancelling An Active Cancellable Task
+To cancel an active task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/cancel?cancelUUID=foobar`
+
+==== cancelUUID Parameter
+This parameter is used to specify the UUID of the task to be cancelled.
+
+==== Sample Response
+===== If the task UUID was found and successfully cancelled:
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 cancelled successfully}`
+
+===== If the task UUID was not found
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 not found}`
Review comment:
Yes
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] madrob commented on pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
madrob commented on pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#issuecomment-801460550
Have you done performance testing with this? I'm concerned about how this could affect query throughput if we're funneling everything on a CHM
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] madrob commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
madrob commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r592489533
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
Review comment:
@atris did you see this comment?
##########
File path: solr/core/src/java/org/apache/solr/core/CancellableQueryTracker.java
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.solr.core;
+
+import org.apache.solr.client.solrj.util.Cancellable;
+import org.apache.solr.search.CancellableCollector;
+import org.apache.solr.request.SolrQueryRequest;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.solr.common.params.CommonParams.QUERY_UUID;
+
+/**
+ * Tracks metadata for active queries and provides methods for access
+ */
+public class CancellableQueryTracker {
+ //TODO: This needs to become a time aware storage model
+ private final Map<String, Cancellable> activeCancellableQueries = new ConcurrentHashMap<>();
+ private final Map<String, String> activeQueriesGenerated = new ConcurrentHashMap<>();
+
+ /** Generates a UUID for the given query or if the user provided a UUID
+ * for this query, uses that.
+ */
+ public String generateQueryID(SolrQueryRequest req) {
+ String queryID;
+ String customQueryUUID = req.getParams().get(QUERY_UUID, null);
+
+ if (customQueryUUID != null) {
+ queryID = customQueryUUID;
+ } else {
+ queryID = UUID.randomUUID().toString();
+ }
+
+ if (activeQueriesGenerated.containsKey(queryID)) {
+ if (customQueryUUID != null) {
+ throw new IllegalArgumentException("Duplicate query UUID given");
Review comment:
So it's possible that when using a mix of generated and non-generated query IDs that we will get a conflict. Hmm... I would call this case out specifically in the docs, something like recommending that users either use all generated or all custom.
##########
File path: solr/core/src/java/org/apache/solr/core/CancellableQueryTracker.java
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.solr.core;
+
+import org.apache.solr.client.solrj.util.Cancellable;
+import org.apache.solr.search.CancellableCollector;
+import org.apache.solr.request.SolrQueryRequest;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.solr.common.params.CommonParams.QUERY_UUID;
+
+/**
+ * Tracks metadata for active queries and provides methods for access
+ */
+public class CancellableQueryTracker {
+ //TODO: This needs to become a time aware storage model
+ private final Map<String, Cancellable> activeCancellableQueries = new ConcurrentHashMap<>();
+ private final Map<String, String> activeQueriesGenerated = new ConcurrentHashMap<>();
+
+ /** Generates a UUID for the given query or if the user provided a UUID
+ * for this query, uses that.
+ */
+ public String generateQueryID(SolrQueryRequest req) {
+ String queryID;
+ String customQueryUUID = req.getParams().get(QUERY_UUID, null);
+
+ if (customQueryUUID != null) {
+ queryID = customQueryUUID;
+ } else {
+ queryID = UUID.randomUUID().toString();
Review comment:
I mentioned this previously before the file moved - UUID.randomUUID is eventually slow because it eats system entropy. And it's also synchronized under the hood, getting the bytes, so not great for multi-threaded use. We can leave it in for now and replace with a different implementation later.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#issuecomment-800495022
@madrob Fixed, please see
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r589251072
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/response/QueryResponse.java
##########
@@ -184,6 +187,15 @@ else if ( "terms".equals( n ) ) {
else if ( "moreLikeThis".equals( n ) ) {
_moreLikeThisInfo = (NamedList<SolrDocumentList>) res.getVal( i );
}
+ else if ("taskList".equals( n )) {
Review comment:
I didnt quite parse that -- you mean, reuse `taskInfo` across? How would that work, since each of the three calls returns a different type?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r589948751
##########
File path: solr/core/src/java/org/apache/solr/core/SolrCore.java
##########
@@ -3245,6 +3252,75 @@ public void postClose(SolrCore core) {
return blobRef;
}
+ /** Generates a UUID for the given query or if the user provided a UUID
+ * for this query, uses that.
+ */
+ public String generateQueryID(SolrQueryRequest req) {
+ String queryID;
+ String customQueryUUID = req.getParams().get(CUSTOM_QUERY_UUID, null);
+
+ if (customQueryUUID != null) {
+ queryID = customQueryUUID;
+ } else {
+ queryID = UUID.randomUUID().toString();
Review comment:
I didnt quite parse that -- you mean, not use `UUID`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r592889091
##########
File path: solr/core/src/java/org/apache/solr/core/SolrCore.java
##########
@@ -3245,6 +3252,75 @@ public void postClose(SolrCore core) {
return blobRef;
}
+ /** Generates a UUID for the given query or if the user provided a UUID
+ * for this query, uses that.
+ */
+ public String generateQueryID(SolrQueryRequest req) {
+ String queryID;
+ String customQueryUUID = req.getParams().get(CUSTOM_QUERY_UUID, null);
+
+ if (customQueryUUID != null) {
+ queryID = customQueryUUID;
+ } else {
+ queryID = UUID.randomUUID().toString();
Review comment:
Added a todo for this
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] anshumg commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
anshumg commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r589872439
##########
File path: lucene/core/src/java/org/apache/lucene/search/CancellableCollector.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, CancellableTask {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
+
+ private Collector collector;
+ private AtomicBoolean isQueryCancelled;
Review comment:
This can also be declared final
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
Review comment:
Seems like nothing in here throws an Exception.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.apache.solr.common.util.Utils.fromJSONString;
Review comment:
This is also unused
##########
File path: lucene/core/src/test/org/apache/lucene/search/TestCancellableCollector.java
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.lucene.util.NamedThreadFactory;
+
+public class TestCancellableCollector extends LuceneTestCase {
+ Directory dir;
+ IndexReader reader;
+ IndexSearcher searcher;
+ ExecutorService executor = null;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ dir = newDirectory();
+
+ RandomIndexWriter iw = new RandomIndexWriter(random(), dir);
+ Random random = random();
+ for (int i = 0; i < 100; i++) {
+ Document doc = new Document();
+ doc.add(newStringField("field", Integer.toString(i), Field.Store.NO));
+ doc.add(newStringField("field2", Boolean.toString(i % 2 == 0), Field.Store.NO));
+ doc.add(new SortedDocValuesField("field2", new BytesRef(Boolean.toString(i % 2 == 0))));
+ iw.addDocument(doc);
+
+ if (random.nextBoolean()) {
+ iw.commit();
+ }
+ }
+ reader = iw.getReader();
+ iw.close();
+
+ searcher = new IndexSearcher(reader);
+
+ executor =
+ new ThreadPoolExecutor(
+ 4,
+ 4,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>(),
+ new NamedThreadFactory("TestIndexSearcher"));
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ reader.close();
+ dir.close();
+
+ if (executor != null) {
+ executor.shutdown();
+ }
+
+ executor = null;
+ }
+
+ private CancellableCollector buildCancellableCollector(
+ final int numHits, boolean delayStart, boolean delayCollection) {
+ TopScoreDocCollector topScoreDocCollector = TopScoreDocCollector.create(numHits, null, 1);
+ CancellableCollector collector = new CancellableCollector(topScoreDocCollector);
+
+ DummyCancellableCollector dummyCancellableCollector =
+ new DummyCancellableCollector(collector, delayStart, delayCollection);
+
+ return dummyCancellableCollector;
+ }
+
+ private void executeSearchTest(
+ IndexSearcher searcher, Query query, CancellableCollector cancellableCollector, int numHits)
+ throws Exception {
+ TopDocs topDocs = searcher.search(query, numHits);
+
+ searcher.search(query, cancellableCollector);
+
+ CancellableCollector internalCancellableCollector =
+ (CancellableCollector) cancellableCollector.getInternalCollector();
+ TopScoreDocCollector topScoreDocCollector =
+ (TopScoreDocCollector) internalCancellableCollector.getInternalCollector();
+
+ assertEquals(topDocs.totalHits.value, topScoreDocCollector.totalHits);
+ }
+
+ private void cancelQuery(CancellableCollector cancellableCollector, final int sleepTime) {
+ executor.submit(
+ () -> {
+
+ // Wait for some time to let the query start
+ try {
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+
+ cancellableCollector.cancelTask();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e.getMessage());
+ }
+ });
+ }
+
+ public void testSearchWithoutCancellation() throws Exception {
+ CancellableCollector cancellableCollector = buildCancellableCollector(50, false, false);
+
+ Query query = new TermQuery(new Term("field", "1"));
+
+ executeSearchTest(searcher, query, cancellableCollector, 50);
+
+ query = new MatchAllDocsQuery();
+
+ cancellableCollector = buildCancellableCollector(100, false, false);
+
+ executeSearchTest(searcher, query, cancellableCollector, 50);
+ }
+
+ public void testSearchWithCancellationBeforeActualDocumentCollection() {
+ Query query = new MatchAllDocsQuery();
+
+ CancellableCollector cancellableCollector = buildCancellableCollector(5000, true, false);
+
+ expectThrows(
+ CancellableCollector.QueryCancelledException.class,
+ () -> {
+ // Cancel the query before the document collection starts
+ cancelQuery(cancellableCollector, 0);
+
+ executeSearchTest(searcher, query, cancellableCollector, 5000);
+ });
+ }
+
+ public void testSearchWithCancellationBetweenActualDocumentCollection() {
+ Query query = new MatchAllDocsQuery();
+
+ CancellableCollector cancellableCollector = buildCancellableCollector(5000, false, true);
+
+ expectThrows(
+ CancellableCollector.QueryCancelledException.class,
+ () -> {
+ // Cancel the query before the document collection starts
+ cancelQuery(cancellableCollector, 0);
+
+ executeSearchTest(searcher, query, cancellableCollector, 5000);
+ });
+ }
+
+ public class DummyCancellableCollector extends CancellableCollector {
+ private CancellableCollector collector;
Review comment:
All of these could be final
##########
File path: solr/core/src/java/org/apache/solr/handler/component/TaskManagementHandler.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.ShardParams;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.handler.RequestHandlerBase;
+import org.apache.solr.request.SolrQueryRequest;
+import org.apache.solr.response.SolrQueryResponse;
+import org.apache.solr.security.PermissionNameProvider;
+import org.apache.solr.util.plugin.SolrCoreAware;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.solr.common.params.CommonParams.DISTRIB;
+import static org.apache.solr.common.params.CommonParams.PATH;
+
+/**
+ * Abstract class which serves as the root of all task managing handlers
+ */
+public abstract class TaskManagementHandler extends RequestHandlerBase implements SolrCoreAware, PermissionNameProvider {
+ private ShardHandlerFactory shardHandlerFactory;
+
+ @Override
+ public void inform(SolrCore core) {
+ this.shardHandlerFactory = core.getCoreContainer().getShardHandlerFactory();
+ }
+
+ /**
+ * Process the actual request.
+ * extraParams is required for allowing sub handlers to pass in custom parameters to be put in the
+ * outgoing shard request
+ */
+ protected void processRequest(SolrQueryRequest req, ResponseBuilder rb, Map<String, String> extraParams) throws IOException {
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
+ List<SearchComponent> components = rb.components;
+
+ shardHandler.prepDistributed(rb);
+
+ for(SearchComponent c : components) {
+ c.prepare(rb);
+ }
+
+ if (!rb.isDistrib) {
+ for (SearchComponent component : components) {
+ component.process(rb);
+ }
+ } else {
+ ShardRequest sreq = new ShardRequest();
+
+ // Distribute to all shards
+ sreq.shards = rb.shards;
+ sreq.actualShards = sreq.shards;
+
+ sreq.responses = new ArrayList<>(sreq.actualShards.length);
+ rb.finished = new ArrayList<>();
+
+ for (String shard : sreq.actualShards) {
+ ModifiableSolrParams params = new ModifiableSolrParams(sreq.params);
+ String reqPath = (String) req.getContext().get(PATH);
+
+ params.set(CommonParams.QT, reqPath);
+ params.remove(ShardParams.SHARDS); // not a top-level request
+ params.set(DISTRIB, "false"); // not a top-level request
+ params.remove("indent");
+ params.remove(CommonParams.HEADER_ECHO_PARAMS);
+ params.set(ShardParams.IS_SHARD, true); // a sub (shard) request
+ params.set(ShardParams.SHARDS_PURPOSE, sreq.purpose);
+ params.set(ShardParams.SHARD_URL, shard); // so the shard knows what was asked
+ params.set(CommonParams.OMIT_HEADER, false);
+
+ if (extraParams != null) {
+ Iterator<Map.Entry<String, String>> iterator = extraParams.entrySet().iterator();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ params.set(entry.getKey(), entry.getValue());
Review comment:
Perhaps just use forEach?
##########
File path: lucene/core/src/test/org/apache/lucene/search/TestCancellableCollector.java
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.lucene.util.NamedThreadFactory;
+
+public class TestCancellableCollector extends LuceneTestCase {
+ Directory dir;
+ IndexReader reader;
+ IndexSearcher searcher;
+ ExecutorService executor = null;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ dir = newDirectory();
+
+ RandomIndexWriter iw = new RandomIndexWriter(random(), dir);
+ Random random = random();
+ for (int i = 0; i < 100; i++) {
+ Document doc = new Document();
+ doc.add(newStringField("field", Integer.toString(i), Field.Store.NO));
+ doc.add(newStringField("field2", Boolean.toString(i % 2 == 0), Field.Store.NO));
+ doc.add(new SortedDocValuesField("field2", new BytesRef(Boolean.toString(i % 2 == 0))));
+ iw.addDocument(doc);
+
+ if (random.nextBoolean()) {
+ iw.commit();
+ }
+ }
+ reader = iw.getReader();
+ iw.close();
+
+ searcher = new IndexSearcher(reader);
+
+ executor =
+ new ThreadPoolExecutor(
+ 4,
+ 4,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>(),
+ new NamedThreadFactory("TestIndexSearcher"));
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ reader.close();
+ dir.close();
+
+ if (executor != null) {
+ executor.shutdown();
+ }
+
+ executor = null;
+ }
+
+ private CancellableCollector buildCancellableCollector(
+ final int numHits, boolean delayStart, boolean delayCollection) {
+ TopScoreDocCollector topScoreDocCollector = TopScoreDocCollector.create(numHits, null, 1);
+ CancellableCollector collector = new CancellableCollector(topScoreDocCollector);
+
+ DummyCancellableCollector dummyCancellableCollector =
Review comment:
Just return here instead of the assignment?
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
+ ConcurrentHashSet<Integer> notFoundIdsSet = new ConcurrentHashSet<>();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertTrue(queryIdsSet.size() + notFoundIdsSet.size() == 90);
+ }
+
+ @Test
+ public void testListCancellableQueries() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/list");
+
+ for (int i = 0; i < 50; i++) {
+ executeQueryAsync(Integer.toString(i));
+ }
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ NamedList<String> result = (NamedList<String>) queryResponse.get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ Set<Integer> presentQueryIDs = new HashSet<>();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ presentQueryIDs.add(Integer.parseInt(entry.getKey()));
+ }
+
+ assertTrue(presentQueryIDs.size() > 0 && presentQueryIDs.size() <= 50);
+
+ Iterator<Integer> integerIterator = presentQueryIDs.iterator();
+
+ while (integerIterator.hasNext()) {
+ int value = integerIterator.next();
+
+ assertTrue (value >= 0 && value < 50);
+ }
+ }
+
+ @Test
+ public void testCheckSpecificQueryStatus() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("taskUUID", "25");
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+
+ request.setPath("/tasks/list");
+
+ NamedList<Object> queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ String result = (String) queryResponse.get("taskStatus");
+
+ assertFalse(result.contains("true"));
+ }
+
+ private CompletableFuture<Void> cancelQuery(final String queryID, final int sleepTime, Set<Integer> cancelledQueryIdsSet,
Review comment:
`sleepTime` is always 4000. Do you really want this parameterized here?
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
+ ConcurrentHashSet<Integer> notFoundIdsSet = new ConcurrentHashSet<>();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertTrue(queryIdsSet.size() + notFoundIdsSet.size() == 90);
+ }
+
+ @Test
+ public void testListCancellableQueries() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/list");
+
+ for (int i = 0; i < 50; i++) {
+ executeQueryAsync(Integer.toString(i));
+ }
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ NamedList<String> result = (NamedList<String>) queryResponse.get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ Set<Integer> presentQueryIDs = new HashSet<>();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ presentQueryIDs.add(Integer.parseInt(entry.getKey()));
+ }
+
+ assertTrue(presentQueryIDs.size() > 0 && presentQueryIDs.size() <= 50);
+
+ Iterator<Integer> integerIterator = presentQueryIDs.iterator();
Review comment:
Simplify with
```
for (int value : presentQueryIDs) {
assertTrue(value >= 0 && value < 50);
}
```
##########
File path: lucene/core/src/java/org/apache/lucene/search/CancellableCollector.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, CancellableTask {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
+
+ private Collector collector;
Review comment:
Can be final
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
Review comment:
Don't need to initialize with `null`
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
+ ConcurrentHashSet<Integer> notFoundIdsSet = new ConcurrentHashSet<>();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertTrue(queryIdsSet.size() + notFoundIdsSet.size() == 90);
+ }
+
+ @Test
+ public void testListCancellableQueries() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/list");
+
+ for (int i = 0; i < 50; i++) {
+ executeQueryAsync(Integer.toString(i));
+ }
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ NamedList<String> result = (NamedList<String>) queryResponse.get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ Set<Integer> presentQueryIDs = new HashSet<>();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ presentQueryIDs.add(Integer.parseInt(entry.getKey()));
+ }
+
+ assertTrue(presentQueryIDs.size() > 0 && presentQueryIDs.size() <= 50);
+
+ Iterator<Integer> integerIterator = presentQueryIDs.iterator();
+
+ while (integerIterator.hasNext()) {
+ int value = integerIterator.next();
+
+ assertTrue (value >= 0 && value < 50);
+ }
+ }
+
+ @Test
+ public void testCheckSpecificQueryStatus() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("taskUUID", "25");
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+
+ request.setPath("/tasks/list");
+
+ NamedList<Object> queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ String result = (String) queryResponse.get("taskStatus");
+
+ assertFalse(result.contains("true"));
+ }
+
+ private CompletableFuture<Void> cancelQuery(final String queryID, final int sleepTime, Set<Integer> cancelledQueryIdsSet,
+ Set<Integer> notFoundQueryIdSet) {
+ return CompletableFuture.runAsync(() -> {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", queryID);
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ // Wait for some time to let the query start
+ try {
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+
+ try {
+ NamedList<Object> queryResponse = null;
Review comment:
Don't need to initialize w/ `null` here either.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/QueryCancellationComponent.java
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.lucene.search.CancellableTask;
+
+import java.io.IOException;
+
+/** Responsible for handling query cancellation requests */
+public class QueryCancellationComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "querycancellation";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException
+ {
+ if (rb.isCancellation()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ String cancellationUUID = rb.getCancellationUUID();
+
+ if (cancellationUUID == null) {
+ throw new RuntimeException("Null query UUID seen");
+ }
+
+ CancellableTask cancellableTask = rb.req.getCore().getCancellableQueryTracker().getCancellableTask(cancellationUUID);
+
+ if (cancellableTask != null) {
+ cancellableTask.cancelTask();
+ rb.rsp.add("cancellationResult", "success");
+ } else {
+ rb.rsp.add("cancellationResult", "not found");
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void handleResponses(ResponseBuilder rb, ShardRequest sreq) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ boolean failureSeen = false;
+ boolean queryNotFound = false;
+
+ for (ShardResponse r : sreq.responses) {
+
+ String cancellationResult = (String) r.getSolrResponse()
+ .getResponse().get("cancellationResult");
+
+ if (!cancellationResult.equalsIgnoreCase("success")) {
+ if (cancellationResult.equalsIgnoreCase("not found")) {
+ queryNotFound = true;
+ } else {
+ failureSeen = true;
+ }
+
+ break;
+ }
+ }
+
+ if (failureSeen) {
+ rb.rsp.getValues().add("status", "Query with queryID " + rb.getCancellationUUID() +
+ " could not be cancelled successfully");
+ } else if (queryNotFound) {
+ rb.rsp.getValues().add("status", "Query with queryID " + rb.getCancellationUUID() +
+ " not found");
+ } else {
+ rb.rsp.getValues().add("status", "Query with queryID " + rb.getCancellationUUID() +
+ " cancelled successfully");
+ }
+ }
+
+ @Override
+ public String getDescription() {
+ return "querycancellation";
Review comment:
Do you want to add more to the description? Whatever be the case, it'd be good to be consistent between here and in `ActiveTasksListComponent.java`
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
+ ConcurrentHashSet<Integer> notFoundIdsSet = new ConcurrentHashSet<>();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertTrue(queryIdsSet.size() + notFoundIdsSet.size() == 90);
+ }
+
+ @Test
+ public void testListCancellableQueries() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/list");
+
+ for (int i = 0; i < 50; i++) {
+ executeQueryAsync(Integer.toString(i));
+ }
+
+ NamedList<Object> queryResponse = null;
Review comment:
Don't need to initialize as `null`
##########
File path: lucene/core/src/java/org/apache/lucene/search/CancellableCollector.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, CancellableTask {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
Review comment:
Or just reuse that?
##########
File path: lucene/core/src/test/org/apache/lucene/search/TestCancellableCollector.java
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.lucene.util.NamedThreadFactory;
+
+public class TestCancellableCollector extends LuceneTestCase {
+ Directory dir;
+ IndexReader reader;
+ IndexSearcher searcher;
+ ExecutorService executor = null;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ dir = newDirectory();
+
+ RandomIndexWriter iw = new RandomIndexWriter(random(), dir);
+ Random random = random();
+ for (int i = 0; i < 100; i++) {
+ Document doc = new Document();
+ doc.add(newStringField("field", Integer.toString(i), Field.Store.NO));
+ doc.add(newStringField("field2", Boolean.toString(i % 2 == 0), Field.Store.NO));
+ doc.add(new SortedDocValuesField("field2", new BytesRef(Boolean.toString(i % 2 == 0))));
+ iw.addDocument(doc);
+
+ if (random.nextBoolean()) {
+ iw.commit();
+ }
+ }
+ reader = iw.getReader();
+ iw.close();
+
+ searcher = new IndexSearcher(reader);
+
+ executor =
+ new ThreadPoolExecutor(
+ 4,
+ 4,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>(),
+ new NamedThreadFactory("TestIndexSearcher"));
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ reader.close();
+ dir.close();
+
+ if (executor != null) {
+ executor.shutdown();
+ }
+
+ executor = null;
+ }
+
+ private CancellableCollector buildCancellableCollector(
+ final int numHits, boolean delayStart, boolean delayCollection) {
+ TopScoreDocCollector topScoreDocCollector = TopScoreDocCollector.create(numHits, null, 1);
+ CancellableCollector collector = new CancellableCollector(topScoreDocCollector);
+
+ DummyCancellableCollector dummyCancellableCollector =
+ new DummyCancellableCollector(collector, delayStart, delayCollection);
+
+ return dummyCancellableCollector;
+ }
+
+ private void executeSearchTest(
+ IndexSearcher searcher, Query query, CancellableCollector cancellableCollector, int numHits)
+ throws Exception {
+ TopDocs topDocs = searcher.search(query, numHits);
+
+ searcher.search(query, cancellableCollector);
+
+ CancellableCollector internalCancellableCollector =
+ (CancellableCollector) cancellableCollector.getInternalCollector();
+ TopScoreDocCollector topScoreDocCollector =
+ (TopScoreDocCollector) internalCancellableCollector.getInternalCollector();
+
+ assertEquals(topDocs.totalHits.value, topScoreDocCollector.totalHits);
+ }
+
+ private void cancelQuery(CancellableCollector cancellableCollector, final int sleepTime) {
+ executor.submit(
+ () -> {
+
+ // Wait for some time to let the query start
+ try {
+ if (sleepTime > 0) {
Review comment:
The value of sleepTime being passed is also always 0. Not sure if you were planning to add something here.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.apache.solr.common.util.Utils.fromJSONString;
+
+/** List the active tasks that can be cancelled */
+public class ActiveTasksListComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "activetaskslistcomponent";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException {
+ if (rb.isTaskListRequest()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isActiveOnThisShard = rb.req.getCore().getCancellableQueryTracker().isQueryIdActive(rb.getTaskStatusCheckUUID());
+
+ rb.rsp.add("taskStatus", isActiveOnThisShard);
+ return;
+ }
+
+ rb.rsp.add("taskList", (MapWriter) ew -> {
+ Iterator<Map.Entry<String, String>> iterator = rb.req.getCore().getCancellableQueryTracker().getActiveQueriesGenerated();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+ ew.put(entry.getKey(), entry.getValue());
+ }
+ });
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void handleResponses(ResponseBuilder rb, ShardRequest sreq) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ NamedList<String> resultList = new NamedList<>();
+
+ for (ShardResponse r : sreq.responses) {
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isTaskActiveOnShard = (boolean) r.getSolrResponse().getResponse().get("taskStatus");
+
+ if (isTaskActiveOnShard == true) {
+ rb.rsp.getValues().add("taskStatus", rb.getTaskStatusCheckUUID() + ":" + isTaskActiveOnShard);
Review comment:
+1
##########
File path: solr/core/src/java/org/apache/solr/handler/component/TaskManagementHandler.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.ShardParams;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.handler.RequestHandlerBase;
+import org.apache.solr.request.SolrQueryRequest;
+import org.apache.solr.response.SolrQueryResponse;
+import org.apache.solr.security.PermissionNameProvider;
+import org.apache.solr.util.plugin.SolrCoreAware;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.solr.common.params.CommonParams.DISTRIB;
+import static org.apache.solr.common.params.CommonParams.PATH;
+
+/**
+ * Abstract class which serves as the root of all task managing handlers
+ */
+public abstract class TaskManagementHandler extends RequestHandlerBase implements SolrCoreAware, PermissionNameProvider {
+ private ShardHandlerFactory shardHandlerFactory;
+
+ @Override
+ public void inform(SolrCore core) {
+ this.shardHandlerFactory = core.getCoreContainer().getShardHandlerFactory();
+ }
+
+ /**
+ * Process the actual request.
+ * extraParams is required for allowing sub handlers to pass in custom parameters to be put in the
+ * outgoing shard request
+ */
+ protected void processRequest(SolrQueryRequest req, ResponseBuilder rb, Map<String, String> extraParams) throws IOException {
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
+ List<SearchComponent> components = rb.components;
+
+ shardHandler.prepDistributed(rb);
+
+ for(SearchComponent c : components) {
+ c.prepare(rb);
+ }
+
+ if (!rb.isDistrib) {
+ for (SearchComponent component : components) {
+ component.process(rb);
+ }
+ } else {
+ ShardRequest sreq = new ShardRequest();
+
+ // Distribute to all shards
+ sreq.shards = rb.shards;
+ sreq.actualShards = sreq.shards;
+
+ sreq.responses = new ArrayList<>(sreq.actualShards.length);
+ rb.finished = new ArrayList<>();
+
+ for (String shard : sreq.actualShards) {
+ ModifiableSolrParams params = new ModifiableSolrParams(sreq.params);
+ String reqPath = (String) req.getContext().get(PATH);
+
+ params.set(CommonParams.QT, reqPath);
+ params.remove(ShardParams.SHARDS); // not a top-level request
+ params.set(DISTRIB, "false"); // not a top-level request
+ params.remove("indent");
+ params.remove(CommonParams.HEADER_ECHO_PARAMS);
+ params.set(ShardParams.IS_SHARD, true); // a sub (shard) request
+ params.set(ShardParams.SHARDS_PURPOSE, sreq.purpose);
+ params.set(ShardParams.SHARD_URL, shard); // so the shard knows what was asked
+ params.set(CommonParams.OMIT_HEADER, false);
Review comment:
Can you reuse the code from SearchHandler ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] madrob commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
madrob commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r589676923
##########
File path: solr/core/src/test/org/apache/solr/MinimalSchemaTest.java
##########
@@ -117,7 +117,8 @@ public void testAllConfiguredHandlers() {
handler.startsWith("/terms") ||
handler.startsWith("/analysis/")||
handler.startsWith("/debug/") ||
- handler.startsWith("/replication")
+ handler.startsWith("/replication") ||
+ handler.startsWith("/tasks")
Review comment:
fix indent
##########
File path: lucene/core/src/java/org/apache/lucene/search/CancellableCollector.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, CancellableTask {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
+
+ private Collector collector;
+ private AtomicBoolean isQueryCancelled;
+
+ public CancellableCollector(Collector collector) {
+ Objects.requireNonNull(collector, "Internal collector not provided but wrapper collector accessed");
+
+ this.collector = collector;
+ this.isQueryCancelled = new AtomicBoolean();
+ }
+
+ @Override
+ public LeafCollector getLeafCollector(LeafReaderContext context) throws IOException {
+
+ if (isQueryCancelled.compareAndSet(true, false)) {
Review comment:
Why does getLeafCollector "uncancel" it? Is this collector instance meant to be reused?
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.apache.solr.common.util.Utils.fromJSONString;
+
+/** List the active tasks that can be cancelled */
+public class ActiveTasksListComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "activetaskslistcomponent";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException {
+ if (rb.isTaskListRequest()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isActiveOnThisShard = rb.req.getCore().getCancellableQueryTracker().isQueryIdActive(rb.getTaskStatusCheckUUID());
+
+ rb.rsp.add("taskStatus", isActiveOnThisShard);
+ return;
+ }
+
+ rb.rsp.add("taskList", (MapWriter) ew -> {
+ Iterator<Map.Entry<String, String>> iterator = rb.req.getCore().getCancellableQueryTracker().getActiveQueriesGenerated();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+ ew.put(entry.getKey(), entry.getValue());
+ }
+ });
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void handleResponses(ResponseBuilder rb, ShardRequest sreq) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ NamedList<String> resultList = new NamedList<>();
+
+ for (ShardResponse r : sreq.responses) {
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isTaskActiveOnShard = (boolean) r.getSolrResponse().getResponse().get("taskStatus");
+
+ if (isTaskActiveOnShard == true) {
Review comment:
simplify
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.apache.solr.common.util.Utils.fromJSONString;
+
+/** List the active tasks that can be cancelled */
+public class ActiveTasksListComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "activetaskslistcomponent";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException {
+ if (rb.isTaskListRequest()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isActiveOnThisShard = rb.req.getCore().getCancellableQueryTracker().isQueryIdActive(rb.getTaskStatusCheckUUID());
+
+ rb.rsp.add("taskStatus", isActiveOnThisShard);
+ return;
+ }
+
+ rb.rsp.add("taskList", (MapWriter) ew -> {
+ Iterator<Map.Entry<String, String>> iterator = rb.req.getCore().getCancellableQueryTracker().getActiveQueriesGenerated();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+ ew.put(entry.getKey(), entry.getValue());
+ }
+ });
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void handleResponses(ResponseBuilder rb, ShardRequest sreq) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ NamedList<String> resultList = new NamedList<>();
+
+ for (ShardResponse r : sreq.responses) {
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isTaskActiveOnShard = (boolean) r.getSolrResponse().getResponse().get("taskStatus");
+
+ if (isTaskActiveOnShard == true) {
+ rb.rsp.getValues().add("taskStatus", rb.getTaskStatusCheckUUID() + ":" + isTaskActiveOnShard);
+ return;
+ } else {
+ continue;
+ }
+ }
+
+ NamedList<String> result = (NamedList<String>) r.getSolrResponse()
+ .getResponse().get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ resultList.add(entry.getKey(), entry.getValue());
+ }
Review comment:
resultList.addAll(result)
##########
File path: lucene/core/src/java/org/apache/lucene/search/CancellableCollector.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, CancellableTask {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
Review comment:
Should this extend CollectionTerminatedException?
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,73 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface (TMI) supports the following types of operations:
+
+1. List all currently running cancellable tasks.
+2. Cancel a specific task.
+3. Query the status of a specific task.
+
+== Listing All Active Cancellable Tasks
+To list all the active cancellable tasks currently running, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list`
+
+==== Sample Response
+
+`{responseHeader={status=0, QTime=11370}, taskList={0=q=*%3A*&canCancel=true&queryUUID=0&_stateVer_=collection1%3A4&wt=javabin&version=2, 5=q=*%3A*&canCancel=true&queryUUID=5&_stateVer_=collection1%3A4&wt=javabin&version=2, 7=q=*%3A*&canCancel=true&queryUUID=7&_stateVer_=collection1%3A4&wt=javabin&version=2}`
+
+== Cancelling An Active Cancellable Task
+To cancel an active task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/cancel?cancelUUID=foobar`
+
+==== cancelUUID Parameter
+This parameter is used to specify the UUID of the task to be cancelled.
+
+==== Sample Response
+===== If the task UUID was found and successfully cancelled:
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 cancelled successfully}`
+
+===== If the task UUID was not found
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 not found}`
+
+===== If the cancellation failed
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 could not be cancelled successfully}`
Review comment:
Do we get a log message in the server logs in this case? Any more detail why could not be cancelled?
##########
File path: lucene/core/src/java/org/apache/lucene/search/CancellableTask.java
##########
@@ -0,0 +1,22 @@
+/*
+ * 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.lucene.search;
+
+/** Interface which defines a task which can be cancelled */
+public interface CancellableTask {
Review comment:
we already have `o.a.s.client.solrj.util.Cancellable` - would be nice to tie these together.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.apache.solr.common.util.Utils.fromJSONString;
+
+/** List the active tasks that can be cancelled */
+public class ActiveTasksListComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "activetaskslistcomponent";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException {
+ if (rb.isTaskListRequest()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isActiveOnThisShard = rb.req.getCore().getCancellableQueryTracker().isQueryIdActive(rb.getTaskStatusCheckUUID());
+
+ rb.rsp.add("taskStatus", isActiveOnThisShard);
+ return;
+ }
+
+ rb.rsp.add("taskList", (MapWriter) ew -> {
+ Iterator<Map.Entry<String, String>> iterator = rb.req.getCore().getCancellableQueryTracker().getActiveQueriesGenerated();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+ ew.put(entry.getKey(), entry.getValue());
+ }
+ });
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void handleResponses(ResponseBuilder rb, ShardRequest sreq) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ NamedList<String> resultList = new NamedList<>();
+
+ for (ShardResponse r : sreq.responses) {
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isTaskActiveOnShard = (boolean) r.getSolrResponse().getResponse().get("taskStatus");
Review comment:
`getBooleanArg`
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
Review comment:
unused, failing precommit here.
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
Review comment:
Use `ConcurrentHashMap.newKeySet` instead
##########
File path: solr/solr-ref-guide/src/common-query-parameters.adoc
##########
@@ -84,6 +84,18 @@ You can use the `rows` parameter to paginate results from a query. The parameter
The default value is `10`. That is, by default, Solr returns 10 documents at a time in response to a query.
+== canCancel Parameter
+
+This parameter defines if this query is cancellable i.e. can be cancelled during execution using the
+task management interface.
+
+== queryUUID Parameter
+
+For cancellable queries, this allows specifying a custom UUID to identify the query with. If `canCancel` is specified and `queryUUID` is not set, an auto generated UUID will be assigned to the query.
+
+If `queryUUID` is specified, this UUID will be used for identifying the query. Note that if using `queryUUID`, the responsibility of ensuring uniqueness of the UUID lies with the caller.
Review comment:
What happens if a uuid is accidentally reused?
##########
File path: solr/core/src/java/org/apache/solr/core/SolrCore.java
##########
@@ -3245,6 +3252,75 @@ public void postClose(SolrCore core) {
return blobRef;
}
+ /** Generates a UUID for the given query or if the user provided a UUID
+ * for this query, uses that.
+ */
+ public String generateQueryID(SolrQueryRequest req) {
+ String queryID;
+ String customQueryUUID = req.getParams().get(CUSTOM_QUERY_UUID, null);
+
+ if (customQueryUUID != null) {
+ queryID = customQueryUUID;
+ } else {
+ queryID = UUID.randomUUID().toString();
Review comment:
This doesn't need to be cryptographically secure, use a different generator here.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.apache.solr.common.util.Utils.fromJSONString;
+
+/** List the active tasks that can be cancelled */
+public class ActiveTasksListComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "activetaskslistcomponent";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException {
+ if (rb.isTaskListRequest()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isActiveOnThisShard = rb.req.getCore().getCancellableQueryTracker().isQueryIdActive(rb.getTaskStatusCheckUUID());
+
+ rb.rsp.add("taskStatus", isActiveOnThisShard);
+ return;
+ }
+
+ rb.rsp.add("taskList", (MapWriter) ew -> {
+ Iterator<Map.Entry<String, String>> iterator = rb.req.getCore().getCancellableQueryTracker().getActiveQueriesGenerated();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+ ew.put(entry.getKey(), entry.getValue());
+ }
+ });
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void handleResponses(ResponseBuilder rb, ShardRequest sreq) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ NamedList<String> resultList = new NamedList<>();
+
+ for (ShardResponse r : sreq.responses) {
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isTaskActiveOnShard = (boolean) r.getSolrResponse().getResponse().get("taskStatus");
+
+ if (isTaskActiveOnShard == true) {
+ rb.rsp.getValues().add("taskStatus", rb.getTaskStatusCheckUUID() + ":" + isTaskActiveOnShard);
Review comment:
I don't think returning UUID:true here makes semantic sense. Maybe UUID + ":active"?
##########
File path: lucene/core/src/test/org/apache/lucene/search/TestCancellableCollector.java
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.lucene.util.NamedThreadFactory;
+
+public class TestCancellableCollector extends LuceneTestCase {
+ Directory dir;
+ IndexReader reader;
+ IndexSearcher searcher;
+ ExecutorService executor = null;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ dir = newDirectory();
+
+ RandomIndexWriter iw = new RandomIndexWriter(random(), dir);
+ Random random = random();
+ for (int i = 0; i < 100; i++) {
+ Document doc = new Document();
+ doc.add(newStringField("field", Integer.toString(i), Field.Store.NO));
+ doc.add(newStringField("field2", Boolean.toString(i % 2 == 0), Field.Store.NO));
+ doc.add(new SortedDocValuesField("field2", new BytesRef(Boolean.toString(i % 2 == 0))));
+ iw.addDocument(doc);
+
+ if (random.nextBoolean()) {
+ iw.commit();
+ }
+ }
+ reader = iw.getReader();
+ iw.close();
+
+ searcher = new IndexSearcher(reader);
+
+ executor =
+ new ThreadPoolExecutor(
+ 4,
+ 4,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>(),
Review comment:
nit, use `<>`
##########
File path: lucene/core/src/test/org/apache/lucene/search/TestCancellableCollector.java
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.lucene.util.NamedThreadFactory;
+
+public class TestCancellableCollector extends LuceneTestCase {
+ Directory dir;
+ IndexReader reader;
+ IndexSearcher searcher;
+ ExecutorService executor = null;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
Review comment:
Is a lot of this copied from LuceneTestCase? Can we reuse that code better instead?
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
+ ConcurrentHashSet<Integer> notFoundIdsSet = new ConcurrentHashSet<>();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertTrue(queryIdsSet.size() + notFoundIdsSet.size() == 90);
Review comment:
nit: assertEquals, and include a message
##########
File path: lucene/core/src/test/org/apache/lucene/search/TestCancellableCollector.java
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.lucene.util.NamedThreadFactory;
+
+public class TestCancellableCollector extends LuceneTestCase {
+ Directory dir;
+ IndexReader reader;
+ IndexSearcher searcher;
+ ExecutorService executor = null;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ dir = newDirectory();
+
+ RandomIndexWriter iw = new RandomIndexWriter(random(), dir);
+ Random random = random();
+ for (int i = 0; i < 100; i++) {
+ Document doc = new Document();
+ doc.add(newStringField("field", Integer.toString(i), Field.Store.NO));
+ doc.add(newStringField("field2", Boolean.toString(i % 2 == 0), Field.Store.NO));
+ doc.add(new SortedDocValuesField("field2", new BytesRef(Boolean.toString(i % 2 == 0))));
+ iw.addDocument(doc);
+
+ if (random.nextBoolean()) {
+ iw.commit();
+ }
+ }
+ reader = iw.getReader();
+ iw.close();
+
+ searcher = new IndexSearcher(reader);
+
+ executor =
+ new ThreadPoolExecutor(
+ 4,
+ 4,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>(),
+ new NamedThreadFactory("TestIndexSearcher"));
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ reader.close();
+ dir.close();
+
+ if (executor != null) {
+ executor.shutdown();
+ }
+
+ executor = null;
+ }
+
+ private CancellableCollector buildCancellableCollector(
+ final int numHits, boolean delayStart, boolean delayCollection) {
+ TopScoreDocCollector topScoreDocCollector = TopScoreDocCollector.create(numHits, null, 1);
+ CancellableCollector collector = new CancellableCollector(topScoreDocCollector);
+
+ DummyCancellableCollector dummyCancellableCollector =
+ new DummyCancellableCollector(collector, delayStart, delayCollection);
+
+ return dummyCancellableCollector;
+ }
+
+ private void executeSearchTest(
+ IndexSearcher searcher, Query query, CancellableCollector cancellableCollector, int numHits)
+ throws Exception {
+ TopDocs topDocs = searcher.search(query, numHits);
+
+ searcher.search(query, cancellableCollector);
+
+ CancellableCollector internalCancellableCollector =
+ (CancellableCollector) cancellableCollector.getInternalCollector();
+ TopScoreDocCollector topScoreDocCollector =
+ (TopScoreDocCollector) internalCancellableCollector.getInternalCollector();
+
+ assertEquals(topDocs.totalHits.value, topScoreDocCollector.totalHits);
+ }
+
+ private void cancelQuery(CancellableCollector cancellableCollector, final int sleepTime) {
+ executor.submit(
+ () -> {
+
+ // Wait for some time to let the query start
+ try {
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+
+ cancellableCollector.cancelTask();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e.getMessage());
+ }
+ });
+ }
+
+ public void testSearchWithoutCancellation() throws Exception {
+ CancellableCollector cancellableCollector = buildCancellableCollector(50, false, false);
+
+ Query query = new TermQuery(new Term("field", "1"));
+
+ executeSearchTest(searcher, query, cancellableCollector, 50);
+
+ query = new MatchAllDocsQuery();
+
+ cancellableCollector = buildCancellableCollector(100, false, false);
+
+ executeSearchTest(searcher, query, cancellableCollector, 50);
+ }
+
+ public void testSearchWithCancellationBeforeActualDocumentCollection() {
+ Query query = new MatchAllDocsQuery();
+
+ CancellableCollector cancellableCollector = buildCancellableCollector(5000, true, false);
+
+ expectThrows(
+ CancellableCollector.QueryCancelledException.class,
+ () -> {
+ // Cancel the query before the document collection starts
+ cancelQuery(cancellableCollector, 0);
+
+ executeSearchTest(searcher, query, cancellableCollector, 5000);
+ });
+ }
+
+ public void testSearchWithCancellationBetweenActualDocumentCollection() {
+ Query query = new MatchAllDocsQuery();
+
+ CancellableCollector cancellableCollector = buildCancellableCollector(5000, false, true);
+
+ expectThrows(
+ CancellableCollector.QueryCancelledException.class,
+ () -> {
+ // Cancel the query before the document collection starts
+ cancelQuery(cancellableCollector, 0);
+
+ executeSearchTest(searcher, query, cancellableCollector, 5000);
+ });
+ }
+
+ public class DummyCancellableCollector extends CancellableCollector {
Review comment:
class should be static
##########
File path: solr/core/src/test/org/apache/solr/search/TestTaskManagement.java
##########
@@ -0,0 +1,263 @@
+/*
+ * 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.solr.search;
+
+import org.apache.lucene.util.BytesRef;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
+import org.eclipse.jetty.util.ConcurrentHashSet;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+public class TestTaskManagement extends SolrCloudTestCase {
+ private static final String COLLECTION_NAME = "collection1";
+
+ private ExecutorService executorService;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ initCore("solrconfig.xml", "schema11.xml");
+
+ configureCluster(4)
+ .addConfig("conf", configset("sql"))
+ .configure();
+ }
+
+ @AfterClass
+ public static void tearDownCluster() throws Exception {
+ shutdownCluster();
+ }
+
+ @Before
+ public void setup() throws Exception {
+ super.setUp();
+
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 1)
+ .setPerReplicaState(SolrCloudTestCase.USE_PER_REPLICA_STATE)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 2);
+ cluster.getSolrClient().setDefaultCollection(COLLECTION_NAME);
+
+ cluster.getSolrClient().setDefaultCollection("collection1");
+
+ executorService = ExecutorUtil.newMDCAwareCachedThreadPool("TestTaskManagement");
+
+ List<SolrInputDocument> docs = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", i);
+ doc.addField("foo1_s", Integer.toString(i));
+ doc.addField("foo2_s", Boolean.toString(i % 2 == 0));
+ doc.addField("foo4_s", new BytesRef(Boolean.toString(i % 2 == 0)));
+
+ docs.add(doc);
+ }
+
+ cluster.getSolrClient().add(docs);
+ cluster.getSolrClient().commit();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ CollectionAdminRequest.deleteCollection(COLLECTION_NAME).process(cluster.getSolrClient());
+ executorService.shutdown();
+ super.tearDown();
+ }
+
+ @Test
+ public void testNonExistentQuery() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", "foobar");
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ assertEquals("Query with queryID foobar not found", queryResponse.get("status"));
+ }
+
+ @Test
+ public void testCancellationQuery() throws Exception {
+ ConcurrentHashSet<Integer> queryIdsSet = new ConcurrentHashSet<>();
+ ConcurrentHashSet<Integer> notFoundIdsSet = new ConcurrentHashSet<>();
+
+ List<CompletableFuture<Void>> queryFutures = new ArrayList<>();
+
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future = executeQueryAsync(Integer.toString(i));
+
+ queryFutures.add(future);
+ }
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < 90; i++) {
+ CompletableFuture<Void> future = cancelQuery(Integer.toString(i), 4000, queryIdsSet, notFoundIdsSet);
+
+ futures.add(future);
+ }
+
+ futures.forEach(CompletableFuture::join);
+
+ queryFutures.forEach(CompletableFuture::join);
+
+ assertTrue(queryIdsSet.size() + notFoundIdsSet.size() == 90);
+ }
+
+ @Test
+ public void testListCancellableQueries() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/list");
+
+ for (int i = 0; i < 50; i++) {
+ executeQueryAsync(Integer.toString(i));
+ }
+
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ NamedList<String> result = (NamedList<String>) queryResponse.get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ Set<Integer> presentQueryIDs = new HashSet<>();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ presentQueryIDs.add(Integer.parseInt(entry.getKey()));
+ }
+
+ assertTrue(presentQueryIDs.size() > 0 && presentQueryIDs.size() <= 50);
+
+ Iterator<Integer> integerIterator = presentQueryIDs.iterator();
+
+ while (integerIterator.hasNext()) {
+ int value = integerIterator.next();
+
+ assertTrue (value >= 0 && value < 50);
+ }
+ }
+
+ @Test
+ public void testCheckSpecificQueryStatus() throws Exception {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("taskUUID", "25");
+
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+
+ request.setPath("/tasks/list");
+
+ NamedList<Object> queryResponse = cluster.getSolrClient().request(request);
+
+ @SuppressWarnings({"unchecked"})
+ String result = (String) queryResponse.get("taskStatus");
+
+ assertFalse(result.contains("true"));
+ }
+
+ private CompletableFuture<Void> cancelQuery(final String queryID, final int sleepTime, Set<Integer> cancelledQueryIdsSet,
+ Set<Integer> notFoundQueryIdSet) {
+ return CompletableFuture.runAsync(() -> {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+
+ params.set("cancelUUID", queryID);
+ @SuppressWarnings({"rawtypes"})
+ SolrRequest request = new QueryRequest(params);
+ request.setPath("/tasks/cancel");
+
+ // Wait for some time to let the query start
+ try {
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+
+ try {
+ NamedList<Object> queryResponse = null;
+
+ queryResponse = cluster.getSolrClient().request(request);
+
+ String cancellationResult = (String) queryResponse.get("status");
+ if (cancellationResult.contains("cancelled successfully")) {
+ cancelledQueryIdsSet.add(Integer.parseInt(queryID));
+ } else if (cancellationResult.contains("not found")) {
+ notFoundQueryIdSet.add(Integer.parseInt(queryID));
+ }
Review comment:
If we're having to do string comparisons like this, then that suggests the API is deficient in some way. We should return this as part of the response better. Maybe an enum or other kinds of well known values.
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,73 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface (TMI) supports the following types of operations:
+
+1. List all currently running cancellable tasks.
+2. Cancel a specific task.
+3. Query the status of a specific task.
+
+== Listing All Active Cancellable Tasks
+To list all the active cancellable tasks currently running, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list`
+
+==== Sample Response
+
+`{responseHeader={status=0, QTime=11370}, taskList={0=q=*%3A*&canCancel=true&queryUUID=0&_stateVer_=collection1%3A4&wt=javabin&version=2, 5=q=*%3A*&canCancel=true&queryUUID=5&_stateVer_=collection1%3A4&wt=javabin&version=2, 7=q=*%3A*&canCancel=true&queryUUID=7&_stateVer_=collection1%3A4&wt=javabin&version=2}`
+
+== Cancelling An Active Cancellable Task
+To cancel an active task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/cancel?cancelUUID=foobar`
+
+==== cancelUUID Parameter
+This parameter is used to specify the UUID of the task to be cancelled.
+
+==== Sample Response
+===== If the task UUID was found and successfully cancelled:
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 cancelled successfully}`
+
+===== If the task UUID was not found
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 not found}`
Review comment:
This is the case where the query has already completed, right?
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,73 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface (TMI) supports the following types of operations:
Review comment:
We don't need an acronym for it :)
##########
File path: solr/core/src/java/org/apache/solr/handler/component/QueryComponent.java
##########
@@ -1531,6 +1567,26 @@ private void doProcessUngroupedSearch(ResponseBuilder rb, QueryCommand cmd, Quer
doPrefetch(rb);
}
+ private static String generateQueryID(SolrQueryRequest req) {
+ ZkController zkController = req.getCore().getCoreContainer().getZkController();
+ String nodeName = req.getCore().getCoreContainer().getHostName();
+
+ if (zkController != null) {
+ nodeName = req.getCore().getCoreContainer().
Review comment:
use controller directly, don't need to chain through the core again
##########
File path: solr/core/src/java/org/apache/solr/handler/component/QueryCancellationComponent.java
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.lucene.search.CancellableTask;
+
+import java.io.IOException;
+
+/** Responsible for handling query cancellation requests */
+public class QueryCancellationComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "querycancellation";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException
+ {
+ if (rb.isCancellation()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ String cancellationUUID = rb.getCancellationUUID();
+
+ if (cancellationUUID == null) {
+ throw new RuntimeException("Null query UUID seen");
+ }
+
+ CancellableTask cancellableTask = rb.req.getCore().getCancellableQueryTracker().getCancellableTask(cancellationUUID);
+
+ if (cancellableTask != null) {
+ cancellableTask.cancelTask();
+ rb.rsp.add("cancellationResult", "success");
+ } else {
+ rb.rsp.add("cancellationResult", "not found");
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void handleResponses(ResponseBuilder rb, ShardRequest sreq) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ boolean failureSeen = false;
+ boolean queryNotFound = false;
+
+ for (ShardResponse r : sreq.responses) {
+
+ String cancellationResult = (String) r.getSolrResponse()
+ .getResponse().get("cancellationResult");
+
+ if (!cancellationResult.equalsIgnoreCase("success")) {
+ if (cancellationResult.equalsIgnoreCase("not found")) {
+ queryNotFound = true;
+ } else {
+ failureSeen = true;
+ }
+
+ break;
+ }
+ }
+
+ if (failureSeen) {
+ rb.rsp.getValues().add("status", "Query with queryID " + rb.getCancellationUUID() +
+ " could not be cancelled successfully");
+ } else if (queryNotFound) {
+ rb.rsp.getValues().add("status", "Query with queryID " + rb.getCancellationUUID() +
+ " not found");
+ } else {
+ rb.rsp.getValues().add("status", "Query with queryID " + rb.getCancellationUUID() +
+ " cancelled successfully");
+ }
+ }
+
+ @Override
+ public String getDescription() {
+ return "querycancellation";
Review comment:
return COMPONENT_NAME
##########
File path: lucene/core/src/test/org/apache/lucene/search/TestCancellableCollector.java
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.SortedDocValuesField;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.lucene.util.NamedThreadFactory;
+
+public class TestCancellableCollector extends LuceneTestCase {
+ Directory dir;
+ IndexReader reader;
+ IndexSearcher searcher;
+ ExecutorService executor = null;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ dir = newDirectory();
+
+ RandomIndexWriter iw = new RandomIndexWriter(random(), dir);
+ Random random = random();
+ for (int i = 0; i < 100; i++) {
+ Document doc = new Document();
+ doc.add(newStringField("field", Integer.toString(i), Field.Store.NO));
+ doc.add(newStringField("field2", Boolean.toString(i % 2 == 0), Field.Store.NO));
+ doc.add(new SortedDocValuesField("field2", new BytesRef(Boolean.toString(i % 2 == 0))));
+ iw.addDocument(doc);
+
+ if (random.nextBoolean()) {
+ iw.commit();
+ }
+ }
+ reader = iw.getReader();
+ iw.close();
+
+ searcher = new IndexSearcher(reader);
+
+ executor =
+ new ThreadPoolExecutor(
+ 4,
+ 4,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>(),
+ new NamedThreadFactory("TestIndexSearcher"));
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ reader.close();
+ dir.close();
+
+ if (executor != null) {
+ executor.shutdown();
+ }
+
+ executor = null;
+ }
+
+ private CancellableCollector buildCancellableCollector(
+ final int numHits, boolean delayStart, boolean delayCollection) {
+ TopScoreDocCollector topScoreDocCollector = TopScoreDocCollector.create(numHits, null, 1);
+ CancellableCollector collector = new CancellableCollector(topScoreDocCollector);
+
+ DummyCancellableCollector dummyCancellableCollector =
+ new DummyCancellableCollector(collector, delayStart, delayCollection);
+
+ return dummyCancellableCollector;
+ }
+
+ private void executeSearchTest(
+ IndexSearcher searcher, Query query, CancellableCollector cancellableCollector, int numHits)
+ throws Exception {
+ TopDocs topDocs = searcher.search(query, numHits);
+
+ searcher.search(query, cancellableCollector);
+
+ CancellableCollector internalCancellableCollector =
+ (CancellableCollector) cancellableCollector.getInternalCollector();
+ TopScoreDocCollector topScoreDocCollector =
+ (TopScoreDocCollector) internalCancellableCollector.getInternalCollector();
+
+ assertEquals(topDocs.totalHits.value, topScoreDocCollector.totalHits);
+ }
+
+ private void cancelQuery(CancellableCollector cancellableCollector, final int sleepTime) {
+ executor.submit(
+ () -> {
+
+ // Wait for some time to let the query start
+ try {
+ if (sleepTime > 0) {
Review comment:
I don't understand when we want to use this. It relies on non-deterministic scheduling in the threads for the test? If we need to ensure that a task has started, have it trigger a latch that the code calling this method is waiting on.
##########
File path: solr/solr-ref-guide/src/task-management.adoc
##########
@@ -0,0 +1,73 @@
+= Task Management
+// 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.
+
+Solr allows users to control their running tasks by monitoring them, specifying tasks as cancellation enabled and allowing
+cancellation of the same.
+
+This is achieved using the task management interface. Currently, this is supported for queries.
+
+== Types of Operations
+Task management interface (TMI) supports the following types of operations:
+
+1. List all currently running cancellable tasks.
+2. Cancel a specific task.
+3. Query the status of a specific task.
+
+== Listing All Active Cancellable Tasks
+To list all the active cancellable tasks currently running, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list`
+
+==== Sample Response
+
+`{responseHeader={status=0, QTime=11370}, taskList={0=q=*%3A*&canCancel=true&queryUUID=0&_stateVer_=collection1%3A4&wt=javabin&version=2, 5=q=*%3A*&canCancel=true&queryUUID=5&_stateVer_=collection1%3A4&wt=javabin&version=2, 7=q=*%3A*&canCancel=true&queryUUID=7&_stateVer_=collection1%3A4&wt=javabin&version=2}`
+
+== Cancelling An Active Cancellable Task
+To cancel an active task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/cancel?cancelUUID=foobar`
+
+==== cancelUUID Parameter
+This parameter is used to specify the UUID of the task to be cancelled.
+
+==== Sample Response
+===== If the task UUID was found and successfully cancelled:
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 cancelled successfully}`
+
+===== If the task UUID was not found
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 not found}`
+
+===== If the cancellation failed
+
+`{responseHeader={status=0, QTime=39}, status=Query with queryID 85 could not be cancelled successfully}`
+
+== Check Status of a Specific Task
+To check the status of a specific task, please use the following syntax:
+
+`\http://localhost:8983/solr/tasks/list?taskUUID=foobar`
+
+==== taskUUID Parameter
+`taskUUID` parameter can be used to specify a task UUID whose status can be checked.
+
+==== Sample Response
+`{responseHeader={status=0, QTime=6128}, taskStatus=foobar:true}`
Review comment:
this does not feel meaningful. It's just an existence check?
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+import static java.util.Arrays.asList;
+import static org.apache.solr.common.util.Utils.fromJSONString;
+
+/** List the active tasks that can be cancelled */
+public class ActiveTasksListComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "activetaskslistcomponent";
Review comment:
name should not end in component
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] sigram commented on a change in pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
sigram commented on a change in pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#discussion_r582707248
##########
File path: lucene/core/src/java/org/apache/lucene/search/CancellableCollector.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, CancellableTask {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
+
+ private Collector collector;
+ private AtomicBoolean isQueryCancelled;
+
+ public CancellableCollector(Collector collector) {
+ if (collector == null) {
+ throw new IllegalStateException(
+ "Internal collector not provided but wrapper collector accessed");
+ }
+
+ this.collector = collector;
+ this.isQueryCancelled = new AtomicBoolean();
+ }
+
+ @Override
+ public LeafCollector getLeafCollector(LeafReaderContext context) throws IOException {
+
+ if (isQueryCancelled.compareAndSet(true, false)) {
+ throw new QueryCancelledException();
+ }
+
+ return new FilterLeafCollector(collector.getLeafCollector(context)) {
+
+ @Override
+ public void collect(int doc) throws IOException {
+ if (isQueryCancelled.compareAndSet(true, false)) {
+ throw new QueryCancelledException();
+ }
+
Review comment:
Pretty sure some of the whitespace in this method could be removed ;)
##########
File path: solr/core/src/java/org/apache/solr/core/SolrCore.java
##########
@@ -3245,6 +3252,75 @@ public void postClose(SolrCore core) {
return blobRef;
}
+ /** Generates a UUID for the given query or if the user provided a UUID
+ * for this query, uses that.
+ */
+ public String generateQueryID(SolrQueryRequest req) {
+ String queryID;
+ String customQueryUUID = req.getParams().get(CUSTOM_QUERY_UUID, null);
+
+ if (customQueryUUID != null) {
+ queryID = customQueryUUID;
+ } else {
+ queryID = UUID.randomUUID().toString();
+ }
+
+ if (activeQueriesGenerated.containsKey(queryID)) {
+ if (customQueryUUID != null) {
+ throw new IllegalArgumentException("Duplicate query UUID given");
+ } else {
+ while (activeQueriesGenerated.get(queryID) != null) {
+ queryID = UUID.randomUUID().toString();
+ }
+ }
+ }
+
+ activeQueriesGenerated.put(queryID, req.getHttpSolrCall().getReq().getQueryString());
+
+ return queryID;
+ }
+
+ public void releaseQueryID(String inputQueryID) {
+ if (inputQueryID == null) {
+ return;
+ }
+
+ activeQueriesGenerated.remove(inputQueryID);
+ }
+
+ public boolean isQueryIdActive(String queryID) {
+ return activeQueriesGenerated.containsKey(queryID);
+ }
+
+ public void addShardLevelActiveQuery(String queryID, CancellableCollector collector) {
+ if (queryID == null) {
+ return;
+ }
+
+ activeCancellableQueries.put(queryID, collector);
+ }
+
+ public CancellableTask getCancellableTask(String queryID) {
+ if (queryID == null) {
+ throw new IllegalArgumentException("Input queryID is null");
+ }
+
+ return activeCancellableQueries.get(queryID);
+ }
+
+ public void removeCancellableQuery(String queryID) {
+ if (queryID == null) {
+ // Some components, such as CaffeineCache, use the searcher to fire internal queries which are not tracked
+ return;
+ }
+
+ activeCancellableQueries.remove(queryID);
+ }
+
+ public Iterator<Map.Entry<String, String>> getActiveQueriesGenerated() {
+ return activeQueriesGenerated.entrySet().iterator();
+ }
+
Review comment:
All these methods add a significant API surface to the already bloated SolrCore. I suggest putting all of the methods and the tracking maps into a utility class, say `CancellableQueryTracker`.
##########
File path: solr/solrj/src/java/org/apache/solr/client/solrj/response/QueryResponse.java
##########
@@ -184,6 +187,15 @@ else if ( "terms".equals( n ) ) {
else if ( "moreLikeThis".equals( n ) ) {
_moreLikeThisInfo = (NamedList<SolrDocumentList>) res.getVal( i );
}
+ else if ("taskList".equals( n )) {
Review comment:
Do we need three separate sections for these? I don't think it's possible to return all three in the same call, it's either this or that, so one section eg. `taskInfo` should be enough?
##########
File path: solr/solrj/src/java/org/apache/solr/common/params/CommonParams.java
##########
@@ -160,6 +160,26 @@
*/
String TIME_ALLOWED = "timeAllowed";
+ /**
+ * Is the query cancellable?
+ */
+ String IS_QUERY_CANCELLABLE = "canCancel";
+
+ /**
+ * Custom query UUID if provided.
+ */
+ String CUSTOM_QUERY_UUID = "queryUUID";
Review comment:
We can drop the CUSTOM prefix here.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/TaskManagementHandler.java
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.ShardParams;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.handler.RequestHandlerBase;
+import org.apache.solr.request.SolrQueryRequest;
+import org.apache.solr.response.SolrQueryResponse;
+import org.apache.solr.security.PermissionNameProvider;
+import org.apache.solr.util.plugin.SolrCoreAware;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.solr.common.params.CommonParams.DISTRIB;
+import static org.apache.solr.common.params.CommonParams.PATH;
+
+/**
+ * Abstract class which serves as the root of all task managing handlers
+ */
+public abstract class TaskManagementHandler extends RequestHandlerBase implements SolrCoreAware, PermissionNameProvider {
+ private ShardHandlerFactory shardHandlerFactory;
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public abstract void handleRequestBody(SolrQueryRequest req, SolrQueryResponse rsp) throws Exception;
Review comment:
There's no need to override this abstract method here.
##########
File path: solr/solrj/src/java/org/apache/solr/common/params/CommonParams.java
##########
@@ -160,6 +160,26 @@
*/
String TIME_ALLOWED = "timeAllowed";
+ /**
+ * Is the query cancellable?
+ */
+ String IS_QUERY_CANCELLABLE = "canCancel";
+
+ /**
+ * Custom query UUID if provided.
+ */
+ String CUSTOM_QUERY_UUID = "queryUUID";
+
+ /**
+ * UUID for query to be cancelled
+ */
+ String QUERY_CANCELLATION_UUID = "cancelUUID";
Review comment:
Hmm. Why do we need to use a separate parameter name? We either submit a new query (and it's handled by one handler) or cancel a query (and it's handled by a completely separate handler) so there's no danger of confusion. I think we should just use `queryUUID` in all places.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ResponseBuilder.java
##########
@@ -65,12 +65,20 @@
public boolean doAnalytics;
public MergeStrategy mergeFieldHandler;
+ public String queryID;
+
private boolean needDocList = false;
private boolean needDocSet = false;
private int fieldFlags = 0;
//private boolean debug = false;
private boolean debugTimings, debugQuery, debugResults, debugTrack;
+ private boolean isCancellation;
+ private String cancellationUUID;
+
+ private boolean isTaskListRequest;
+ private String taskStatusCheckUUID;
Review comment:
These could be probably limited to just a presence / absence of the UUID, without the additional boolean flag - just check that the corresponding UUID exists, and this will imply that the flag is turned on.
##########
File path: solr/solrj/src/java/org/apache/solr/common/params/ShardParams.java
##########
@@ -28,6 +28,9 @@
public interface ShardParams {
/** the shards to use (distributed configuration) */
String SHARDS = "shards";
+
+ /** UUID of the query */
+ String QUERY_ID = "queryID";
Review comment:
Why not `queryUUID` (and reference the same constant as in other places)?
##########
File path: solr/core/src/java/org/apache/solr/search/SolrIndexSearcher.java
##########
@@ -224,6 +232,11 @@ private Collector buildAndRunCollectorChain(QueryResult qr, Query query, Collect
if (collector instanceof DelegatingCollector) {
((DelegatingCollector) collector).finish();
}
+
+ if (cmd.isQueryCancellable()) {
Review comment:
This code should go under the `finally` section to make sure we always properly handle de-registration of queries even if they throw exceptions.
##########
File path: solr/solrj/src/java/org/apache/solr/common/params/CommonParams.java
##########
@@ -160,6 +160,26 @@
*/
String TIME_ALLOWED = "timeAllowed";
+ /**
+ * Is the query cancellable?
+ */
+ String IS_QUERY_CANCELLABLE = "canCancel";
+
+ /**
+ * Custom query UUID if provided.
+ */
+ String CUSTOM_QUERY_UUID = "queryUUID";
+
+ /**
+ * UUID for query to be cancelled
+ */
+ String QUERY_CANCELLATION_UUID = "cancelUUID";
+
+ /**
+ * UUID of the task whose status is to be checked
+ */
+ String TASK_CHECK_UUID = "taskUUID";
Review comment:
We may leave this as is (although I would be tempted to use `queryUUID` here too) if you think we will have other types of cancellable tasks in the future.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+/** List the active tasks that can be cancelled */
+public class ActiveTasksListComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "activetaskslistcomponent";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException {
+ if (rb.isTaskListRequest()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isActiveOnThisShard = rb.req.getCore().isQueryIdActive(rb.getTaskStatusCheckUUID());
+
+ rb.rsp.add("taskStatus", isActiveOnThisShard);
+ return;
+ }
+
+ NamedList<String> temp = new NamedList<>();
+
+ Iterator<Map.Entry<String, String>> iterator = rb.req.getCore().getActiveQueriesGenerated();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+ temp.add(entry.getKey(), entry.getValue());
+ }
+
+ rb.rsp.add("taskList", temp);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void handleResponses(ResponseBuilder rb, ShardRequest sreq) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ NamedList<String> resultList = new NamedList<>();
+
+ for (ShardResponse r : sreq.responses) {
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isTaskActiveOnShard = (boolean) r.getSolrResponse().getResponse().get("taskStatus");
+
+ if (isTaskActiveOnShard == true) {
+ rb.rsp.getValues().add("taskStatus", rb.getTaskStatusCheckUUID() + ":" + isTaskActiveOnShard);
+ return;
+ } else {
+ continue;
+ }
+ }
+
+ NamedList<String> result = (NamedList<String>) r.getSolrResponse()
+ .getResponse().get("taskList");
+
+ Iterator<Map.Entry<String, String>> iterator = result.iterator();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
+ resultList.add(entry.getKey(), entry.getValue());
+ }
+ }
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ // We got here with the specific taskID check being specified -- this means that the taskID was not
+ // found in active tasks on any shard
+ rb.rsp.getValues().add("taskStatus", rb.getTaskStatusCheckUUID() + ":" + false);
+ return;
+ }
+
+ rb.rsp.getValues().add("taskList", resultList);
+ }
+
+ @Override
+ public String getDescription() {
+ return "activetaskslist";
Review comment:
CamelCase? or just use a human-readable description.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/TaskManagementHandler.java
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.ShardParams;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.handler.RequestHandlerBase;
+import org.apache.solr.request.SolrQueryRequest;
+import org.apache.solr.response.SolrQueryResponse;
+import org.apache.solr.security.PermissionNameProvider;
+import org.apache.solr.util.plugin.SolrCoreAware;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.solr.common.params.CommonParams.DISTRIB;
+import static org.apache.solr.common.params.CommonParams.PATH;
+
+/**
+ * Abstract class which serves as the root of all task managing handlers
+ */
+public abstract class TaskManagementHandler extends RequestHandlerBase implements SolrCoreAware, PermissionNameProvider {
+ private ShardHandlerFactory shardHandlerFactory;
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public abstract void handleRequestBody(SolrQueryRequest req, SolrQueryResponse rsp) throws Exception;
+
+ @Override
+ public void inform(SolrCore core) {
+ this.shardHandlerFactory = core.getCoreContainer().getShardHandlerFactory();
+ }
+
+ /**
+ * Process the actual request.
+ * extraParams is required for allowing sub handlers to pass in custom parameters to be put in the
+ * outgoing shard request
+ */
+ protected void processRequest(SolrQueryRequest req, ResponseBuilder rb, Map<String, String> extraParams) throws IOException {
+ ShardHandler shardHandler = shardHandlerFactory.getShardHandler();
+ List<SearchComponent> components = rb.components;
+
+ shardHandler.prepDistributed(rb);
+
+ for(SearchComponent c : components) {
+ c.prepare(rb);
+ }
+
+ if (!rb.isDistrib) {
+ for (SearchComponent component : components) {
+ component.process(rb);
+ }
+ } else {
+ ShardRequest sreq = new ShardRequest();
+
+ // Distribute to all shards
+ sreq.shards = rb.shards;
+ sreq.actualShards = sreq.shards;
+
+ sreq.responses = new ArrayList<>(sreq.actualShards.length);
+ rb.finished = new ArrayList<>();
+
+ for (String shard : sreq.actualShards) {
+ ModifiableSolrParams params = new ModifiableSolrParams(sreq.params);
+ String reqPath = (String) req.getContext().get(PATH);
+
+ params.set(CommonParams.QT, reqPath);
+ params.remove(ShardParams.SHARDS); // not a top-level request
+ params.set(DISTRIB, "false"); // not a top-level request
+ params.remove("indent");
+ params.remove(CommonParams.HEADER_ECHO_PARAMS);
+ params.set(ShardParams.IS_SHARD, true); // a sub (shard) request
+ params.set(ShardParams.SHARDS_PURPOSE, sreq.purpose);
+ params.set(ShardParams.SHARD_URL, shard); // so the shard knows what was asked
+ params.set(CommonParams.OMIT_HEADER, false);
+
+ if (extraParams != null) {
+ Iterator<Map.Entry<String, String>> iterator = extraParams.entrySet().iterator();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+
Review comment:
Too much whitespace..
##########
File path: solr/core/src/java/org/apache/solr/handler/component/ActiveTasksListComponent.java
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.solr.handler.component;
+
+import org.apache.solr.common.util.NamedList;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+/** List the active tasks that can be cancelled */
+public class ActiveTasksListComponent extends SearchComponent {
+ public static final String COMPONENT_NAME = "activetaskslistcomponent";
+
+ private boolean shouldProcess;
+
+ @Override
+ public void prepare(ResponseBuilder rb) throws IOException {
+ if (rb.isTaskListRequest()) {
+ shouldProcess = true;
+ }
+ }
+
+ @Override
+ public void process(ResponseBuilder rb) {
+ if (!shouldProcess) {
+ return;
+ }
+
+ if (rb.getTaskStatusCheckUUID() != null) {
+ boolean isActiveOnThisShard = rb.req.getCore().isQueryIdActive(rb.getTaskStatusCheckUUID());
+
+ rb.rsp.add("taskStatus", isActiveOnThisShard);
+ return;
+ }
+
+ NamedList<String> temp = new NamedList<>();
+
+ Iterator<Map.Entry<String, String>> iterator = rb.req.getCore().getActiveQueriesGenerated();
+
+ while (iterator.hasNext()) {
+ Map.Entry<String, String> entry = iterator.next();
+ temp.add(entry.getKey(), entry.getValue());
+ }
+
+ rb.rsp.add("taskList", temp);
Review comment:
you could use `MapWriter` here to avoid allocating the NamedList.
##########
File path: solr/core/src/java/org/apache/solr/handler/component/SearchHandler.java
##########
@@ -435,6 +435,9 @@ public void handleRequestBody(SolrQueryRequest req, SolrQueryResponse rsp) throw
params.set(ShardParams.SHARDS_PURPOSE, sreq.purpose);
params.set(ShardParams.SHARD_URL, shard); // so the shard knows what was asked
params.set(CommonParams.OMIT_HEADER, false);
+
+ // Distributed request -- need to send queryID as a part of the distributed request
+ params.set(ShardParams.QUERY_ID, rb.queryID);
Review comment:
`params.setNonNull`.
##########
File path: lucene/core/src/java/org/apache/lucene/search/CancellableCollector.java
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.lucene.search;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.lucene.index.LeafReaderContext;
+
+/** Allows a query to be cancelled */
+public class CancellableCollector implements Collector, CancellableTask {
+
+ /** Thrown when a query gets cancelled */
+ public static class QueryCancelledException extends RuntimeException {}
+
+ private Collector collector;
+ private AtomicBoolean isQueryCancelled;
+
+ public CancellableCollector(Collector collector) {
+ if (collector == null) {
Review comment:
Maybe use `Objects.requireNonNull` instead?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
[GitHub] [lucene-solr] atris commented on pull request #2403: SOLR-15164: Implement Task Management Interface
Posted by GitBox <gi...@apache.org>.
atris commented on pull request #2403:
URL: https://github.com/apache/lucene-solr/pull/2403#issuecomment-797319603
@sigram @madrob @anshumg I missed some comments in the previous iteration -- fixed all comments now. Please see and share your thoughts and comments.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org