cryptoe commented on code in PR #17816:
URL: https://github.com/apache/druid/pull/17816#discussion_r2007066493
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java:
##########
@@ -509,30 +507,25 @@ public Response getWorkerConfigHistory(
@ResourceFilters(StateResourceFilter.class)
public Response doAction(final TaskActionHolder holder)
{
+ final Task task = holder.getTask();
return asLeaderWith(
- taskMaster.getTaskActionClient(holder.getTask()),
- new Function<>()
- {
- @Override
- public Response apply(TaskActionClient taskActionClient)
- {
- final Map<String, Object> retMap;
-
- // It would be great to verify that this worker is actually
supposed to be running the task before
- // actually doing the action. Some ideas for how that could be
done would be using some sort of attempt_id
- // or token that gets passed around.
-
- try {
- final Object ret = taskActionClient.submit(holder.getAction());
- retMap = new HashMap<>();
- retMap.put("result", ret);
- }
- catch (Exception e) {
- log.warn(e, "Failed to perform task action");
- return Response.serverError().entity(ImmutableMap.of("error",
e.getMessage())).build();
- }
-
- return Response.ok().entity(retMap).build();
+ taskMaster.getTaskActionClient(task),
+ taskActionClient -> {
+ try {
+ final Object result = taskActionClient.submit(holder.getAction());
+ return Response.ok().entity(Map.of("result", result)).build();
+ }
+ catch (DruidException e) {
+ log.noStackTrace().warn(
Review Comment:
Logging the stack-trace would be beneficial no ?
##########
server/src/main/java/org/apache/druid/indexing/overlord/SegmentPublishResult.java:
##########
@@ -34,22 +34,14 @@
import java.util.Set;
/**
- * Result of an operation that attempts to publish segments. Indicates the set
of segments actually published
- * and whether or not the transaction was a success.
- *
- * If "success" is false then the segments set will be empty.
- *
- * It's possible for the segments set to be empty even if "success" is true,
since the segments set only
- * includes segments actually published as part of the transaction. The
requested segments could have been
- * published by a different transaction (e.g. in the case of replica sets) and
this one would still succeed.
+ * Result of a segment publish operation.
*/
public class SegmentPublishResult
{
+ private final boolean canRetry;
private final Set<DataSegment> segments;
private final boolean success;
Review Comment:
Super nit cause you love beautiful code.
Should line 41 go to line 44 after `private final boolean success;` the
constructor args and the class fields are in the same order.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentPublishAction.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.indexing.common.actions;
+
+import org.apache.druid.indexing.common.task.IndexTaskUtils;
+import org.apache.druid.indexing.common.task.Task;
+import org.apache.druid.indexing.overlord.SegmentPublishResult;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+
+/**
+ * Task action to publish segments. Contains common code used by insert,
replace
+ * and append actions.
+ */
+public abstract class SegmentPublishAction implements
TaskAction<SegmentPublishResult>
+{
+ private static final int QUIET_RETRIES = 3;
+ private static final int MAX_RETRIES = 5;
+
+ @Override
+ public SegmentPublishResult perform(Task task, TaskActionToolbox toolbox)
+ {
+ int attemptCount = 0;
+ final String taskId = task.getId();
+
+ // Retry until success or until max retries are exhausted
+ SegmentPublishResult result = tryPublishSegments(task, toolbox);
+ while (!result.isSuccess() && result.canRetry() && attemptCount++ <
MAX_RETRIES) {
+ awaitNextRetry(taskId, result, attemptCount);
+ result = tryPublishSegments(task, toolbox);
+ }
+
+ IndexTaskUtils.emitSegmentPublishMetrics(result, task, toolbox);
+ return result;
+ }
+
+ /**
+ * Sleeps until the next attempt.
+ */
+ private static void awaitNextRetry(String taskId, SegmentPublishResult
lastResult, int attemptCount)
+ {
+ try {
+ RetryUtils.awaitNextRetry(
Review Comment:
Do we want to adjust the sleep duration for the retry.
One second seems a lot no since it will hold the jetty thread for that long ?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
