You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by su...@apache.org on 2019/07/19 21:12:09 UTC
[incubator-gobblin] branch master updated: [GOBBLIN-816] Implement
a workaround to abort Helix TaskDriver#getWorkflows() after a timeout[]
This is an automated email from the ASF dual-hosted git repository.
suvasude pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new 51edd5d [GOBBLIN-816] Implement a workaround to abort Helix TaskDriver#getWorkflows() after a timeout[]
51edd5d is described below
commit 51edd5d82f4246db24b05de0f9e45f86bc7a7af2
Author: sv2000 <su...@gmail.com>
AuthorDate: Fri Jul 19 14:12:00 2019 -0700
[GOBBLIN-816] Implement a workaround to abort Helix TaskDriver#getWorkflows() after a timeout[]
Closes #2680 from sv2000/getWorkflowsWorkaround
---
.../cluster/GobblinClusterConfigurationKeys.java | 3 +++
.../gobblin/cluster/GobblinHelixJobScheduler.java | 28 +++++++++++++++++++++-
2 files changed, 30 insertions(+), 1 deletion(-)
diff --git a/gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinClusterConfigurationKeys.java b/gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinClusterConfigurationKeys.java
index d1267c5..4881a99 100644
--- a/gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinClusterConfigurationKeys.java
+++ b/gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinClusterConfigurationKeys.java
@@ -151,6 +151,9 @@ public class GobblinClusterConfigurationKeys {
public static final String HELIX_WORKFLOW_DELETE_TIMEOUT_SECONDS = GOBBLIN_CLUSTER_PREFIX + "workflowDeleteTimeoutSeconds";
public static final long DEFAULT_HELIX_WORKFLOW_DELETE_TIMEOUT_SECONDS = 300;
+ public static final String HELIX_WORKFLOW_LISTING_TIMEOUT_SECONDS = GOBBLIN_CLUSTER_PREFIX + "workflowListingTimeoutSeconds";
+ public static final long DEFAULT_HELIX_WORKFLOW_LISTING_TIMEOUT_SECONDS = 300;
+
public static final String CLEAN_ALL_DIST_JOBS = GOBBLIN_CLUSTER_PREFIX + "bootup.clean.dist.jobs";
public static final boolean DEFAULT_CLEAN_ALL_DIST_JOBS = false;
diff --git a/gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobScheduler.java b/gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobScheduler.java
index 3054d2e..6349471 100644
--- a/gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobScheduler.java
+++ b/gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobScheduler.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
+import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -36,6 +37,11 @@ import org.apache.helix.task.TaskDriver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.github.rholder.retry.AttemptTimeLimiters;
+import com.github.rholder.retry.RetryException;
+import com.github.rholder.retry.Retryer;
+import com.github.rholder.retry.RetryerBuilder;
+import com.github.rholder.retry.StopStrategies;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Maps;
import com.google.common.eventbus.EventBus;
@@ -98,6 +104,7 @@ public class GobblinHelixJobScheduler extends JobScheduler implements StandardMe
final GobblinHelixPlanningJobLauncherMetrics planningJobLauncherMetrics;
final HelixJobsMapping jobsMapping;
final Striped<Lock> locks = Striped.lazyWeakLock(256);
+ private final long helixWorkflowListingTimeoutMillis;
private boolean startServicesCompleted;
private final long helixJobStopTimeoutMillis;
@@ -150,6 +157,10 @@ public class GobblinHelixJobScheduler extends JobScheduler implements StandardMe
this.helixJobStopTimeoutMillis = ConfigUtils.getLong(jobConfig, GobblinClusterConfigurationKeys.HELIX_JOB_STOP_TIMEOUT_SECONDS,
GobblinClusterConfigurationKeys.DEFAULT_HELIX_JOB_STOP_TIMEOUT_SECONDS) * 1000;
+
+ this.helixWorkflowListingTimeoutMillis = ConfigUtils.getLong(jobConfig, GobblinClusterConfigurationKeys.HELIX_WORKFLOW_LISTING_TIMEOUT_SECONDS,
+ GobblinClusterConfigurationKeys.DEFAULT_HELIX_WORKFLOW_LISTING_TIMEOUT_SECONDS) * 1000;
+
}
@Override
@@ -360,8 +371,22 @@ public class GobblinHelixJobScheduler extends JobScheduler implements StandardMe
if (PropertiesUtils.getPropAsBoolean(jobConfig, GobblinClusterConfigurationKeys.CANCEL_RUNNING_JOB_ON_DELETE,
GobblinClusterConfigurationKeys.DEFAULT_CANCEL_RUNNING_JOB_ON_DELETE)) {
LOGGER.info("Cancelling workflow: {}", deleteJobArrival.getJobName());
- Map<String, String> jobNameToWorkflowIdMap = HelixUtils.getWorkflowIdsFromJobNames(this.jobHelixManager,
+
+ //Workaround for preventing indefinite hangs observed in TaskDriver.getWorkflows() call.
+ Callable<Map<String, String>> workflowsCallable = () -> HelixUtils.getWorkflowIdsFromJobNames(this.jobHelixManager,
Collections.singletonList(deleteJobArrival.getJobName()));
+ Retryer<Map<String, String>> retryer = RetryerBuilder.<Map<String, String>>newBuilder()
+ .retryIfException()
+ .withStopStrategy(StopStrategies.stopAfterAttempt(1))
+ .withAttemptTimeLimiter(AttemptTimeLimiters.fixedTimeLimit(this.helixWorkflowListingTimeoutMillis, TimeUnit.MILLISECONDS)).build();
+ Map<String, String> jobNameToWorkflowIdMap;
+ try {
+ jobNameToWorkflowIdMap = retryer.call(workflowsCallable);
+ } catch (ExecutionException | RetryException e) {
+ LOGGER.error("Exception encountered when getting workflows from Helix; likely a Helix/Zk issue.", e);
+ return;
+ }
+
if (jobNameToWorkflowIdMap.containsKey(deleteJobArrival.getJobName())) {
String workflowId = jobNameToWorkflowIdMap.get(deleteJobArrival.getJobName());
TaskDriver taskDriver = new TaskDriver(this.jobHelixManager);
@@ -374,6 +399,7 @@ public class GobblinHelixJobScheduler extends JobScheduler implements StandardMe
}
}
}
+
/**
* This class is responsible for running non-scheduled jobs.
*/