You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2022/11/11 06:29:54 UTC
[GitHub] [druid] cryptoe opened a new pull request, #13353: [DRAFT] Worker retry for MSQ task
cryptoe opened a new pull request, #13353:
URL: https://github.com/apache/druid/pull/13353
<!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
<!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
Fixes #XXXX.
<!-- Replace XXXX with the id of the issue fixed in this PR. Remove this section if there is no corresponding issue. Don't reference the issue in the title of this pull-request. -->
<!-- If you are a committer, follow the PR action item checklist for committers:
https://github.com/apache/druid/blob/master/dev/committer-instructions.md#pr-and-issue-action-item-checklist-for-committers. -->
### Description
<!-- Describe the goal of this PR, what problem are you fixing. If there is a corresponding issue (referenced above), it's not necessary to repeat the description here, however, you may choose to keep one summary sentence. -->
<!-- Describe your patch: what did you change in code? How did you fix the problem? -->
<!-- If there are several relatively logically separate changes in this PR, create a mini-section for each of them. For example: -->
#### Fixed the bug ...
#### Renamed the class ...
#### Added a forbidden-apis entry ...
<!--
In each section, please describe design decisions made, including:
- Choice of algorithms
- Behavioral aspects. What configuration values are acceptable? How are corner cases and error conditions handled, such as when there are insufficient resources?
- Class organization and design (how the logic is split between classes, inheritance, composition, design patterns)
- Method organization and design (how the logic is split between methods, parameters and return types)
- Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
-->
<!-- It's good to describe an alternative design (or mention an alternative name) for every design (or naming) decision point and compare the alternatives with the designs that you've implemented (or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
<!-- If there was a discussion of the design of the feature implemented in this PR elsewhere (e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list), link to that discussion from this PR description and explain what have changed in your final design compared to your original proposal or the consensus version in the end of the discussion. If something hasn't changed since the original discussion, you can omit a detailed discussion of those aspects of the design here, perhaps apart from brief mentioning for the sake of readability of this PR description. -->
<!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
#### Release note
<!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users.
If your change doesn't have end user impact, you can skip this section.
-->
For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
<hr>
##### Key changed/added classes in this PR
* `MyFoo`
* `OurBar`
* `TheirBaz`
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033598551
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -551,6 +563,15 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
id(),
task.getDataSource(),
context,
+ (failedTask, fault) -> {
+ addToKernelManipulationQueue((kernel) -> {
+ if (isDurableStorageEnabled) {
Review Comment:
> IMO, it should default to false at first
Sorry what is it here , `isDurableStorageEnabled` or a new context param `faultTolerant`which is by default false?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033558947
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/MSQFaultUtils.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.druid.msq.indexing.error;
+
+public class MSQFaultUtils
+{
+
+ public static final String ERROR_CODE_DELIMITER = ": ";
+
+ /**
+ * Generate string message with error code delimited by {@link MSQFaultUtils#ERROR_CODE_DELIMITER}
+ */
+ public static String generateMessageWithErrorCode(MSQFault msqFault)
+ {
+ final String message = msqFault.getErrorMessage();
+
+ if (message != null && !message.isEmpty()) {
+ return msqFault.getErrorCode() + ERROR_CODE_DELIMITER + message;
+ } else {
+ return msqFault.getErrorCode();
+ }
+ }
+
+ /**
+ * Gets the error code from the message. If the messay is empty or null, {@link UnknownFault#CODE} is returned. This method
+ * does not gurantee that the error code we get out of the message is a valid error code.
+ */
+ public static String getErrorCodeFromMessage(String message)
Review Comment:
Umm, I do not think so as task status might be received from the tasks table where the status calls are chomped.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033557499
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -546,4 +614,65 @@ private static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefi
return retVal;
}
+
+ /**
+ * Checks the {@link MSQFault#getErrorCode()} is eligible for retry.
+ * <br/>
+ * If yes, transitions the stage to{@link ControllerStagePhase#RETRYING} and returns all the {@link WorkOrder}
+ * <br/>
+ * else throw {@link MSQException}
+ *
+ * @param workerNumber
+ * @param msqFault
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ public List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int workerNumber, MSQFault msqFault)
Review Comment:
Used this now: getWorkInCaseWorkerEligibleForRetryElseThrow. I want the caller to be aware that based on the fault the call may throw an exception.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1034291614
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -165,10 +188,88 @@ ClusterByPartitions getResultPartitionBoundaries()
}
}
+
+ /**
+ * Get workers which need to be sent partition boundaries
+ *
+ * @return
+ */
+ IntSet getWorkersToSendParitionBoundaries()
+ {
+ if (!getStageDefinition().doesShuffle()) {
+ throw new ISE("Result partition information is not relevant to this stage because it does not shuffle");
+ }
+ IntAVLTreeSet workers = new IntAVLTreeSet();
+ for (Integer worker : workerToPhase.keySet()) {
+ if (WorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES.equals(workerToPhase.get(worker))) {
+ workers.add(worker);
+ }
+ }
+ return workers;
+ }
+
+ /**
+ * Indicates that the work order for worker has been sent. Transistions the state to {@link WorkerStagePhase#READING_INPUT}
+ * if no more work orders need to be sent.
+ *
+ * @param worker
+ */
+ void workOrderSentForWorker(int worker)
+ {
+
+ workerToPhase.compute(worker, (wk, state) -> {
+ if (state == null) {
+ throw new ISE("Worker[%d] not found for stage[%s]", wk, stageDef.getStageNumber());
+ }
+ if (!WorkerStagePhase.READING_INPUT.canTransitionFrom(state)) {
+ throw new ISE(
+ "Worker[%d] cannot transistion from state[%s] to state[%s] while sending work order",
+ worker,
+ state,
+ WorkerStagePhase.READING_INPUT
+ );
+ }
+ return WorkerStagePhase.READING_INPUT;
+ });
+ if (phase != ControllerStagePhase.READING_INPUT) {
+ if (workOrdersNeedToBeSent()) {
Review Comment:
Updated the method name and the java doc to `allWorkOrdersSent`
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] lgtm-com[bot] commented on pull request #13353: [DRAFT] Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on PR #13353:
URL: https://github.com/apache/druid/pull/13353#issuecomment-1312799359
This pull request **introduces 1 alert** when merging c92b21749c18b665b8d85448645662413264ac5a into 81d005f267484ce883fa3d9beb76e75995412f94 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-b970839690a894895e844481d4513f65e2d574f9)
**new alerts:**
* 1 for Container contents are never accessed
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033634271
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -564,6 +585,34 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
return queryDef;
}
+ /**
+ * Adds the workorders for worker to {@link ControllerImpl#workOrdersToRetry} if the {@link ControllerQueryKernel} determines that there
+ * are work orders which needs reprocessing.
+ */
+ private void addToRetryQueue(ControllerQueryKernel kernel, int worker, MSQFault fault)
+ {
+ List<WorkOrder> retriableWorkOrders = kernel.getWorkInCaseWorkerElgibileForRetryElseThrow(worker, fault);
+ if (retriableWorkOrders.size() != 0) {
+ log.info("Submitting worker[%s] for relaunch because of fault[%s]", worker, fault);
Review Comment:
That information is inside the MSQWorkerTaskLauncher.
We donot have it here. What we do have is worker which is part of the log already.
What I have done is added the relaunch counter to the log message inside the MSQWorkerTaskLauncher#relaunchTasks
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -564,6 +585,34 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
return queryDef;
}
+ /**
+ * Adds the workorders for worker to {@link ControllerImpl#workOrdersToRetry} if the {@link ControllerQueryKernel} determines that there
+ * are work orders which needs reprocessing.
+ */
+ private void addToRetryQueue(ControllerQueryKernel kernel, int worker, MSQFault fault)
+ {
+ List<WorkOrder> retriableWorkOrders = kernel.getWorkInCaseWorkerElgibileForRetryElseThrow(worker, fault);
+ if (retriableWorkOrders.size() != 0) {
+ log.info("Submitting worker[%s] for relaunch because of fault[%s]", worker, fault);
Review Comment:
That information is inside the MSQWorkerTaskLauncher.
We do not have it here. What we do have is worker which is part of the log already.
What I have done is added the relaunch counter to the log message inside the MSQWorkerTaskLauncher#relaunchTasks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] adarshsanjeev commented on pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
adarshsanjeev commented on PR #13353:
URL: https://github.com/apache/druid/pull/13353#issuecomment-1342824473
The functions imported after resolving the conflict with sequential merge LGTM and should function in the same way as before. Will finish reviewing the rest as well.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1024799188
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -107,6 +117,19 @@
*/
private final Set<StageId> effectivelyFinishedStages = new HashSet<>();
+
+ /**
+ * Store the work orders for the stage so that we can retrieve that in case of worker retry
+ */
+ private final Map<StageId, Int2ObjectMap<WorkOrder>> stageWorkOrders;
+
+ /**
+ * {@link MSQFault#getErrorCode()} which are retried.
+ */
+ private final Set<String> retriableErrorCodes = ImmutableSet.of(CanceledFault.CODE, UnknownFault.CODE,
Review Comment:
I think we should define it in a separate constants file (a new one might do), in case you feel this will be required in other places as well.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java:
##########
@@ -398,33 +458,125 @@ private void updateTaskTrackersAndTaskIds()
/**
* Used by the main loop to generate exceptions if any tasks have failed, have taken too long to start up, or
* have gone inexplicably missing.
- *
+ * <p>
* Throws an exception if some task is erroneous.
*/
private void checkForErroneousTasks()
{
final int numTasks = taskTrackers.size();
- for (final Map.Entry<String, TaskTracker> taskEntry : taskTrackers.entrySet()) {
+ Iterator<Map.Entry<String, TaskTracker>> taskTrackerIterator = taskTrackers.entrySet().iterator();
+
+ while (taskTrackerIterator.hasNext()) {
+ final Map.Entry<String, TaskTracker> taskEntry = taskTrackerIterator.next();
final String taskId = taskEntry.getKey();
final TaskTracker tracker = taskEntry.getValue();
-
- if (tracker.status == null) {
- throw new MSQException(UnknownFault.forMessage(StringUtils.format("Task [%s] status missing", taskId)));
+ if (tracker.isRetrying()) {
+ continue;
}
- if (tracker.didRunTimeOut(maxTaskStartDelayMillis) && !canceledWorkerTasks.contains(taskId)) {
+ if (tracker.status == null) {
+ removeWorkerFromFullyStartedWorkers(tracker);
+ final String errorMessage = StringUtils.format("Task [%s] status missing", taskId);
+ log.info(errorMessage + ". Trying to relaunch the worker");
+ tracker.enableRetrying();
+ retryTask.retry(
+ tracker.msqWorkerTask,
+ UnknownFault.forMessage(errorMessage)
+ );
+
+ } else if (tracker.didRunTimeOut(maxTaskStartDelayMillis) && !canceledWorkerTasks.contains(taskId)) {
+ removeWorkerFromFullyStartedWorkers(tracker);
throw new MSQException(new TaskStartTimeoutFault(numTasks + 1));
+ } else if (tracker.didFail() && !canceledWorkerTasks.contains(taskId)) {
+ removeWorkerFromFullyStartedWorkers(tracker);
+ log.info("Task[%s] failed because %s. Trying to relaunch the worker", taskId, tracker.status.getErrorMsg());
+ tracker.enableRetrying();
+ retryTask.retry(tracker.msqWorkerTask, new WorkerFailedFault(taskId, tracker.status.getErrorMsg()));
}
+ }
+ }
- if (tracker.didFail() && !canceledWorkerTasks.contains(taskId)) {
- throw new MSQException(new WorkerFailedFault(taskId, tracker.status.getErrorMsg()));
- }
+ private void removeWorkerFromFullyStartedWorkers(TaskTracker tracker)
+ {
+ synchronized (taskIds) {
+ fullyStartedTasks.remove(tracker.msqWorkerTask.getWorkerNumber());
+ }
+ }
+
+
+ private void relaunchTasks()
+ {
+ Iterator<Integer> iterator = workersToRelaunch.iterator();
+
+ while (iterator.hasNext()) {
+ int worker = iterator.next();
+ workerToTaskIds.compute(worker, (workerId, taskHistory) -> {
+
+ if (taskHistory == null || taskHistory.isEmpty()) {
+ throw new ISE("TaskHistory cannot by null for worker %d", workerId);
+ }
+ String latestTaskId = taskHistory.get(taskHistory.size() - 1);
+
+ TaskTracker tracker = taskTrackers.get(latestTaskId);
+ if (tracker == null) {
+ throw new ISE("Did not find taskTracker for latest taskId[%s]", latestTaskId);
+ }
+ // if task is not failed donot retry
+ if (!tracker.isComplete()) {
+ return taskHistory;
+ }
+
+ MSQWorkerTask toRelaunch = tracker.msqWorkerTask;
+ MSQWorkerTask relaunchedTask = toRelaunch.getRetryTask();
+
+ // check relaunch limits
+ checkRelaunchLimits(tracker, toRelaunch);
+ // clean up trackers and tasks
+ tasksToCleanup.add(latestTaskId);
+ taskTrackers.remove(latestTaskId);
+ currentRelaunchCount += 1;
+ taskTrackers.put(relaunchedTask.getId(), new TaskTracker(relaunchedTask.getWorkerNumber(), relaunchedTask));
+ context.workerManager().run(relaunchedTask.getId(), relaunchedTask);
+ taskHistory.add(relaunchedTask.getId());
+
+ synchronized (taskIds) {
+ // replace taskId with the retry taskID for the same worker number
+ taskIds.set(toRelaunch.getWorkerNumber(), relaunchedTask.getId());
+ fullyStartedTasks.remove(relaunchedTask.getWorkerNumber());
+ taskIds.notifyAll();
+ }
+ return taskHistory;
+
+ });
+ iterator.remove();
+ }
+ }
+
+ private void checkRelaunchLimits(TaskTracker tracker, MSQWorkerTask relaunchTask)
Review Comment:
The method name (or signature) should mention that this can throw.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033598551
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -551,6 +563,15 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
id(),
task.getDataSource(),
context,
+ (failedTask, fault) -> {
+ addToKernelManipulationQueue((kernel) -> {
+ if (isDurableStorageEnabled) {
Review Comment:
> IMO, it should default to false at first
Sorry what is it here , `isDurableStorageEnabled` or a new context param `faultTolerant`which is by default false?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064643237
##########
extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java:
##########
@@ -223,6 +224,14 @@ public class MSQTestBase extends BaseCalciteQueryTest
.put(GroupByQueryConfig.CTX_KEY_ENABLE_MULTI_VALUE_UNNESTING, false)
.build();
+ public static final Map<String, Object> MSQ_CONTEXT_WITHOUT_DURABLE_STORAGE =
+ ImmutableMap.<String, Object>builder()
Review Comment:
Removed this. Added parametrized tests which should increase are testing coverage.
##########
extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java:
##########
@@ -223,6 +224,14 @@ public class MSQTestBase extends BaseCalciteQueryTest
.put(GroupByQueryConfig.CTX_KEY_ENABLE_MULTI_VALUE_UNNESTING, false)
.build();
+ public static final Map<String, Object> MSQ_CONTEXT_WITHOUT_DURABLE_STORAGE =
+ ImmutableMap.<String, Object>builder()
Review Comment:
Removed this. Added parametrized tests which should increase out testing coverage.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] adarshsanjeev commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
adarshsanjeev commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1052101272
##########
extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java:
##########
@@ -223,6 +224,14 @@ public class MSQTestBase extends BaseCalciteQueryTest
.put(GroupByQueryConfig.CTX_KEY_ENABLE_MULTI_VALUE_UNNESTING, false)
.build();
+ public static final Map<String, Object> MSQ_CONTEXT_WITHOUT_DURABLE_STORAGE =
+ ImmutableMap.<String, Object>builder()
Review Comment:
small nit: We cound construct this from the default context to reflect any future changes
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
gianm commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1027845939
##########
docs/multi-stage-query/known-issues.md:
##########
@@ -29,7 +29,7 @@ sidebar_label: Known issues
## Multi-stage query task runtime
-- Fault tolerance is not implemented. If any task fails, the entire query fails.
+- Fault tolerance is partially implemented. Workers get relaunched when they are killed unexpectedly. The controller does not get relaunched if its killed unexpectedly.
Review Comment:
"it is killed unexpectedly" (spelling)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -564,6 +585,34 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
return queryDef;
}
+ /**
+ * Adds the workorders for worker to {@link ControllerImpl#workOrdersToRetry} if the {@link ControllerQueryKernel} determines that there
+ * are work orders which needs reprocessing.
+ */
+ private void addToRetryQueue(ControllerQueryKernel kernel, int worker, MSQFault fault)
+ {
+ List<WorkOrder> retriableWorkOrders = kernel.getWorkInCaseWorkerElgibileForRetryElseThrow(worker, fault);
Review Comment:
Eligible (spelling)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -44,18 +52,20 @@
import org.apache.druid.msq.statistics.ClusterByStatisticsSnapshot;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
Review Comment:
ControllerQueryKernel is supposed to be a simple class that doesn't need to be thread-safe, because it's only accessed from the main logic thread of the controller. It's suspicious that ConcurrentHashMap is introduced here. What's it needed for? Can we achieve the same thing while keeping the simple/non-threadsafe nature of this class?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -986,19 +1036,79 @@ private Int2ObjectMap<List<SegmentIdWithShardSpec>> makeSegmentGeneratorWorkerFa
return retVal;
}
- private void contactWorkersForStage(final TaskContactFn contactFn, final IntSet workers)
+ /**
+ * A blocking function used to contact multiple workers. Checks if all the workers are running before contacting them.
+ *
+ * @param queryKernel
+ * @param contactFn
+ * @param workers set of workers to contact
+ * @param successCallBack on successfull api call, custom callback
+ * @param retryOnFailure if set to true, adds this worker to retry queue. If false, cancel all the futures and propergate the exception to the caller.
+ */
+ private void contactWorkersForStage(
+ final ControllerQueryKernel queryKernel,
+ final TaskContactFn contactFn,
+ final IntSet workers,
+ final TaskContactSuccesss successCallBack,
+ final boolean retryOnFailure
+ )
{
final List<String> taskIds = getTaskIds();
- final List<ListenableFuture<Void>> taskFutures = new ArrayList<>(workers.size());
+ final List<ListenableFuture<Boolean>> taskFutures = new ArrayList<>(workers.size());
+
+ try {
+ workerTaskLauncher.waitUntilWorkersReady(workers);
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ Set<Pair<Integer, String>> failedCalls = ConcurrentHashMap.newKeySet();
for (int workerNumber : workers) {
final String taskId = taskIds.get(workerNumber);
- taskFutures.add(contactFn.contactTask(netClient, taskId, workerNumber));
+ SettableFuture<Boolean> settableFuture = SettableFuture.create();
+ ListenableFuture<Void> apiFuture = contactFn.contactTask(netClient, taskId, workerNumber);
+ Futures.addCallback(apiFuture, new FutureCallback<Void>()
+ {
+ @Override
+ public void onSuccess(@Nullable Void result)
+ {
+ successCallBack.onSuccess(taskId, workerNumber);
+ settableFuture.set(true);
+ }
+
+ @Override
+ public void onFailure(Throwable t)
+ {
+ if (retryOnFailure) {
+ log.info(
+ t,
+ "Detected failure while contacting task[%s]. Iniitiating relaunch of worker[%d] if applicable",
Review Comment:
Initiating (spelling)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -392,6 +543,36 @@ private void generateResultPartitionsAndBoundaries()
}
}
+ /**
+ * True if all partitions stats are present, else false.
+ */
+ private boolean allPartitionStatisticsPresent()
+ {
+ return workerToPhase.values()
+ .stream()
+ .filter(stagePhase -> stagePhase.equals(WorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES)
+ || stagePhase.equals(WorkerStagePhase.PRESHUFFLE_WRITING_OUTPUT)
+ || stagePhase.equals(WorkerStagePhase.RESULTS_READY))
+ .count()
+ == workerCount;
+ }
+
+ /**
+ * True if work orders needs to be sent else false.
Review Comment:
This javadoc doesn't seem quite right either: it doesn't seem true to me that if all workers are in `RESULTS_READY` then work orders need to be sent. Please expand the javadoc to explain what this really means.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -392,6 +543,36 @@ private void generateResultPartitionsAndBoundaries()
}
}
+ /**
+ * True if all partitions stats are present, else false.
+ */
+ private boolean allPartitionStatisticsPresent()
Review Comment:
What do you think about making the `filter` from this method, and from `workOrdersNeedToBeSent`, into methods on `ControllerStagePhase` itself? Like, `stagePhase.workOrderNeedsToBeSent()`. It would keep all the phase-related code together in `ControllerStagePhase.java`, which may improve readability.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -38,27 +41,35 @@
import org.apache.druid.msq.input.stage.StageInputSlice;
import org.apache.druid.msq.kernel.StageDefinition;
import org.apache.druid.msq.kernel.WorkerAssignmentStrategy;
+import org.apache.druid.msq.kernel.worker.WorkerStagePhase;
import org.apache.druid.msq.statistics.ClusterByStatisticsCollector;
import org.apache.druid.msq.statistics.ClusterByStatisticsSnapshot;
import javax.annotation.Nullable;
import java.util.List;
+import java.util.stream.IntStream;
/**
* Controller-side state machine for each stage. Used by {@link ControllerQueryKernel} to form the overall state
* machine for an entire query.
- *
+ * <p>
* Package-private: stage trackers are an internal implementation detail of {@link ControllerQueryKernel}, not meant
* for separate use.
*/
class ControllerStageTracker
{
+ private static final Logger log = new Logger(ControllerStageTracker.class);
private final StageDefinition stageDef;
+
private final int workerCount;
private final WorkerInputs workerInputs;
+
+ // worker-> workerStagePhase
+ private final Int2ObjectMap<WorkerStagePhase> workerToPhase = new Int2ObjectOpenHashMap<>();
Review Comment:
This is a big change. Previously `WorkerStagePhase` was localized to the worker. Now it's on the controller too. Is the controller state kept in sync with the worker state? If so how does that work? Please explain this stuff in the comment.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -258,18 +355,41 @@ ControllerStagePhase addResultKeyStatisticsForWorker(
throw new IAE("Invalid workerNumber [%s]", workerNumber);
}
+ if (phase != ControllerStagePhase.READING_INPUT && phase != ControllerStagePhase.RETRYING) {
+ throw new ISE("Cannot add result key statistics from stage [%s]", phase);
+ }
+
+ WorkerStagePhase currentPhase = workerToPhase.get(workerNumber);
+
+ if (currentPhase == null) {
+ throw new ISE("Worker[%d] not found for stage[%s]", workerNumber, stageDef.getStageNumber());
+ }
+
try {
- if (workersWithResultKeyStatistics.add(workerNumber)) {
- resultKeyStatisticsCollector.addAll(snapshot);
+ if (WorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES.canTransitionFrom(currentPhase)) {
+ workerToPhase.put(workerNumber, WorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES);
- if (workersWithResultKeyStatistics.size() == workerCount) {
- generateResultPartitionsAndBoundaries();
+ // if stats already recieved for worker, donot update the sketch.
Review Comment:
received (spelling), do not (spelling)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -261,6 +269,9 @@ public class ControllerImpl implements Controller
// Time at which the query started.
// For live reports. Written by the main controller thread, read by HTTP threads.
+
+ // WorkerNumber -> WorkOrders
Review Comment:
Please expand on this comment. It should answer the following things:
- What is this map used for? What updates it and what reads it? Why does it need to be thread-safe?
- Is the `Set<WorkOrder>` a concurrent set? Does it need to be synchronized in some way when you use it?
##########
docs/multi-stage-query/reference.md:
##########
@@ -234,6 +234,9 @@ The following table lists query limits:
| Number of output columns for any one stage. | 2,000 | `TooManyColumns` |
| Number of workers for any one stage. | Hard limit is 1,000. Memory-dependent soft limit may be lower. | `TooManyWorkers` |
| Maximum memory occupied by broadcasted tables. | 30% of each [processor memory bundle](concepts.md#memory-usage). | `BroadcastTablesTooLarge` |
+| Maximum relaunches per worker. Initial run is not a relaunch. The worker will be spawned 1 + workerRelaunchLimit times before erroring out. | 2 | `WorkerRelaunchedTooManyTimes` |
+| Maximum relaunches across all workers. | 30 | `TotalRelaunchLimitExceededFault` |
Review Comment:
It'd be better if the two faults had similar names, like `TooManyAttemptsForWorker` and `TooManyAttemptsForJob`. Or something like that. Also, don't include `Fault` in the name of faults.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -107,6 +117,19 @@
*/
private final Set<StageId> effectivelyFinishedStages = new HashSet<>();
+
+ /**
+ * Store the work orders for the stage so that we can retrieve that in case of worker retry
+ */
+ private final Map<StageId, Int2ObjectMap<WorkOrder>> stageWorkOrders;
+
+ /**
+ * {@link MSQFault#getErrorCode()} which are retried.
+ */
+ private final Set<String> retriableErrorCodes = ImmutableSet.of(CanceledFault.CODE, UnknownFault.CODE,
Review Comment:
`static`?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/MSQFaultUtils.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.druid.msq.indexing.error;
+
+public class MSQFaultUtils
+{
+
+ public static final String ERROR_CODE_DELIMITER = ": ";
+
+ /**
+ * Generate string message with error code delimited by {@link MSQFaultUtils#ERROR_CODE_DELIMITER}
+ */
+ public static String generateMessageWithErrorCode(MSQFault msqFault)
+ {
+ final String message = msqFault.getErrorMessage();
+
+ if (message != null && !message.isEmpty()) {
+ return msqFault.getErrorCode() + ERROR_CODE_DELIMITER + message;
+ } else {
+ return msqFault.getErrorCode();
+ }
+ }
+
+ /**
+ * Gets the error code from the message. If the messay is empty or null, {@link UnknownFault#CODE} is returned. This method
Review Comment:
It looks like the purpose of this method is to extract the original error code from a `WorkerFailedFault`. This is the what structured error details are for. Please consider removing this method, and instead adding a `workerError` (type MSQFault) or `workerErrorCode` (type String) field to `WorkerFailedFault`.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStagePhase.java:
##########
@@ -85,6 +85,15 @@ public boolean canTransitionFrom(final ControllerStagePhase priorPhase)
{
return true;
}
+ },
+
+ // Stage currently under retry. priorPhase did not publish its final results yet.
Review Comment:
What does "under retry" mean? What is the expected next state after this one? Please elaborate in the comment.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -603,6 +652,7 @@ public void updateStatus(int stageNumber, int workerNumber, Object keyStatistics
@Override
public void workerError(MSQErrorReport errorReport)
{
+ // move inside kernel
Review Comment:
Is this a todo that you mean to do prior to committing?
##########
docs/multi-stage-query/reference.md:
##########
@@ -234,6 +234,9 @@ The following table lists query limits:
| Number of output columns for any one stage. | 2,000 | `TooManyColumns` |
| Number of workers for any one stage. | Hard limit is 1,000. Memory-dependent soft limit may be lower. | `TooManyWorkers` |
| Maximum memory occupied by broadcasted tables. | 30% of each [processor memory bundle](concepts.md#memory-usage). | `BroadcastTablesTooLarge` |
+| Maximum relaunches per worker. Initial run is not a relaunch. The worker will be spawned 1 + workerRelaunchLimit times before erroring out. | 2 | `WorkerRelaunchedTooManyTimes` |
+| Maximum relaunches across all workers. | 30 | `TotalRelaunchLimitExceededFault` |
Review Comment:
Seems low, especially since a single machine can be running many workers that would all fail at the same time.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -107,6 +117,19 @@
*/
private final Set<StageId> effectivelyFinishedStages = new HashSet<>();
+
+ /**
+ * Store the work orders for the stage so that we can retrieve that in case of worker retry
Review Comment:
Comment should include what the int keys of the inner map are. I guess it's worker number, but that should be explicit in the comment.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -352,22 +401,25 @@ public Object getResultObjectForStage(final StageId stageId)
/**
* Checks if the stage can be started, delegates call to {@link ControllerStageTracker#start()} for internal phase
- * transition and registers the transition in this queryKernel
+ * transition and registers the transition in this queryKernel. Work orders need to created via {@link ControllerQueryKernel#createWorkOrders(int, Int2ObjectMap)} before calling this method.
Review Comment:
"need to be created" (grammar). Also this line is long.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -392,6 +543,36 @@ private void generateResultPartitionsAndBoundaries()
}
}
+ /**
+ * True if all partitions stats are present, else false.
Review Comment:
This javadoc doesn't seem quite right: it's going to possibly return `true` even if `!stageDef.mustGatherResultKeyStatistics()`, but in that case of course there are no partition statistics. This behavior is fine, but the javadoc should be clearer. Please expand the javadoc to explain what this really means.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -484,6 +551,7 @@ public void transitionStageKernel(StageId stageId, ControllerStagePhase newPhase
}
}
+ // might need to change this
Review Comment:
Why?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -340,20 +490,21 @@ void fail()
/**
* Sets {@link #resultPartitions} (always) and {@link #resultPartitionBoundaries}.
- *
+ * <p>
* If {@link StageDefinition#mustGatherResultKeyStatistics()} is true, this method cannot be called until after
* statistics have been provided to {@link #addResultKeyStatisticsForWorker} for all workers.
*/
private void generateResultPartitionsAndBoundaries()
{
if (resultPartitions != null) {
- throw new ISE("Result partitions have already been generated");
+ log.debug("Partition boundaries already generated for stage %d", stageDef.getStageNumber());
Review Comment:
Please include a comment about why it makes sense to return quietly if `resultPartitions` is already set.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -546,4 +614,65 @@ private static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefi
return retVal;
}
+
+ /**
+ * Checks the {@link MSQFault#getErrorCode()} is eligible for retry.
+ * <br/>
+ * If yes, transitions the stage to{@link ControllerStagePhase#RETRYING} and returns all the {@link WorkOrder}
+ * <br/>
+ * else throw {@link MSQException}
+ *
+ * @param workerNumber
+ * @param msqFault
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ public List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int workerNumber, MSQFault msqFault)
+ {
+
+ final String errorCode;
+ if (msqFault instanceof WorkerFailedFault) {
+ errorCode = MSQFaultUtils.getErrorCodeFromMessage((((WorkerFailedFault) msqFault).getErrorMsg()));
+ } else {
+ errorCode = msqFault.getErrorCode();
+ }
+
+ if (retriableErrorCodes.contains(errorCode)) {
+ return getWorkInCaseWorkerElgibileForRetryElseThrow(workerNumber);
+
+ } else {
+ throw new MSQException(msqFault);
+ }
+ }
+
+ /**
+ * Gets all the stages currently being tracked and filters out all effectively finished stages.
+ * <br/>
+ * From the remaining stages, checks if (stage,worker) needs to be retried.
+ * <br/>
+ * If yes adds the workOrder for that stage to the return list and transitions the stage kernel to {@link ControllerStagePhase#RETRYING}
+ *
+ * @param worker
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ private List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int worker)
+ {
+ List<StageId> trackedSet = new ArrayList<>(getActiveStages());
+ // no need to retry effectively finished stages
+ List<StageId> getEffictivelyFinishedStages = getEffectivelyFinishedStageIds();
+ trackedSet.removeAll(getEffictivelyFinishedStages);
+
+ List<WorkOrder> workOrders = new ArrayList<>();
+
+ for (StageId stageId : trackedSet) {
+ ControllerStageTracker controllerStageTracker = getStageKernelOrThrow(stageId);
+ if (ControllerStagePhase.RETRYING.canTransitionFrom(controllerStageTracker.getPhase())
+ && controllerStageTracker.retryIfNeeded(worker)) {
+ workOrders.add(getWorkOrder(worker, stageId));
+ // should be a no-op. Calling for code patterns.
Review Comment:
What does "calling for code patterns" mean?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -564,6 +585,34 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
return queryDef;
}
+ /**
+ * Adds the workorders for worker to {@link ControllerImpl#workOrdersToRetry} if the {@link ControllerQueryKernel} determines that there
+ * are work orders which needs reprocessing.
+ */
+ private void addToRetryQueue(ControllerQueryKernel kernel, int worker, MSQFault fault)
+ {
+ List<WorkOrder> retriableWorkOrders = kernel.getWorkInCaseWorkerElgibileForRetryElseThrow(worker, fault);
+ if (retriableWorkOrders.size() != 0) {
+ log.info("Submitting worker[%s] for relaunch because of fault[%s]", worker, fault);
Review Comment:
Include the worker and job relaunch counter in this log message?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -551,6 +563,15 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
id(),
task.getDataSource(),
context,
+ (failedTask, fault) -> {
+ addToKernelManipulationQueue((kernel) -> {
+ if (isDurableStorageEnabled) {
Review Comment:
Agreed — it should all be controllable by the same `faultTolerant` setting. We don't want people to have to tweak a bunch of different settings in order to get the full package of behavior.
IMO, it should default to false at first, while we're experimenting with it, but we should change the default to true once we're feeling that it's stable.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -406,7 +459,6 @@ public void addResultKeyStatisticsForStageAndWorker(
// If the phase is POST_READING or FAILED, that implies the kernel has transitioned. We need to account for that
switch (newPhase) {
- case POST_READING:
Review Comment:
Comment above is no longer accurate, it seems.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -165,10 +188,88 @@ ClusterByPartitions getResultPartitionBoundaries()
}
}
+
+ /**
+ * Get workers which need to be sent partition boundaries
+ *
+ * @return
+ */
+ IntSet getWorkersToSendParitionBoundaries()
+ {
+ if (!getStageDefinition().doesShuffle()) {
+ throw new ISE("Result partition information is not relevant to this stage because it does not shuffle");
+ }
+ IntAVLTreeSet workers = new IntAVLTreeSet();
+ for (Integer worker : workerToPhase.keySet()) {
+ if (WorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES.equals(workerToPhase.get(worker))) {
+ workers.add(worker);
+ }
+ }
+ return workers;
+ }
+
+ /**
+ * Indicates that the work order for worker has been sent. Transistions the state to {@link WorkerStagePhase#READING_INPUT}
+ * if no more work orders need to be sent.
+ *
+ * @param worker
+ */
+ void workOrderSentForWorker(int worker)
+ {
+
+ workerToPhase.compute(worker, (wk, state) -> {
+ if (state == null) {
+ throw new ISE("Worker[%d] not found for stage[%s]", wk, stageDef.getStageNumber());
+ }
+ if (!WorkerStagePhase.READING_INPUT.canTransitionFrom(state)) {
+ throw new ISE(
+ "Worker[%d] cannot transistion from state[%s] to state[%s] while sending work order",
+ worker,
+ state,
+ WorkerStagePhase.READING_INPUT
+ );
+ }
+ return WorkerStagePhase.READING_INPUT;
+ });
+ if (phase != ControllerStagePhase.READING_INPUT) {
+ if (workOrdersNeedToBeSent()) {
Review Comment:
Something seems backwards here? The check is `workOrdersNeedToBeSent()`, but the comment says "if no more work orders need to be sent…". Please clarify.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStagePhase.java:
##########
@@ -35,7 +35,7 @@
@Override
public boolean canTransitionFrom(final ControllerStagePhase priorPhase)
{
- return false;
+ return true;
Review Comment:
Seems strange to me too; under which cases can a stage transition back to NEW?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064634362
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -38,27 +41,35 @@
import org.apache.druid.msq.input.stage.StageInputSlice;
import org.apache.druid.msq.kernel.StageDefinition;
import org.apache.druid.msq.kernel.WorkerAssignmentStrategy;
+import org.apache.druid.msq.kernel.worker.WorkerStagePhase;
import org.apache.druid.msq.statistics.ClusterByStatisticsCollector;
import org.apache.druid.msq.statistics.ClusterByStatisticsSnapshot;
import javax.annotation.Nullable;
import java.util.List;
+import java.util.stream.IntStream;
/**
* Controller-side state machine for each stage. Used by {@link ControllerQueryKernel} to form the overall state
* machine for an entire query.
- *
+ * <p>
* Package-private: stage trackers are an internal implementation detail of {@link ControllerQueryKernel}, not meant
* for separate use.
*/
class ControllerStageTracker
{
+ private static final Logger log = new Logger(ControllerStageTracker.class);
private final StageDefinition stageDef;
+
private final int workerCount;
private final WorkerInputs workerInputs;
+
+ // worker-> workerStagePhase
+ private final Int2ObjectMap<WorkerStagePhase> workerToPhase = new Int2ObjectOpenHashMap<>();
Review Comment:
// worker-> workerStagePhase
// Controller keeps track of the stage with this map.
// Currently, we rely on the serial nature of the state machine to keep things in sync between the controller and the worker.
// So the worker state in the controller can go out of sync with the actual worker state.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -38,27 +41,35 @@
import org.apache.druid.msq.input.stage.StageInputSlice;
import org.apache.druid.msq.kernel.StageDefinition;
import org.apache.druid.msq.kernel.WorkerAssignmentStrategy;
+import org.apache.druid.msq.kernel.worker.WorkerStagePhase;
import org.apache.druid.msq.statistics.ClusterByStatisticsCollector;
import org.apache.druid.msq.statistics.ClusterByStatisticsSnapshot;
import javax.annotation.Nullable;
import java.util.List;
+import java.util.stream.IntStream;
/**
* Controller-side state machine for each stage. Used by {@link ControllerQueryKernel} to form the overall state
* machine for an entire query.
- *
+ * <p>
* Package-private: stage trackers are an internal implementation detail of {@link ControllerQueryKernel}, not meant
* for separate use.
*/
class ControllerStageTracker
{
+ private static final Logger log = new Logger(ControllerStageTracker.class);
private final StageDefinition stageDef;
+
private final int workerCount;
private final WorkerInputs workerInputs;
+
+ // worker-> workerStagePhase
+ private final Int2ObjectMap<WorkerStagePhase> workerToPhase = new Int2ObjectOpenHashMap<>();
Review Comment:
Added the 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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064652075
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQueryHA.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.msq.util.MultiStageQueryContext;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.ITRetryUtil;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.utils.DruidClusterAdminClient;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQueryHA
+{
+ private static final Logger LOG = new Logger(ITMultiStageQueryHA.class);
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ private SqlResourceTestClient msqClient;
+
+ @Inject
+ private IntegrationTestingConfig config;
+
+ @Inject
+ private ObjectMapper jsonMapper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Inject
+ private CoordinatorResourceTestClient coordinatorClient;
+
+ @Inject
+ private DruidClusterAdminClient druidClusterAdminClient;
+
+ private static final String QUERY_FILE = "/multi-stage-query/wikipedia_msq_select_query_ha.json";
+
+ @Test
+ public void testMsqIngestionAndQuerying() throws Exception
+ {
+ String datasource = "dst";
+
+ // Clear up the datasource from the previous runs
+ coordinatorClient.unloadSegmentsForDataSource(datasource);
+
+ String queryLocal =
+ StringUtils.format(
+ "INSERT INTO %s\n"
+ + "SELECT\n"
+ + " TIME_PARSE(\"timestamp\") AS __time,\n"
+ + " isRobot,\n"
+ + " diffUrl,\n"
+ + " added,\n"
+ + " countryIsoCode,\n"
+ + " regionName,\n"
+ + " channel,\n"
+ + " flags,\n"
+ + " delta,\n"
+ + " isUnpatrolled,\n"
+ + " isNew,\n"
+ + " deltaBucket,\n"
+ + " isMinor,\n"
+ + " isAnonymous,\n"
+ + " deleted,\n"
+ + " cityName,\n"
+ + " metroCode,\n"
+ + " namespace,\n"
+ + " comment,\n"
+ + " page,\n"
+ + " commentLength,\n"
+ + " countryName,\n"
+ + " user,\n"
+ + " regionIsoCode\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + " '{\"type\":\"local\",\"files\":[\"/resources/data/batch_index/json/wikipedia_index_data1.json\",\"/resources/data/batch_index/json/wikipedia_index_data1.json\"]}',\n"
+ + " '{\"type\":\"json\"}',\n"
+ + " '[{\"type\":\"string\",\"name\":\"timestamp\"},{\"type\":\"string\",\"name\":\"isRobot\"},{\"type\":\"string\",\"name\":\"diffUrl\"},{\"type\":\"long\",\"name\":\"added\"},{\"type\":\"string\",\"name\":\"countryIsoCode\"},{\"type\":\"string\",\"name\":\"regionName\"},{\"type\":\"string\",\"name\":\"channel\"},{\"type\":\"string\",\"name\":\"flags\"},{\"type\":\"long\",\"name\":\"delta\"},{\"type\":\"string\",\"name\":\"isUnpatrolled\"},{\"type\":\"string\",\"name\":\"isNew\"},{\"type\":\"double\",\"name\":\"deltaBucket\"},{\"type\":\"string\",\"name\":\"isMinor\"},{\"type\":\"string\",\"name\":\"isAnonymous\"},{\"type\":\"long\",\"name\":\"deleted\"},{\"type\":\"string\",\"name\":\"cityName\"},{\"type\":\"long\",\"name\":\"metroCode\"},{\"type\":\"string\",\"name\":\"namespace\"},{\"type\":\"string\",\"name\":\"comment\"},{\"type\":\"string\",\"name\":\"page\"},{\"type\":\"long\",\"name\":\"commentLength\"},{\"type\":\"string\",\"name\":\"countryName\"},{\"type\":
\"string\",\"name\":\"user\"},{\"type\":\"string\",\"name\":\"regionIsoCode\"}]'\n"
+ + " )\n"
+ + ")\n"
+ + "PARTITIONED BY DAY\n"
+ + "CLUSTERED BY \"__time\"",
+ datasource
+ );
+
+ // Submit the task and wait for the datasource to get loaded
+ SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTask(
+ queryLocal,
+ ImmutableMap.of(
+ MultiStageQueryContext.CTX_DURABLE_SHUFFLE_STORAGE,
+ "true",
+ MultiStageQueryContext.CTX_MAX_NUM_TASKS,
+ 3
+ )
+ );
+
+ if (sqlTaskStatus.getState().isFailure()) {
+ Assert.fail(StringUtils.format(
+ "Unable to start the task successfully.\nPossible exception: %s",
+ sqlTaskStatus.getError()
+ ));
+ }
+
+
+ String taskIdToKill = sqlTaskStatus.getTaskId() + "-worker1_0";
+ killTaskAbruptly(taskIdToKill);
+
+ msqHelper.pollTaskIdForCompletion(sqlTaskStatus.getTaskId());
+ dataLoaderHelper.waitUntilDatasourceIsReady(datasource);
+
+ msqHelper.testQueriesFromFile(QUERY_FILE, datasource);
+ }
+
+ private void killTaskAbruptly(String taskIdToKill)
+ {
+
+ String command = "jps -mlv | grep -i peon | grep -i " + taskIdToKill + " |awk '{print $1}'";
+
+ ITRetryUtil.retryUntil(() -> {
+
+ Pair<String, String> stdOut = druidClusterAdminClient.runCommandInMiddleManagerContainer("/bin/bash", "-c",
Review Comment:
as tasks here are only for MM, I think we should be okay.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on PR #13353:
URL: https://github.com/apache/druid/pull/13353#issuecomment-1378152486
Thanks, @rohangarg @LakshSingla @adarshsanjeev for the 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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] lgtm-com[bot] commented on pull request #13353: [DRAFT] Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on PR #13353:
URL: https://github.com/apache/druid/pull/13353#issuecomment-1312824782
This pull request **introduces 1 alert** when merging 58da3cf8f7fcf2b12f379b169bc05a16ab790acb into a3edda3b634d00942f8729930dd45e2dd6b2eaff - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-7f5a284ddd3de40551a4e9df194b2bf591db8d50)
**new alerts:**
* 1 for Container contents are never accessed
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033634271
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -564,6 +585,34 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
return queryDef;
}
+ /**
+ * Adds the workorders for worker to {@link ControllerImpl#workOrdersToRetry} if the {@link ControllerQueryKernel} determines that there
+ * are work orders which needs reprocessing.
+ */
+ private void addToRetryQueue(ControllerQueryKernel kernel, int worker, MSQFault fault)
+ {
+ List<WorkOrder> retriableWorkOrders = kernel.getWorkInCaseWorkerElgibileForRetryElseThrow(worker, fault);
+ if (retriableWorkOrders.size() != 0) {
+ log.info("Submitting worker[%s] for relaunch because of fault[%s]", worker, fault);
Review Comment:
That information is inside the MSQWorkerTaskLauncher. We donot have it here. What we do have is worker which is part of the log already. What I have done is added the relaunch counter inside the MSQWorkerTaskLauncher#relaunchTasks
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1051751976
##########
core/src/main/java/org/apache/druid/java/util/common/function/TriConsumer.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.druid.java.util.common.function;
+
+import java.util.Objects;
+
+/**
+ * Based on {@link java.util.function.BiConsumer}
+ */
+@FunctionalInterface
+public interface TriConsumer<T, U, V>
+{
+ /**
+ * Performs this operation on the given arguments.
+ *
+ * @param t the first input argument
+ * @param u the second input argument
+ * @param v the third input argument
+ */
+ void accept(T t, U u, V v);
+
+ /**
+ * Returns a composed {@code BiConsumer} that performs, in sequence, this
Review Comment:
I think we are returning a TriConsumer from the method and this should be updated. Same for the definition of `@return`
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQueryHA.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.msq.util.MultiStageQueryContext;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.ITRetryUtil;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.utils.DruidClusterAdminClient;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQueryHA
Review Comment:
Can you please elaborate on what `HA` at the end is?
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQueryHA.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.msq.util.MultiStageQueryContext;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.ITRetryUtil;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.utils.DruidClusterAdminClient;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQueryHA
+{
+ private static final Logger LOG = new Logger(ITMultiStageQueryHA.class);
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ private SqlResourceTestClient msqClient;
+
+ @Inject
+ private IntegrationTestingConfig config;
+
+ @Inject
+ private ObjectMapper jsonMapper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Inject
+ private CoordinatorResourceTestClient coordinatorClient;
+
+ @Inject
+ private DruidClusterAdminClient druidClusterAdminClient;
+
+ private static final String QUERY_FILE = "/multi-stage-query/wikipedia_msq_select_query_ha.json";
+
+ @Test
+ public void testMsqIngestionAndQuerying() throws Exception
+ {
+ String datasource = "dst";
+
+ // Clear up the datasource from the previous runs
+ coordinatorClient.unloadSegmentsForDataSource(datasource);
+
+ String queryLocal =
+ StringUtils.format(
+ "INSERT INTO %s\n"
+ + "SELECT\n"
+ + " TIME_PARSE(\"timestamp\") AS __time,\n"
+ + " isRobot,\n"
+ + " diffUrl,\n"
+ + " added,\n"
+ + " countryIsoCode,\n"
+ + " regionName,\n"
+ + " channel,\n"
+ + " flags,\n"
+ + " delta,\n"
+ + " isUnpatrolled,\n"
+ + " isNew,\n"
+ + " deltaBucket,\n"
+ + " isMinor,\n"
+ + " isAnonymous,\n"
+ + " deleted,\n"
+ + " cityName,\n"
+ + " metroCode,\n"
+ + " namespace,\n"
+ + " comment,\n"
+ + " page,\n"
+ + " commentLength,\n"
+ + " countryName,\n"
+ + " user,\n"
+ + " regionIsoCode\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + " '{\"type\":\"local\",\"files\":[\"/resources/data/batch_index/json/wikipedia_index_data1.json\",\"/resources/data/batch_index/json/wikipedia_index_data1.json\"]}',\n"
+ + " '{\"type\":\"json\"}',\n"
+ + " '[{\"type\":\"string\",\"name\":\"timestamp\"},{\"type\":\"string\",\"name\":\"isRobot\"},{\"type\":\"string\",\"name\":\"diffUrl\"},{\"type\":\"long\",\"name\":\"added\"},{\"type\":\"string\",\"name\":\"countryIsoCode\"},{\"type\":\"string\",\"name\":\"regionName\"},{\"type\":\"string\",\"name\":\"channel\"},{\"type\":\"string\",\"name\":\"flags\"},{\"type\":\"long\",\"name\":\"delta\"},{\"type\":\"string\",\"name\":\"isUnpatrolled\"},{\"type\":\"string\",\"name\":\"isNew\"},{\"type\":\"double\",\"name\":\"deltaBucket\"},{\"type\":\"string\",\"name\":\"isMinor\"},{\"type\":\"string\",\"name\":\"isAnonymous\"},{\"type\":\"long\",\"name\":\"deleted\"},{\"type\":\"string\",\"name\":\"cityName\"},{\"type\":\"long\",\"name\":\"metroCode\"},{\"type\":\"string\",\"name\":\"namespace\"},{\"type\":\"string\",\"name\":\"comment\"},{\"type\":\"string\",\"name\":\"page\"},{\"type\":\"long\",\"name\":\"commentLength\"},{\"type\":\"string\",\"name\":\"countryName\"},{\"type\":
\"string\",\"name\":\"user\"},{\"type\":\"string\",\"name\":\"regionIsoCode\"}]'\n"
+ + " )\n"
+ + ")\n"
+ + "PARTITIONED BY DAY\n"
+ + "CLUSTERED BY \"__time\"",
+ datasource
+ );
+
+ // Submit the task and wait for the datasource to get loaded
+ SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTask(
+ queryLocal,
+ ImmutableMap.of(
+ MultiStageQueryContext.CTX_DURABLE_SHUFFLE_STORAGE,
+ "true",
+ MultiStageQueryContext.CTX_MAX_NUM_TASKS,
+ 3
+ )
+ );
+
+ if (sqlTaskStatus.getState().isFailure()) {
+ Assert.fail(StringUtils.format(
+ "Unable to start the task successfully.\nPossible exception: %s",
+ sqlTaskStatus.getError()
+ ));
+ }
+
+
+ String taskIdToKill = sqlTaskStatus.getTaskId() + "-worker1_0";
+ killTaskAbruptly(taskIdToKill);
+
+ msqHelper.pollTaskIdForCompletion(sqlTaskStatus.getTaskId());
+ dataLoaderHelper.waitUntilDatasourceIsReady(datasource);
+
+ msqHelper.testQueriesFromFile(QUERY_FILE, datasource);
+ }
+
+ private void killTaskAbruptly(String taskIdToKill)
Review Comment:
Should this be moved to the IT Framework's core methods instead of being in a specific IT test? Also, is there a better way to do it without invoking the command?
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQueryHA.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.msq.util.MultiStageQueryContext;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.ITRetryUtil;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.utils.DruidClusterAdminClient;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQueryHA
+{
+ private static final Logger LOG = new Logger(ITMultiStageQueryHA.class);
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ private SqlResourceTestClient msqClient;
+
+ @Inject
+ private IntegrationTestingConfig config;
+
+ @Inject
+ private ObjectMapper jsonMapper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Inject
+ private CoordinatorResourceTestClient coordinatorClient;
+
+ @Inject
+ private DruidClusterAdminClient druidClusterAdminClient;
+
+ private static final String QUERY_FILE = "/multi-stage-query/wikipedia_msq_select_query_ha.json";
+
+ @Test
+ public void testMsqIngestionAndQuerying() throws Exception
+ {
+ String datasource = "dst";
+
+ // Clear up the datasource from the previous runs
+ coordinatorClient.unloadSegmentsForDataSource(datasource);
+
+ String queryLocal =
+ StringUtils.format(
+ "INSERT INTO %s\n"
+ + "SELECT\n"
+ + " TIME_PARSE(\"timestamp\") AS __time,\n"
+ + " isRobot,\n"
+ + " diffUrl,\n"
+ + " added,\n"
+ + " countryIsoCode,\n"
+ + " regionName,\n"
+ + " channel,\n"
+ + " flags,\n"
+ + " delta,\n"
+ + " isUnpatrolled,\n"
+ + " isNew,\n"
+ + " deltaBucket,\n"
+ + " isMinor,\n"
+ + " isAnonymous,\n"
+ + " deleted,\n"
+ + " cityName,\n"
+ + " metroCode,\n"
+ + " namespace,\n"
+ + " comment,\n"
+ + " page,\n"
+ + " commentLength,\n"
+ + " countryName,\n"
+ + " user,\n"
+ + " regionIsoCode\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + " '{\"type\":\"local\",\"files\":[\"/resources/data/batch_index/json/wikipedia_index_data1.json\",\"/resources/data/batch_index/json/wikipedia_index_data1.json\"]}',\n"
+ + " '{\"type\":\"json\"}',\n"
+ + " '[{\"type\":\"string\",\"name\":\"timestamp\"},{\"type\":\"string\",\"name\":\"isRobot\"},{\"type\":\"string\",\"name\":\"diffUrl\"},{\"type\":\"long\",\"name\":\"added\"},{\"type\":\"string\",\"name\":\"countryIsoCode\"},{\"type\":\"string\",\"name\":\"regionName\"},{\"type\":\"string\",\"name\":\"channel\"},{\"type\":\"string\",\"name\":\"flags\"},{\"type\":\"long\",\"name\":\"delta\"},{\"type\":\"string\",\"name\":\"isUnpatrolled\"},{\"type\":\"string\",\"name\":\"isNew\"},{\"type\":\"double\",\"name\":\"deltaBucket\"},{\"type\":\"string\",\"name\":\"isMinor\"},{\"type\":\"string\",\"name\":\"isAnonymous\"},{\"type\":\"long\",\"name\":\"deleted\"},{\"type\":\"string\",\"name\":\"cityName\"},{\"type\":\"long\",\"name\":\"metroCode\"},{\"type\":\"string\",\"name\":\"namespace\"},{\"type\":\"string\",\"name\":\"comment\"},{\"type\":\"string\",\"name\":\"page\"},{\"type\":\"long\",\"name\":\"commentLength\"},{\"type\":\"string\",\"name\":\"countryName\"},{\"type\":
\"string\",\"name\":\"user\"},{\"type\":\"string\",\"name\":\"regionIsoCode\"}]'\n"
+ + " )\n"
+ + ")\n"
+ + "PARTITIONED BY DAY\n"
+ + "CLUSTERED BY \"__time\"",
+ datasource
+ );
+
+ // Submit the task and wait for the datasource to get loaded
+ SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTask(
+ queryLocal,
+ ImmutableMap.of(
+ MultiStageQueryContext.CTX_DURABLE_SHUFFLE_STORAGE,
+ "true",
+ MultiStageQueryContext.CTX_MAX_NUM_TASKS,
+ 3
+ )
+ );
+
+ if (sqlTaskStatus.getState().isFailure()) {
+ Assert.fail(StringUtils.format(
+ "Unable to start the task successfully.\nPossible exception: %s",
+ sqlTaskStatus.getError()
+ ));
+ }
+
+
+ String taskIdToKill = sqlTaskStatus.getTaskId() + "-worker1_0";
+ killTaskAbruptly(taskIdToKill);
+
+ msqHelper.pollTaskIdForCompletion(sqlTaskStatus.getTaskId());
+ dataLoaderHelper.waitUntilDatasourceIsReady(datasource);
+
+ msqHelper.testQueriesFromFile(QUERY_FILE, datasource);
+ }
+
+ private void killTaskAbruptly(String taskIdToKill)
+ {
+
+ String command = "jps -mlv | grep -i peon | grep -i " + taskIdToKill + " |awk '{print $1}'";
+
+ ITRetryUtil.retryUntil(() -> {
+
+ Pair<String, String> stdOut = druidClusterAdminClient.runCommandInMiddleManagerContainer("/bin/bash", "-c",
Review Comment:
I think we should rename the method's name here since we are running the command in a specific container however the method doesn't signify as such.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/Limits.java:
##########
@@ -68,4 +68,20 @@
* Maximum size of the kernel manipulation queue in {@link org.apache.druid.msq.indexing.MSQControllerTask}.
*/
public static final int MAX_KERNEL_MANIPULATION_QUEUE_SIZE = 100_000;
+
+ /**
+ * Maximum relaunches across all workers.
+ */
+ public static final int TOTAL_RELAUNCH_LIMIT = 100;
+
+ /**
+ * Maximum relaunches per worker. Initial run is not a relaunch. The worker will be spawned 1 + workerRelaunchLimit times before erroring out.
+ */
+ public static final int PER_WORKER_RELAUNCH_LIMIT = 2;
Review Comment:
Should this be increased? This seems a bit low, what has been your experience with the same?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MSQTasks.java:
##########
@@ -65,9 +71,26 @@ public static String controllerTaskId(@Nullable final String queryId)
/**
* Returns a worker task ID given a SQL query id.
*/
- public static String workerTaskId(final String controllerTaskId, final int workerNumber)
+ public static String workerTaskId(final String controllerTaskId, final int workerNumber, final int retryCount)
{
- return StringUtils.format("%s-worker%d", controllerTaskId, workerNumber);
+ return StringUtils.format("%s-worker%d_%d", controllerTaskId, workerNumber, retryCount);
+ }
+
+ /**
+ * Extract worker from taskId or throw exception if unable to parse out the worker.
+ */
+ public static int workerFromTaskId(final String taskId)
+ {
+ final Matcher matcher = WORKER_PATTERN.matcher(taskId);
+ if (matcher.matches()) {
+ return Integer.parseInt(matcher.group(1));
Review Comment:
If we can use the named group in the regex, then we won't have to use the stray reference to `1` and the code would remain valid irrespective of regex.
```suggestion
return Integer.parseInt(matcher.group("workerNumber"));
```
##########
docs/multi-stage-query/reference.md:
##########
@@ -234,6 +234,9 @@ The following table lists query limits:
| Number of output columns for any one stage. | 2,000 | `TooManyColumns` |
| Number of workers for any one stage. | Hard limit is 1,000. Memory-dependent soft limit may be lower. | `TooManyWorkers` |
| Maximum memory occupied by broadcasted tables. | 30% of each [processor memory bundle](concepts.md#memory-usage). | `BroadcastTablesTooLarge` |
+| Maximum relaunch attempts per worker. Initial run is not a relaunch. The worker will be spawned 1 + workerRelaunchLimit times before erroring out. | 2 | `TooManyAttemptsForWorker` |
+| Maximum relaunch attempts for a job across all workers. | 100 | `TooManyAttemptsForJob` |
Review Comment:
I think the limit here is outdated.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java:
##########
@@ -58,10 +58,14 @@
public static final String CTX_FINALIZE_AGGREGATIONS = "finalizeAggregations";
private static final boolean DEFAULT_FINALIZE_AGGREGATIONS = true;
- public static final String CTX_ENABLE_DURABLE_SHUFFLE_STORAGE = "durableShuffleStorage";
+ public static final String CTX_DURABLE_SHUFFLE_STORAGE = "durableShuffleStorage";
Review Comment:
Since we now have a fault-tolerant mode that is one step ahead of the shuffle storage, does it make sense to deprecate the durable shuffle storage mode entirely since the time benefits of just using the shuffle storage seem negligible in comparison to using the fault-tolerant mode? (My understanding is that most of the time would be spent in writing/reading from the durable storage which both cases do however with fault tolerance we also enable extra code paths which allow worker retry)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MSQTasks.java:
##########
@@ -54,6 +58,8 @@
private static final String TASK_ID_PREFIX = "query-";
+ private static final Pattern WORKER_PATTERN = Pattern.compile(".*-worker([0-9]+)_[0-9]+");
Review Comment:
One more question is the taskId sure not to contain a `-`, otherwise, the regex logic might not work as expected.
##########
core/src/test/java/org/apache/druid/java/util/common/function/TriConsumerTest.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.druid.java.util.common.function;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class TriConsumerTest
+{
+
+ @Test
+ public void sanityTest()
Review Comment:
I think there should be a test where the `consumerA` throws an exception after adding to the set, and then we should see if `consumerB` runs or not.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQueryHA.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.msq.util.MultiStageQueryContext;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.ITRetryUtil;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.utils.DruidClusterAdminClient;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQueryHA
+{
+ private static final Logger LOG = new Logger(ITMultiStageQueryHA.class);
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ private SqlResourceTestClient msqClient;
+
+ @Inject
+ private IntegrationTestingConfig config;
+
+ @Inject
+ private ObjectMapper jsonMapper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Inject
+ private CoordinatorResourceTestClient coordinatorClient;
+
+ @Inject
+ private DruidClusterAdminClient druidClusterAdminClient;
+
+ private static final String QUERY_FILE = "/multi-stage-query/wikipedia_msq_select_query_ha.json";
+
+ @Test
+ public void testMsqIngestionAndQuerying() throws Exception
+ {
+ String datasource = "dst";
+
+ // Clear up the datasource from the previous runs
+ coordinatorClient.unloadSegmentsForDataSource(datasource);
+
+ String queryLocal =
+ StringUtils.format(
+ "INSERT INTO %s\n"
+ + "SELECT\n"
+ + " TIME_PARSE(\"timestamp\") AS __time,\n"
+ + " isRobot,\n"
+ + " diffUrl,\n"
+ + " added,\n"
+ + " countryIsoCode,\n"
+ + " regionName,\n"
+ + " channel,\n"
+ + " flags,\n"
+ + " delta,\n"
+ + " isUnpatrolled,\n"
+ + " isNew,\n"
+ + " deltaBucket,\n"
+ + " isMinor,\n"
+ + " isAnonymous,\n"
+ + " deleted,\n"
+ + " cityName,\n"
+ + " metroCode,\n"
+ + " namespace,\n"
+ + " comment,\n"
+ + " page,\n"
+ + " commentLength,\n"
+ + " countryName,\n"
+ + " user,\n"
+ + " regionIsoCode\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + " '{\"type\":\"local\",\"files\":[\"/resources/data/batch_index/json/wikipedia_index_data1.json\",\"/resources/data/batch_index/json/wikipedia_index_data1.json\"]}',\n"
+ + " '{\"type\":\"json\"}',\n"
+ + " '[{\"type\":\"string\",\"name\":\"timestamp\"},{\"type\":\"string\",\"name\":\"isRobot\"},{\"type\":\"string\",\"name\":\"diffUrl\"},{\"type\":\"long\",\"name\":\"added\"},{\"type\":\"string\",\"name\":\"countryIsoCode\"},{\"type\":\"string\",\"name\":\"regionName\"},{\"type\":\"string\",\"name\":\"channel\"},{\"type\":\"string\",\"name\":\"flags\"},{\"type\":\"long\",\"name\":\"delta\"},{\"type\":\"string\",\"name\":\"isUnpatrolled\"},{\"type\":\"string\",\"name\":\"isNew\"},{\"type\":\"double\",\"name\":\"deltaBucket\"},{\"type\":\"string\",\"name\":\"isMinor\"},{\"type\":\"string\",\"name\":\"isAnonymous\"},{\"type\":\"long\",\"name\":\"deleted\"},{\"type\":\"string\",\"name\":\"cityName\"},{\"type\":\"long\",\"name\":\"metroCode\"},{\"type\":\"string\",\"name\":\"namespace\"},{\"type\":\"string\",\"name\":\"comment\"},{\"type\":\"string\",\"name\":\"page\"},{\"type\":\"long\",\"name\":\"commentLength\"},{\"type\":\"string\",\"name\":\"countryName\"},{\"type\":
\"string\",\"name\":\"user\"},{\"type\":\"string\",\"name\":\"regionIsoCode\"}]'\n"
+ + " )\n"
+ + ")\n"
+ + "PARTITIONED BY DAY\n"
+ + "CLUSTERED BY \"__time\"",
+ datasource
+ );
+
+ // Submit the task and wait for the datasource to get loaded
+ SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTask(
+ queryLocal,
+ ImmutableMap.of(
+ MultiStageQueryContext.CTX_DURABLE_SHUFFLE_STORAGE,
+ "true",
+ MultiStageQueryContext.CTX_MAX_NUM_TASKS,
+ 3
+ )
+ );
+
+ if (sqlTaskStatus.getState().isFailure()) {
Review Comment:
Since we are trying to coordinate between the task running and the IT running, I think we should explore into the `SLEEP()` function. I encountered this in `ITSQLCancelTest`, which allows the queries to run for a certain amount of time.
Taken from ITSqlCancelTest
```
/**
* This query will run exactly for 15 seconds.
*/
private static final String QUERY
= "SELECT sleep(CASE WHEN added > 0 THEN 1 ELSE 0 END) FROM wikipedia_editstream WHERE added > 0 LIMIT 15";
```
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MSQTasks.java:
##########
@@ -54,6 +58,8 @@
private static final String TASK_ID_PREFIX = "query-";
+ private static final Pattern WORKER_PATTERN = Pattern.compile(".*-worker([0-9]+)_[0-9]+");
Review Comment:
nit: I think there should be an explanation behind the pattern, or sample worker patterns which it should match. Also, there can be named groups, I think those should be used to make it clearer.
```suggestion
private static final Pattern WORKER_PATTERN = Pattern.compile(".*-worker(?<workerNumber>[0-9]+)_[0-9]+");
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] github-code-scanning[bot] commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064653537
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MSQTasks.java:
##########
@@ -65,9 +71,26 @@
/**
* Returns a worker task ID given a SQL query id.
*/
- public static String workerTaskId(final String controllerTaskId, final int workerNumber)
+ public static String workerTaskId(final String controllerTaskId, final int workerNumber, final int retryCount)
{
- return StringUtils.format("%s-worker%d", controllerTaskId, workerNumber);
+ return StringUtils.format("%s-worker%d_%d", controllerTaskId, workerNumber, retryCount);
+ }
+
+ /**
+ * Extract worker from taskId or throw exception if unable to parse out the worker.
+ */
+ public static int workerFromTaskId(final String taskId)
+ {
+ final Matcher matcher = WORKER_PATTERN.matcher(taskId);
+ if (matcher.matches()) {
+ return Integer.parseInt(matcher.group(1));
Review Comment:
## Missing catch of NumberFormatException
Potential uncaught 'java.lang.NumberFormatException'.
[Show more details](https://github.com/apache/druid/security/code-scanning/3606)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -165,10 +219,88 @@
}
}
+
+ /**
+ * Get workers which need to be sent partition boundaries
+ *
+ * @return
+ */
+ IntSet getWorkersToSendPartitionBoundaries()
+ {
+ if (!getStageDefinition().doesShuffle()) {
+ throw new ISE("Result partition information is not relevant to this stage because it does not shuffle");
+ }
+ IntAVLTreeSet workers = new IntAVLTreeSet();
+ for (Integer worker : workerToPhase.keySet()) {
+ if (ControllerWorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES.equals(workerToPhase.get(worker))) {
+ workers.add(worker);
Review Comment:
## Deprecated method or constructor invocation
Invoking [AbstractIntCollection.add](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/3602)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -283,6 +466,262 @@
return getPhase();
}
+ private void initializeTimeChunkWorkerTrackers()
+ {
+ workerToRemainingTimeChunks = new HashMap<>();
+ timeChunkToRemainingWorkers = new HashMap<>();
+ completeKeyStatisticsInformation.getTimeSegmentVsWorkerMap().forEach((timeChunk, workers) -> {
+ for (int worker : workers) {
+ this.workerToRemainingTimeChunks.compute(worker, (wk, timeChunks) -> {
+ if (timeChunks == null) {
+ timeChunks = new HashSet<>();
+ }
+ timeChunks.add(timeChunk);
+ return timeChunks;
+ });
+ }
+ timeChunkToRemainingWorkers.put(timeChunk, workers);
+ });
+ }
+
+
+ /**
+ * Merges the {@link ClusterByStatisticsSnapshot} for the worker, time chunk with the stage {@link ClusterByStatisticsCollector} being
+ * tracked at {@link #timeChunkToCollector} for the same time chunk. This method is called when
+ * {@link ClusterStatisticsMergeMode#SEQUENTIAL} is chosen eventually.
+ * <br></br>
+ * <br></br>
+ * If all the stats from the worker are merged, we transition the worker to {@link ControllerWorkerStagePhase#PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES};
+ * <br></br>
+ * If all the stats from all the workers are merged, we transition the stage to {@link ControllerStagePhase#POST_READING}
+ */
+
+ void mergeClusterByStatisticsCollectorForTimeChunk(
+ int workerNumber,
+ Long timeChunk,
+ ClusterByStatisticsSnapshot clusterByStatisticsSnapshot
+ )
+ {
+ if (!stageDef.mustGatherResultKeyStatistics()
+ || !stageDef.doesShuffle()) {
+ throw new ISE("Stage does not gather result key statistics");
+ }
+
+ if (workerNumber < 0 || workerNumber >= workerCount) {
+ throw new IAE("Invalid workerNumber [%s]", workerNumber);
+ }
+
+ if (completeKeyStatisticsInformation == null || !completeKeyStatisticsInformation.isComplete()) {
+ throw new ISE(
+ "Cannot merge worker[%d] time chunk until all the key information is received for stage[%d]",
+ workerNumber,
+ stageDef.getStageNumber()
+ );
+ }
+
+ ControllerWorkerStagePhase workerStagePhase = workerToPhase.get(workerNumber);
+
+ if (workerStagePhase == null) {
+ throw new ISE("Worker[%d] not found for stage[%s]", workerNumber, stageDef.getStageNumber());
+ }
+
+ // only merge in case this worker has remaining time chunks
+ workerToRemainingTimeChunks.computeIfPresent(workerNumber, (wk, timeChunks) -> {
+ if (timeChunks.remove(timeChunk)) {
+
+ // merge the key collector
+ timeChunkToCollector.compute(
+ timeChunk,
+ (ignored, collector) -> {
+ if (collector == null) {
+ collector = stageDef.createResultKeyStatisticsCollector(maxRetainedPartitionSketchBytes);
+ }
+ collector.addAll(clusterByStatisticsSnapshot);
+ return collector;
+ }
+ );
+
+ // if work for one time chunk is finished, generate the ClusterByPartitions for that timeChunk and clear the collector so that we free up controller memory.
+ timeChunkToRemainingWorkers.compute(timeChunk, (tc, workers) -> {
+ if (workers == null || workers.isEmpty()) {
+ throw new ISE(
+ "Remaining workers should not be empty until all the work is finished for time chunk[%d] for stage[%d]",
+ timeChunk,
+ stageDef.getStageNumber()
+ );
+ }
+ workers.remove(workerNumber);
+ if (workers.isEmpty()) {
+ // generate partition boundaries since all work is finished for the time chunk
+ ClusterByStatisticsCollector collector = timeChunkToCollector.get(tc);
+ Either<Long, ClusterByPartitions> countOrPartitions = stageDef.generatePartitionsForShuffle(collector);
+ totalPartitionCount += getPartitionCountFromEither(countOrPartitions);
+ if (totalPartitionCount > stageDef.getMaxPartitionCount()) {
+ failForReason(new TooManyPartitionsFault(stageDef.getMaxPartitionCount()));
+ return null;
+ }
+ timeChunkToBoundaries.put(tc, countOrPartitions.valueOrThrow());
+
+ // clear the collector to give back memory
+ collector.clear();
+ timeChunkToCollector.remove(tc);
+ return null;
+ }
+ return workers;
+ });
+ }
+ return timeChunks.isEmpty() ? null : timeChunks;
+ });
+
+
+ // if all time chunks for worker are taken care off transition worker.
+ if (workerToRemainingTimeChunks.get(workerNumber) == null) {
+ // adding worker to a set so that we do not fetch the worker collectors again.
+ workersFromWhichKeyCollectorFetched.add(workerNumber);
+ if (ControllerWorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES.canTransitionFrom(
+ workerStagePhase)) {
+ workerToPhase.put(workerNumber, ControllerWorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES);
+ } else {
+ throw new ISE(
+ "Worker[%d] for stage[%d] expected to be in state[%s]. Found state[%s]",
+ workerNumber,
+ (stageDef.getStageNumber()),
+ ControllerWorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES,
+ workerStagePhase
+
+ );
+ }
+ }
+
+
+ // if all time chunks have the partition boundaries, merge them to set resultPartitionBoundaries
+ if (workerToRemainingTimeChunks.isEmpty()) {
+ if (resultPartitionBoundaries == null) {
+ timeChunkToBoundaries.forEach((ignored, partitions) -> {
+ if (resultPartitionBoundaries == null) {
+ resultPartitionBoundaries = partitions;
+ } else {
+ abutAndAppendPartitionBoundaries(resultPartitionBoundaries.ranges(), partitions.ranges());
+ }
+ });
+ timeChunkToBoundaries.clear();
+ setClusterByPartitionBoundaries(resultPartitionBoundaries);
+ } else {
+ // we already have result partitions. We can safely transition to POST READING and submit the result boundaries to the workers.
+ transitionTo(ControllerStagePhase.POST_READING);
+ }
+ }
+
+ }
+
+ /**
+ * Merges the entire {@link ClusterByStatisticsSnapshot} for the worker with the stage {@link ClusterByStatisticsCollector} being
+ * tracked at {@link #timeChunkToCollector} with key {@link ControllerStageTracker#STATIC_TIME_CHUNK_FOR_PARALLEL_MERGE}. This method is called when
+ * {@link ClusterStatisticsMergeMode#PARALLEL} is chosen eventually.
+ * <br></br>
+ * <br></br>
+ * If all the stats from the worker are merged, we transition the worker to {@link ControllerWorkerStagePhase#PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES}.
+ * <br></br>
+ * If all the stats from all the workers are merged, we transition the stage to {@link ControllerStagePhase#POST_READING}.
+ */
+
+ void mergeClusterByStatisticsCollectorForAllTimeChunks(
+ int workerNumber,
+ ClusterByStatisticsSnapshot clusterByStatsSnapshot
+ )
+ {
+ if (!stageDef.mustGatherResultKeyStatistics()
+ || !stageDef.doesShuffle()) {
+ throw new ISE("Stage does not gather result key statistics");
+ }
+
+ if (workerNumber < 0 || workerNumber >= workerCount) {
+ throw new IAE("Invalid workerNumber [%s]", workerNumber);
+ }
+
+ ControllerWorkerStagePhase workerStagePhase = workerToPhase.get(workerNumber);
+
+ if (workerStagePhase == null) {
+ throw new ISE("Worker[%d] not found for stage[%s]", workerNumber, stageDef.getStageNumber());
+ }
+
+
+ // To prevent the case where we do not fetch the collector twice, like when worker is retried, we should be okay with the
+ // older collector from the previous run of the worker.
+
+ if (workersFromWhichKeyCollectorFetched.add(workerNumber)) {
+ // in case of parallel merge we use the "ALL" granularity start time to put the sketches
+ timeChunkToCollector.compute(
+ STATIC_TIME_CHUNK_FOR_PARALLEL_MERGE,
+ (timeChunk, stats) -> {
+ if (stats == null) {
+ stats = stageDef.createResultKeyStatisticsCollector(maxRetainedPartitionSketchBytes);
+ }
+ stats.addAll(clusterByStatsSnapshot);
+ return stats;
+ }
+ );
+ } else {
+ log.debug("Already have key collector for worker[%d] stage[%d]", workerNumber, stageDef.getStageNumber());
+ }
+
+ if (ControllerWorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES.canTransitionFrom(workerStagePhase)) {
+ workerToPhase.put(workerNumber, ControllerWorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES);
+ } else {
+ throw new ISE(
+ "Worker[%d] for stage[%d] expected to be in state[%s]. Found state[%s]",
+ workerNumber,
+ (stageDef.getStageNumber()),
+ ControllerWorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES,
+ workerStagePhase
+
+ );
+ }
+
+ if (allResultsStatsFetched()) {
+ if (completeKeyStatisticsInformation == null || !completeKeyStatisticsInformation.isComplete()) {
+ throw new ISE(
+ "Cannot generate partition boundaries until all the key information is received for stage[%d]",
+ workerNumber,
+ stageDef.getStageNumber()
+ );
Review Comment:
## Unused format argument
This format call refers to 1 argument(s) but supplies 2 argument(s).
[Show more details](https://github.com/apache/druid/security/code-scanning/3604)
##########
extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQSelectTest.java:
##########
@@ -1254,13 +1299,15 @@
.setExpectedRowSignature(rowSignature)
.setExpectedResultRows(ImmutableList.of(new Object[]{1L, 6L}))
.verifyResults();
- File successFile = new File(
- localFileStorageDir,
- DurableStorageUtils.getSuccessFilePath("query-test-query", 0, 0)
- );
+ if (DURABLE_STORAGE.equals(contextName) || !FAULT_TOLERANCE.equals(contextName)) {
+ File successFile = new File(
+ localFileStorageDir,
+ DurableStorageUtils.getSuccessFilePath("query-test-query", 0, 0)
+ );
Review Comment:
## Unread local variable
Variable 'File successFile' is never read.
[Show more details](https://github.com/apache/druid/security/code-scanning/3605)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerSketchFetcher.java:
##########
@@ -50,322 +54,236 @@
{
private static final Logger log = new Logger(WorkerSketchFetcher.class);
private static final int DEFAULT_THREAD_COUNT = 4;
- // If the combined size of worker sketches is more than this threshold, SEQUENTIAL merging mode is used.
- static final long BYTES_THRESHOLD = 1_000_000_000L;
- // If there are more workers than this threshold, SEQUENTIAL merging mode is used.
- static final long WORKER_THRESHOLD = 100;
- private final ClusterStatisticsMergeMode clusterStatisticsMergeMode;
- private final int statisticsMaxRetainedBytes;
private final WorkerClient workerClient;
- private final ExecutorService executorService;
+ private final MSQWorkerTaskLauncher workerTaskLauncher;
+
+ private final boolean retryEnabled;
+
+ private AtomicReference<Throwable> isError = new AtomicReference<>();
+ final ExecutorService executorService;
+
public WorkerSketchFetcher(
WorkerClient workerClient,
- ClusterStatisticsMergeMode clusterStatisticsMergeMode,
- int statisticsMaxRetainedBytes
+ MSQWorkerTaskLauncher workerTaskLauncher,
+ boolean retryEnabled
)
{
this.workerClient = workerClient;
- this.clusterStatisticsMergeMode = clusterStatisticsMergeMode;
this.executorService = Execs.multiThreaded(DEFAULT_THREAD_COUNT, "SketchFetcherThreadPool-%d");
- this.statisticsMaxRetainedBytes = statisticsMaxRetainedBytes;
+ this.workerTaskLauncher = workerTaskLauncher;
+ this.retryEnabled = retryEnabled;
}
/**
- * Submits a request to fetch and generate partitions for the given worker statistics and returns a future for it. It
- * decides based on the statistics if it should fetch sketches one by one or together.
+ * Fetches the full {@link ClusterByStatisticsCollector} from all workers and generates partition boundaries from them.
+ * This is faster than fetching them timechunk by timechunk but the collector will be downsampled till it can fit
+ * on the controller, resulting in less accurate partition boundries.
*/
- public CompletableFuture<Either<Long, ClusterByPartitions>> submitFetcherTask(
- CompleteKeyStatisticsInformation completeKeyStatisticsInformation,
- List<String> workerTaskIds,
- StageDefinition stageDefinition,
- IntSet workersForStage
+ public void inMemoryFullSketchMerging(
+ Consumer<Consumer<ControllerQueryKernel>> kernelActions,
+ StageId stageId,
+ Set<String> taskIds,
+ TriConsumer<ControllerQueryKernel, Integer, MSQFault> retryOperation
)
{
- ClusterBy clusterBy = stageDefinition.getClusterBy();
- switch (clusterStatisticsMergeMode) {
- case SEQUENTIAL:
- return sequentialTimeChunkMerging(completeKeyStatisticsInformation, stageDefinition, workerTaskIds);
- case PARALLEL:
- return inMemoryFullSketchMerging(stageDefinition, workerTaskIds, workersForStage);
- case AUTO:
- if (clusterBy.getBucketByCount() == 0) {
- log.info(
- "Query[%s] stage[%d] for AUTO mode: chose PARALLEL mode to merge key statistics",
- stageDefinition.getId().getQueryId(),
- stageDefinition.getStageNumber()
- );
- // If there is no time clustering, there is no scope for sequential merge
- return inMemoryFullSketchMerging(stageDefinition, workerTaskIds, workersForStage);
- } else if (stageDefinition.getMaxWorkerCount() > WORKER_THRESHOLD
- || completeKeyStatisticsInformation.getBytesRetained() > BYTES_THRESHOLD) {
- log.info(
- "Query[%s] stage[%d] for AUTO mode: chose SEQUENTIAL mode to merge key statistics",
- stageDefinition.getId().getQueryId(),
- stageDefinition.getStageNumber()
+ for (String taskId : taskIds) {
+ try {
+ int workerNumber = MSQTasks.workerFromTaskId(taskId);
+ executorService.submit(() -> {
+ fetchStatsFromWorker(
+ kernelActions,
+ () -> workerClient.fetchClusterByStatisticsSnapshot(
+ taskId,
+ stageId.getQueryId(),
+ stageId.getStageNumber()
+ ),
+ taskId,
+ (kernel, snapshot) -> kernel.mergeClusterByStatisticsCollectorForAllTimeChunks(
+ stageId,
+ workerNumber,
+ snapshot
+ ),
+ retryOperation
);
- return sequentialTimeChunkMerging(completeKeyStatisticsInformation, stageDefinition, workerTaskIds);
+ });
+ }
+ catch (RejectedExecutionException rejectedExecutionException) {
+ if (isError.get() == null) {
+ throw rejectedExecutionException;
+ } else {
+ // throw worker error exception
+ throw new ISE("Unable to fetch partitions ", isError.get());
Review Comment:
## Unused format argument
This format call refers to 0 argument(s) but supplies 1 argument(s).
[Show more details](https://github.com/apache/druid/security/code-scanning/3603)
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStageTracker.java:
##########
@@ -165,10 +219,88 @@
}
}
+
+ /**
+ * Get workers which need to be sent partition boundaries
+ *
+ * @return
+ */
+ IntSet getWorkersToSendPartitionBoundaries()
+ {
+ if (!getStageDefinition().doesShuffle()) {
+ throw new ISE("Result partition information is not relevant to this stage because it does not shuffle");
+ }
+ IntAVLTreeSet workers = new IntAVLTreeSet();
+ for (Integer worker : workerToPhase.keySet()) {
+ if (ControllerWorkerStagePhase.PRESHUFFLE_WAITING_FOR_RESULT_PARTITION_BOUNDARIES.equals(workerToPhase.get(worker))) {
Review Comment:
## Deprecated method or constructor invocation
Invoking [Int2ObjectMap.get](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/3601)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064640749
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQueryHA.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.msq.util.MultiStageQueryContext;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.ITRetryUtil;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.utils.DruidClusterAdminClient;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQueryHA
Review Comment:
HA means high availability. Changed it to a better name.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1023465226
##########
docs/multi-stage-query/reference.md:
##########
@@ -265,7 +268,9 @@ The following table describes error codes you may encounter in the `multiStageQu
| `TooManyColumns` | Exceeded the number of columns for a stage. See the [Limits](#limits) table for the specific limit. | `numColumns`: The number of columns requested.<br /><br />`maxColumns`: The limit on columns which was exceeded. |
| `TooManyWarnings` | Exceeded the allowed number of warnings of a particular type. | `rootErrorCode`: The error code corresponding to the exception that exceeded the required limit. <br /><br />`maxWarnings`: Maximum number of warnings that are allowed for the corresponding `rootErrorCode`. |
| `TooManyWorkers` | Exceeded the supported number of workers running simultaneously. See the [Limits](#limits) table for the specific limit. | `workers`: The number of simultaneously running workers that exceeded a hard or soft limit. This may be larger than the number of workers in any one stage if multiple stages are running simultaneously. <br /><br />`maxWorkers`: The hard or soft limit on workers that was exceeded. |
+| `TotalRelaunchLimitExceededFault` | Total relaunch count across all workers exceeded max relaunch limit. See the [Limits](#limits) table for the specific limit. | `maxRelaunchCount`: Max number of relaunches across all the workers defined in the [Limits](#limits) section. <br /><br /> `currentRelaunchCount`: current relaunch counter for the job. <br /><br /> `taskId`: Latest task id which failed <br /> <br /> `rootErrorMessage`: Error message of the latest failed task.|
Review Comment:
Can you please clarify if `currentRelaunchCount` is the relaunch count for the current worker or is the cumulative relaunch count across all the workers?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -551,6 +563,15 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
id(),
task.getDataSource(),
context,
+ (failedTask, fault) -> {
+ addToKernelManipulationQueue((kernel) -> {
+ if (isDurableStorageEnabled) {
Review Comment:
I think across this PR and https://github.com/apache/druid/pull/13368, we might need to unify what enabling durable storage, enabling fault tolerance, and enabling durable storage for intermediate steps mean since that might cause confusion in the code as well as to the end user configuring the properties.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/WorkerRelaunchedTooManyTimes.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.druid.msq.indexing.error;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+
+import java.util.Objects;
+
+@JsonTypeName(WorkerRelaunchedTooManyTimes.CODE)
+public class WorkerRelaunchedTooManyTimes extends BaseMSQFault
+{
+ static final String CODE = "WorkerRelaunchedTooManyTimes";
+
+
+ private final int maxPerWorkerRelaunchCount;
+
+ private final String taskId;
+
+ private final int workerNumber;
+
+
+ private final String rootErrorMessage;
+
+ @JsonCreator
+ public WorkerRelaunchedTooManyTimes(
+ @JsonProperty("maxPerWorkerRelaunchCount") int maxPerWorkerRelaunchCount,
+ @JsonProperty("taskId") String taskId,
+ @JsonProperty("workerNumber") int workerNumber,
+ @JsonProperty("rootErrorMessage") String rootErrorMessage
+ )
+ {
+ super(
+ CODE,
+ "Worker[%d] exceeded max relaunch count of %d for task[%s]. Latest failure reason: %s.",
+ workerNumber,
+ maxPerWorkerRelaunchCount,
+ taskId,
+ rootErrorMessage
+ );
+ this.maxPerWorkerRelaunchCount = maxPerWorkerRelaunchCount;
+ this.taskId = taskId;
+ this.workerNumber = workerNumber;
+ this.rootErrorMessage = rootErrorMessage;
+ }
+
+ @JsonProperty
+ public int getMaxPerWorkerRelaunchCount()
+ {
+ return maxPerWorkerRelaunchCount;
+ }
+
+ @JsonProperty
+ public int getWorkerNumber()
+ {
+ return workerNumber;
+ }
+
+ @JsonProperty
+ public String getTaskId()
+ {
+ return taskId;
+ }
+
+ @JsonProperty
+ public String getRootErrorMessage()
+ {
+ return rootErrorMessage;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ WorkerRelaunchedTooManyTimes that = (WorkerRelaunchedTooManyTimes) o;
+ return maxPerWorkerRelaunchCount == that.maxPerWorkerRelaunchCount
Review Comment:
nit: Indentation seems a bit off
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -546,4 +614,65 @@ private static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefi
return retVal;
}
+
+ /**
+ * Checks the {@link MSQFault#getErrorCode()} is eligible for retry.
+ * <br/>
+ * If yes, transitions the stage to{@link ControllerStagePhase#RETRYING} and returns all the {@link WorkOrder}
+ * <br/>
+ * else throw {@link MSQException}
+ *
+ * @param workerNumber
+ * @param msqFault
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ public List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int workerNumber, MSQFault msqFault)
+ {
+
+ final String errorCode;
+ if (msqFault instanceof WorkerFailedFault) {
+ errorCode = MSQFaultUtils.getErrorCodeFromMessage((((WorkerFailedFault) msqFault).getErrorMsg()));
+ } else {
+ errorCode = msqFault.getErrorCode();
+ }
+
+ if (retriableErrorCodes.contains(errorCode)) {
+ return getWorkInCaseWorkerElgibileForRetryElseThrow(workerNumber);
+
+ } else {
+ throw new MSQException(msqFault);
+ }
+ }
+
+ /**
+ * Gets all the stages currently being tracked and filters out all effectively finished stages.
+ * <br/>
+ * From the remaining stages, checks if (stage,worker) needs to be retried.
+ * <br/>
+ * If yes adds the workOrder for that stage to the return list and transitions the stage kernel to {@link ControllerStagePhase#RETRYING}
+ *
+ * @param worker
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ private List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int worker)
+ {
+ List<StageId> trackedSet = new ArrayList<>(getActiveStages());
+ // no need to retry effectively finished stages
+ List<StageId> getEffictivelyFinishedStages = getEffectivelyFinishedStageIds();
Review Comment:
nit: spelling
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStagePhase.java:
##########
@@ -35,7 +35,7 @@
@Override
public boolean canTransitionFrom(final ControllerStagePhase priorPhase)
{
- return false;
+ return true;
Review Comment:
`NEW` should be the initial state and therefore initialized when we are creating the `ControllerStagePhase`. We shouldn't allow any transition to `NEW` I think.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -546,4 +614,65 @@ private static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefi
return retVal;
}
+
+ /**
+ * Checks the {@link MSQFault#getErrorCode()} is eligible for retry.
+ * <br/>
+ * If yes, transitions the stage to{@link ControllerStagePhase#RETRYING} and returns all the {@link WorkOrder}
+ * <br/>
+ * else throw {@link MSQException}
+ *
+ * @param workerNumber
+ * @param msqFault
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ public List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int workerNumber, MSQFault msqFault)
+ {
+
+ final String errorCode;
+ if (msqFault instanceof WorkerFailedFault) {
+ errorCode = MSQFaultUtils.getErrorCodeFromMessage((((WorkerFailedFault) msqFault).getErrorMsg()));
+ } else {
+ errorCode = msqFault.getErrorCode();
+ }
+
+ if (retriableErrorCodes.contains(errorCode)) {
+ return getWorkInCaseWorkerElgibileForRetryElseThrow(workerNumber);
+
+ } else {
+ throw new MSQException(msqFault);
+ }
+ }
+
+ /**
+ * Gets all the stages currently being tracked and filters out all effectively finished stages.
+ * <br/>
+ * From the remaining stages, checks if (stage,worker) needs to be retried.
+ * <br/>
+ * If yes adds the workOrder for that stage to the return list and transitions the stage kernel to {@link ControllerStagePhase#RETRYING}
+ *
+ * @param worker
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ private List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int worker)
Review Comment:
This method is not thrown anywhere. We should rename it.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064622570
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java:
##########
@@ -58,10 +58,14 @@
public static final String CTX_FINALIZE_AGGREGATIONS = "finalizeAggregations";
private static final boolean DEFAULT_FINALIZE_AGGREGATIONS = true;
- public static final String CTX_ENABLE_DURABLE_SHUFFLE_STORAGE = "durableShuffleStorage";
+ public static final String CTX_DURABLE_SHUFFLE_STORAGE = "durableShuffleStorage";
Review Comment:
I have featured flagged it just in case we break something. In the future, yes we should remove the durableShuffleStorage context param.
The way it works is
if the user sets faultTolerance=true, I enable durableShuffleStorage as well
if the user sets faultTolerance=true, and durableStorage=false its an error.
if the user sets durableStorage=true, faultTolerance context param remains untouched.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe merged pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe merged PR #13353:
URL: https://github.com/apache/druid/pull/13353
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1034617788
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/Limits.java:
##########
@@ -58,4 +58,14 @@
* Maximum size of the kernel manipulation queue in {@link org.apache.druid.msq.indexing.MSQControllerTask}.
*/
public static final int MAX_KERNEL_MANIPULATION_QUEUE_SIZE = 100_000;
+
+ /**
+ * Maximum retries across all workers.
+ */
+ public static final int TOTAL_RETRY_LIMIT = 30;
Review Comment:
Changed this to 100 for now. As this PR is already quite big, we can make this limit dynamic in another PR.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1022617563
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/Limits.java:
##########
@@ -58,4 +58,14 @@
* Maximum size of the kernel manipulation queue in {@link org.apache.druid.msq.indexing.MSQControllerTask}.
*/
public static final int MAX_KERNEL_MANIPULATION_QUEUE_SIZE = 100_000;
+
+ /**
+ * Maximum retries across all workers.
+ */
+ public static final int TOTAL_RETRY_LIMIT = 30;
Review Comment:
Should this be related to the number of workers running the query instead of being a fixed constant? A server with a large number of task slots might shut down temporarily, but restarting the tasks on it would consume a huge chunk of the limit.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/MSQFaultUtils.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.druid.msq.indexing.error;
+
+public class MSQFaultUtils
+{
+
+ public static final String ERROR_CODE_DELIMITER = ": ";
+
+ /**
+ * Generate string message with error code delimited by {@link MSQFaultUtils#ERROR_CODE_DELIMITER}
+ */
+ public static String generateMessageWithErrorCode(MSQFault msqFault)
+ {
+ final String message = msqFault.getErrorMessage();
+
+ if (message != null && !message.isEmpty()) {
+ return msqFault.getErrorCode() + ERROR_CODE_DELIMITER + message;
+ } else {
+ return msqFault.getErrorCode();
+ }
+ }
+
+ /**
+ * Gets the error code from the message. If the messay is empty or null, {@link UnknownFault#CODE} is returned. This method
+ * does not gurantee that the error code we get out of the message is a valid error code.
+ */
+ public static String getErrorCodeFromMessage(String message)
Review Comment:
Instead of relying on the manual parsing of the error codes from the message string and defining our own error message schema, can we use jsonMapper somehow to do it for us?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/MSQFaultUtils.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.druid.msq.indexing.error;
+
+public class MSQFaultUtils
+{
+
+ public static final String ERROR_CODE_DELIMITER = ": ";
+
+ /**
+ * Generate string message with error code delimited by {@link MSQFaultUtils#ERROR_CODE_DELIMITER}
+ */
+ public static String generateMessageWithErrorCode(MSQFault msqFault)
+ {
+ final String message = msqFault.getErrorMessage();
+
+ if (message != null && !message.isEmpty()) {
+ return msqFault.getErrorCode() + ERROR_CODE_DELIMITER + message;
+ } else {
+ return msqFault.getErrorCode();
+ }
+ }
+
+ /**
+ * Gets the error code from the message. If the messay is empty or null, {@link UnknownFault#CODE} is returned. This method
+ * does not gurantee that the error code we get out of the message is a valid error code.
+ */
+ public static String getErrorCodeFromMessage(String message)
+ {
+ if (message == null || message.isEmpty()) {
+ return UnknownFault.CODE;
+ }
+ return message.split(ERROR_CODE_DELIMITER, 2)[0];
Review Comment:
A check that this doesn't go out of bounds (in case generateMessageWithErrorCode wasn't invoked at a point in the future) would be helpful (we can return UnknownFault.CODE in that case).
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -546,4 +614,65 @@ private static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefi
return retVal;
}
+
+ /**
+ * Checks the {@link MSQFault#getErrorCode()} is eligible for retry.
+ * <br/>
+ * If yes, transitions the stage to{@link ControllerStagePhase#RETRYING} and returns all the {@link WorkOrder}
+ * <br/>
+ * else throw {@link MSQException}
+ *
+ * @param workerNumber
+ * @param msqFault
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ public List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int workerNumber, MSQFault msqFault)
Review Comment:
nit:
1. Spelling
```suggestion
public List<WorkOrder> getWorkInCaseWorkerEligibleForRetryElseThrow(int workerNumber, MSQFault msqFault)
```
2. This method name seems unwieldy. Can we drop the elseThrow part, something like `getWorkFromRetriableWorkers` seems cleaner to me. Opinions?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/TotalRelaunchLimitExceededFault.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.druid.msq.indexing.error;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+
+import java.util.Objects;
+
+@JsonTypeName(TotalRelaunchLimitExceededFault.CODE)
+public class TotalRelaunchLimitExceededFault extends BaseMSQFault
Review Comment:
```suggestion
public class TotalRetryLimitExceededFault extends BaseMSQFault
```
Should we change relaunch to retry everywhere in this fault class?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033606783
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerStagePhase.java:
##########
@@ -35,7 +35,7 @@
@Override
public boolean canTransitionFrom(final ControllerStagePhase priorPhase)
{
- return false;
+ return true;
Review Comment:
It should not. Seems like leftover code.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1023468190
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -546,4 +614,65 @@ private static Map<StageId, Set<StageId>> computeStageOutflowMap(final QueryDefi
return retVal;
}
+
+ /**
+ * Checks the {@link MSQFault#getErrorCode()} is eligible for retry.
+ * <br/>
+ * If yes, transitions the stage to{@link ControllerStagePhase#RETRYING} and returns all the {@link WorkOrder}
+ * <br/>
+ * else throw {@link MSQException}
+ *
+ * @param workerNumber
+ * @param msqFault
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ public List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int workerNumber, MSQFault msqFault)
+ {
+
+ final String errorCode;
+ if (msqFault instanceof WorkerFailedFault) {
+ errorCode = MSQFaultUtils.getErrorCodeFromMessage((((WorkerFailedFault) msqFault).getErrorMsg()));
+ } else {
+ errorCode = msqFault.getErrorCode();
+ }
+
+ if (retriableErrorCodes.contains(errorCode)) {
+ return getWorkInCaseWorkerElgibileForRetryElseThrow(workerNumber);
+
+ } else {
+ throw new MSQException(msqFault);
+ }
+ }
+
+ /**
+ * Gets all the stages currently being tracked and filters out all effectively finished stages.
+ * <br/>
+ * From the remaining stages, checks if (stage,worker) needs to be retried.
+ * <br/>
+ * If yes adds the workOrder for that stage to the return list and transitions the stage kernel to {@link ControllerStagePhase#RETRYING}
+ *
+ * @param worker
+ * @return List of {@link WorkOrder} that needs to be retried.
+ */
+ private List<WorkOrder> getWorkInCaseWorkerElgibileForRetryElseThrow(int worker)
Review Comment:
This method does not throw anywhere. I think should rename it.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1023528416
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/RetryTask.java:
##########
@@ -0,0 +1,33 @@
+/*
+ * 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.druid.msq.indexing;
+
+import org.apache.druid.msq.indexing.error.MSQFault;
+
+public interface RetryTask
Review Comment:
I think this would be better named as `RelaunchTask` if we are going with relaunch instead of retry everywhere. WDYT?
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java:
##########
@@ -507,5 +689,23 @@ public boolean didRunTimeOut(final long maxTaskStartDelayMillis)
&& unknownLocation()
&& System.currentTimeMillis() - startTimeMs > maxTaskStartDelayMillis;
}
+
+ /**
+ * Enables retrying for the task
+ */
+ public void enableRetrying()
Review Comment:
```suggestion
public void enableRelaunching()
```
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTask.java:
##########
@@ -58,11 +61,12 @@ public MSQWorkerTask(
@JsonProperty("controllerTaskId") final String controllerTaskId,
@JsonProperty("dataSource") final String dataSource,
@JsonProperty("workerNumber") final int workerNumber,
- @JsonProperty("context") final Map<String, Object> context
+ @JsonProperty("context") final Map<String, Object> context,
+ @JsonProperty(value = "retry", defaultValue = "0") final int retryCount
Review Comment:
```suggestion
@JsonProperty(value = "relaunch", defaultValue = "0") final int relaunchCount
```
I think we should drop retry in favour of relaunch (or change everything back to retry) to avoid confusion unless they mean different things. WDYT?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064638851
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQueryHA.java:
##########
@@ -0,0 +1,187 @@
+/*
+ * 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.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.msq.util.MultiStageQueryContext;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.ITRetryUtil;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.utils.DruidClusterAdminClient;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQueryHA
+{
+ private static final Logger LOG = new Logger(ITMultiStageQueryHA.class);
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ private SqlResourceTestClient msqClient;
+
+ @Inject
+ private IntegrationTestingConfig config;
+
+ @Inject
+ private ObjectMapper jsonMapper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Inject
+ private CoordinatorResourceTestClient coordinatorClient;
+
+ @Inject
+ private DruidClusterAdminClient druidClusterAdminClient;
+
+ private static final String QUERY_FILE = "/multi-stage-query/wikipedia_msq_select_query_ha.json";
+
+ @Test
+ public void testMsqIngestionAndQuerying() throws Exception
+ {
+ String datasource = "dst";
+
+ // Clear up the datasource from the previous runs
+ coordinatorClient.unloadSegmentsForDataSource(datasource);
+
+ String queryLocal =
+ StringUtils.format(
+ "INSERT INTO %s\n"
+ + "SELECT\n"
+ + " TIME_PARSE(\"timestamp\") AS __time,\n"
+ + " isRobot,\n"
+ + " diffUrl,\n"
+ + " added,\n"
+ + " countryIsoCode,\n"
+ + " regionName,\n"
+ + " channel,\n"
+ + " flags,\n"
+ + " delta,\n"
+ + " isUnpatrolled,\n"
+ + " isNew,\n"
+ + " deltaBucket,\n"
+ + " isMinor,\n"
+ + " isAnonymous,\n"
+ + " deleted,\n"
+ + " cityName,\n"
+ + " metroCode,\n"
+ + " namespace,\n"
+ + " comment,\n"
+ + " page,\n"
+ + " commentLength,\n"
+ + " countryName,\n"
+ + " user,\n"
+ + " regionIsoCode\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + " '{\"type\":\"local\",\"files\":[\"/resources/data/batch_index/json/wikipedia_index_data1.json\",\"/resources/data/batch_index/json/wikipedia_index_data1.json\"]}',\n"
+ + " '{\"type\":\"json\"}',\n"
+ + " '[{\"type\":\"string\",\"name\":\"timestamp\"},{\"type\":\"string\",\"name\":\"isRobot\"},{\"type\":\"string\",\"name\":\"diffUrl\"},{\"type\":\"long\",\"name\":\"added\"},{\"type\":\"string\",\"name\":\"countryIsoCode\"},{\"type\":\"string\",\"name\":\"regionName\"},{\"type\":\"string\",\"name\":\"channel\"},{\"type\":\"string\",\"name\":\"flags\"},{\"type\":\"long\",\"name\":\"delta\"},{\"type\":\"string\",\"name\":\"isUnpatrolled\"},{\"type\":\"string\",\"name\":\"isNew\"},{\"type\":\"double\",\"name\":\"deltaBucket\"},{\"type\":\"string\",\"name\":\"isMinor\"},{\"type\":\"string\",\"name\":\"isAnonymous\"},{\"type\":\"long\",\"name\":\"deleted\"},{\"type\":\"string\",\"name\":\"cityName\"},{\"type\":\"long\",\"name\":\"metroCode\"},{\"type\":\"string\",\"name\":\"namespace\"},{\"type\":\"string\",\"name\":\"comment\"},{\"type\":\"string\",\"name\":\"page\"},{\"type\":\"long\",\"name\":\"commentLength\"},{\"type\":\"string\",\"name\":\"countryName\"},{\"type\":
\"string\",\"name\":\"user\"},{\"type\":\"string\",\"name\":\"regionIsoCode\"}]'\n"
+ + " )\n"
+ + ")\n"
+ + "PARTITIONED BY DAY\n"
+ + "CLUSTERED BY \"__time\"",
+ datasource
+ );
+
+ // Submit the task and wait for the datasource to get loaded
+ SqlTaskStatus sqlTaskStatus = msqHelper.submitMsqTask(
+ queryLocal,
+ ImmutableMap.of(
+ MultiStageQueryContext.CTX_DURABLE_SHUFFLE_STORAGE,
+ "true",
+ MultiStageQueryContext.CTX_MAX_NUM_TASKS,
+ 3
+ )
+ );
+
+ if (sqlTaskStatus.getState().isFailure()) {
+ Assert.fail(StringUtils.format(
+ "Unable to start the task successfully.\nPossible exception: %s",
+ sqlTaskStatus.getError()
+ ));
+ }
+
+
+ String taskIdToKill = sqlTaskStatus.getTaskId() + "-worker1_0";
+ killTaskAbruptly(taskIdToKill);
+
+ msqHelper.pollTaskIdForCompletion(sqlTaskStatus.getTaskId());
+ dataLoaderHelper.waitUntilDatasourceIsReady(datasource);
+
+ msqHelper.testQueriesFromFile(QUERY_FILE, datasource);
+ }
+
+ private void killTaskAbruptly(String taskIdToKill)
Review Comment:
Another way is to use a flag which is passed in the processing stack.
Then in the main code you check if this flag is set, exit the jvm with a non zero code.
As that would require us to change a lot of places, I have gone with this approach.
If more things start using this method, then maybe we can put it there.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064626757
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -603,6 +652,7 @@ public void updateStatus(int stageNumber, int workerNumber, Object keyStatistics
@Override
public void workerError(MSQErrorReport errorReport)
{
+ // move inside kernel
Review Comment:
removed this. Thanks for the catch.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064640320
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/Limits.java:
##########
@@ -68,4 +68,20 @@
* Maximum size of the kernel manipulation queue in {@link org.apache.druid.msq.indexing.MSQControllerTask}.
*/
public static final int MAX_KERNEL_MANIPULATION_QUEUE_SIZE = 100_000;
+
+ /**
+ * Maximum relaunches across all workers.
+ */
+ public static final int TOTAL_RELAUNCH_LIMIT = 100;
+
+ /**
+ * Maximum relaunches per worker. Initial run is not a relaunch. The worker will be spawned 1 + workerRelaunchLimit times before erroring out.
+ */
+ public static final int PER_WORKER_RELAUNCH_LIMIT = 2;
Review Comment:
I think 2 is fine. I would argue to keep it 1 :) since if the job fails once let's say due to oom, the likelyhood of the job failing again is very high.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064643950
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/MSQTasks.java:
##########
@@ -54,6 +58,8 @@
private static final String TASK_ID_PREFIX = "query-";
+ private static final Pattern WORKER_PATTERN = Pattern.compile(".*-worker([0-9]+)_[0-9]+");
Review Comment:
Yes tasks id should contain -
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033651666
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -44,18 +52,20 @@
import org.apache.druid.msq.statistics.ClusterByStatisticsSnapshot;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
Review Comment:
Yes we can do that. Changed it back.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033560023
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/MSQFaultUtils.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.druid.msq.indexing.error;
+
+public class MSQFaultUtils
+{
+
+ public static final String ERROR_CODE_DELIMITER = ": ";
+
+ /**
+ * Generate string message with error code delimited by {@link MSQFaultUtils#ERROR_CODE_DELIMITER}
+ */
+ public static String generateMessageWithErrorCode(MSQFault msqFault)
+ {
+ final String message = msqFault.getErrorMessage();
+
+ if (message != null && !message.isEmpty()) {
+ return msqFault.getErrorCode() + ERROR_CODE_DELIMITER + message;
+ } else {
+ return msqFault.getErrorCode();
+ }
+ }
+
+ /**
+ * Gets the error code from the message. If the messay is empty or null, {@link UnknownFault#CODE} is returned. This method
+ * does not gurantee that the error code we get out of the message is a valid error code.
+ */
+ public static String getErrorCodeFromMessage(String message)
+ {
+ if (message == null || message.isEmpty()) {
+ return UnknownFault.CODE;
+ }
+ return message.split(ERROR_CODE_DELIMITER, 2)[0];
Review Comment:
Added a few tests for this and caught a case. 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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] lgtm-com[bot] commented on pull request #13353: [DRAFT] Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on PR #13353:
URL: https://github.com/apache/druid/pull/13353#issuecomment-1311477872
This pull request **introduces 1 alert** when merging 74fa66f37bfb1f5a9a7806393e6510cee6cedc0f into 56d5c9780d6c09d6844136a6c678e3b417388bfe - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-9ebefcd5dd5d77b07f574176476326d1a1971f6b)
**new alerts:**
* 1 for Container contents are never accessed
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1034636851
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/MSQFaultUtils.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.druid.msq.indexing.error;
+
+public class MSQFaultUtils
+{
+
+ public static final String ERROR_CODE_DELIMITER = ": ";
+
+ /**
+ * Generate string message with error code delimited by {@link MSQFaultUtils#ERROR_CODE_DELIMITER}
+ */
+ public static String generateMessageWithErrorCode(MSQFault msqFault)
+ {
+ final String message = msqFault.getErrorMessage();
+
+ if (message != null && !message.isEmpty()) {
+ return msqFault.getErrorCode() + ERROR_CODE_DELIMITER + message;
+ } else {
+ return msqFault.getErrorCode();
+ }
+ }
+
+ /**
+ * Gets the error code from the message. If the messay is empty or null, {@link UnknownFault#CODE} is returned. This method
Review Comment:
As the workerFailed fault is thrown in MSQWorkerTaskLauncher#checkForErroneousTasks.
As the taskLauncher always contacts the OL, for the task status which is intern fetched from the metastore, we need this utility to persist the errorCode inside the db and then eventually parse out the error code from there.
I am not sure how we can remove this utility.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033617245
##########
docs/multi-stage-query/reference.md:
##########
@@ -234,6 +234,9 @@ The following table lists query limits:
| Number of output columns for any one stage. | 2,000 | `TooManyColumns` |
| Number of workers for any one stage. | Hard limit is 1,000. Memory-dependent soft limit may be lower. | `TooManyWorkers` |
| Maximum memory occupied by broadcasted tables. | 30% of each [processor memory bundle](concepts.md#memory-usage). | `BroadcastTablesTooLarge` |
+| Maximum relaunches per worker. Initial run is not a relaunch. The worker will be spawned 1 + workerRelaunchLimit times before erroring out. | 2 | `WorkerRelaunchedTooManyTimes` |
+| Maximum relaunches across all workers. | 30 | `TotalRelaunchLimitExceededFault` |
Review Comment:
Updated the maximum relaunched across all workers to 100 but I do agree with @LakshSingla that it should depend on the number of workers of the job eventually.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033651666
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/controller/ControllerQueryKernel.java:
##########
@@ -44,18 +52,20 @@
import org.apache.druid.msq.statistics.ClusterByStatisticsSnapshot;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
Review Comment:
Yes we can do that. Changed it to simple hashMap.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1033634271
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -564,6 +585,34 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
return queryDef;
}
+ /**
+ * Adds the workorders for worker to {@link ControllerImpl#workOrdersToRetry} if the {@link ControllerQueryKernel} determines that there
+ * are work orders which needs reprocessing.
+ */
+ private void addToRetryQueue(ControllerQueryKernel kernel, int worker, MSQFault fault)
+ {
+ List<WorkOrder> retriableWorkOrders = kernel.getWorkInCaseWorkerElgibileForRetryElseThrow(worker, fault);
+ if (retriableWorkOrders.size() != 0) {
+ log.info("Submitting worker[%s] for relaunch because of fault[%s]", worker, fault);
Review Comment:
That information is inside the MSQWorkerTaskLauncher. We donot have it here.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13353:
URL: https://github.com/apache/druid/pull/13353#discussion_r1064623044
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java:
##########
@@ -551,6 +563,15 @@ private QueryDefinition initializeQueryDefAndState(final Closer closer)
id(),
task.getDataSource(),
context,
+ (failedTask, fault) -> {
+ addToKernelManipulationQueue((kernel) -> {
+ if (isDurableStorageEnabled) {
Review Comment:
Added a context flag.
if the user sets faultTolerance=true, I enable durableShuffleStorage as well
if the user sets faultTolerance=true, and durableStorage=false its an error.
if the user sets durableStorage=true, faultTolerance context param remains untouched.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] lgtm-com[bot] commented on pull request #13353: Worker retry for MSQ task
Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on PR #13353:
URL: https://github.com/apache/druid/pull/13353#issuecomment-1342230115
This pull request **introduces 2 alerts** when merging d32397a454d5e8c8d8a6500ea96ba52ec92dda31 into d85fb8cc4ecef574f1c032d09a2f2e1f61f6b1c3 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-fdfb3d907b7edb68b1e6db48499be805de6eca02)
**new alerts:**
* 2 for Unused format argument
---
**Heads-up: LGTM.com's PR analysis will be disabled on the 5th of December**, and LGTM.com will be shut down ⏻ completely on the 16th of December 2022. Please enable GitHub code scanning, which uses the same CodeQL engine :gear: that powers LGTM.com. For more information, please check out [our post on the GitHub blog](https://github.blog/2022-08-15-the-next-step-for-lgtm-com-github-code-scanning/).
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org