Temporal POC

gobblin-cluster/build.gradle

@@ -47,6 +47,7 @@ dependencies {
   compile externalDependency.hadoopCommon
   compile externalDependency.avroMapredH2
   compile externalDependency.findBugsAnnotations
+  compile externalDependency."temporal-sdk"
   compile (externalDependency.helix) {
     exclude group: 'io.dropwizard.metrics', module: 'metrics-core'
   }

gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinClusterConfigurationKeys.java

@@ -234,5 +234,7 @@ public class GobblinClusterConfigurationKeys {
 
   public static final String HELIX_JOB_SCHEDULING_THROTTLE_TIMEOUT_SECONDS_KEY = "helix.job.scheduling.throttle.timeout.seconds";
   public static final long DEFAULT_HELIX_JOB_SCHEDULING_THROTTLE_TIMEOUT_SECONDS_KEY = Duration.ofMinutes(40).getSeconds();;
-
+  public static final String TEMPORAL_WORKER_SIZE = "temporal.worker.size";
+  public static final String TEMPORAL_ENABLED = "temporal.enabled";
+  public static final Boolean DEFAULT_TEMPORAL_ENABLED = false;
 }

gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinClusterManager.java

@@ -79,7 +79,8 @@
  * The central cluster manager for Gobblin Clusters.
  *
  * <p>
- *   This class runs the {@link GobblinHelixJobScheduler} for scheduling and running Gobblin jobs.
+ *   This class runs the {@link GobblinHelixJobScheduler} for scheduling
+ *   and running Gobblin jobs.
  *   This class serves as the Helix controller and it uses a {@link HelixManager} to work with Helix.
  * </p>
  *

gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobLauncherListener.java

@@ -33,12 +33,12 @@
  * The {@link GobblinHelixJobLauncherMetrics} will always be passed in because
  * it will be be updated accordingly.
  */
-class GobblinHelixJobLauncherListener extends AbstractJobListener {
+public class GobblinHelixJobLauncherListener extends AbstractJobListener {
 
   private final GobblinHelixJobLauncherMetrics jobLauncherMetrics;
   private static final String JOB_START_TIME = "jobStartTime";
 
-  GobblinHelixJobLauncherListener(GobblinHelixJobLauncherMetrics jobLauncherMetrics) {
+  public GobblinHelixJobLauncherListener(GobblinHelixJobLauncherMetrics jobLauncherMetrics) {
     this.jobLauncherMetrics = jobLauncherMetrics;
   }
 

gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobLauncherMetrics.java

@@ -28,7 +28,7 @@
 /**
  * Metrics that relates to jobs launched by {@link GobblinHelixJobLauncher}.
  */
-class GobblinHelixJobLauncherMetrics extends StandardMetricsBridge.StandardMetrics {
+public class GobblinHelixJobLauncherMetrics extends StandardMetricsBridge.StandardMetrics {
     private final String metricsName;
 
     final ContextAwareMeter numJobsLaunched;

gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinHelixJobSchedulerMetrics.java

@@ -33,7 +33,7 @@
 import org.apache.gobblin.runtime.api.JobExecutionLauncher;
 
 
-class GobblinHelixJobSchedulerMetrics extends StandardMetricsBridge.StandardMetrics {
+public class GobblinHelixJobSchedulerMetrics extends StandardMetricsBridge.StandardMetrics {
   public static final String SCHEDULE_CANCELLATION_START = "scheduleCancellationStart";
   public static final String SCHEDULE_CANCELLATION_END = "scheduleCancellationEnd";
   public static final String TIMER_BEFORE_JOB_SCHEDULING = "timerBeforeJobScheduling";
@@ -74,21 +74,21 @@ public GobblinHelixJobSchedulerMetrics (final ExecutorService jobExecutor, final
     this.contextAwareMetrics.add(metricContext.newContextAwareGauge(JobExecutionLauncher.StandardMetrics.EXECUTOR_QUEUE_SIZE, ()->this.threadPoolExecutor.getQueue().size()));
   }
 
-  void updateTimeBeforeJobScheduling (Properties jobProps) {
+  public void updateTimeBeforeJobScheduling (Properties jobProps) {
     long jobCreationTime = Long.parseLong(jobProps.getProperty(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, "0"));
     Instrumented.updateTimer(Optional.of(timeBeforeJobScheduling),
         System.currentTimeMillis() - jobCreationTime,
         TimeUnit.MILLISECONDS);
   }
 
-  void updateTimeBeforeJobLaunching (Properties jobProps) {
+  public void updateTimeBeforeJobLaunching (Properties jobProps) {
     long jobCreationTime = Long.parseLong(jobProps.getProperty(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, "0"));
     Instrumented.updateTimer(Optional.of(timeBeforeJobLaunching),
         System.currentTimeMillis() - jobCreationTime,
         TimeUnit.MILLISECONDS);
   }
 
-  void updateTimeBetweenJobSchedulingAndJobLaunching (long scheduledTime, long launchingTime) {
+  public void updateTimeBetweenJobSchedulingAndJobLaunching (long scheduledTime, long launchingTime) {
     Instrumented.updateTimer(Optional.of(timeBetwenJobSchedulingAndLaunching),
         launchingTime - scheduledTime,
         TimeUnit.MILLISECONDS);

gobblin-cluster/src/main/java/org/apache/gobblin/cluster/GobblinJobLauncher.java

@@ -0,0 +1,279 @@
+/*
+ * 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.gobblin.cluster;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.Optional;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigValueFactory;
+
+import javax.annotation.Nullable;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.annotation.Alpha;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.StateStore;
+import org.apache.gobblin.metrics.Tag;
+import org.apache.gobblin.metrics.event.CountEventBuilder;
+import org.apache.gobblin.metrics.event.JobEvent;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.rest.LauncherTypeEnum;
+import org.apache.gobblin.runtime.AbstractJobLauncher;
+import org.apache.gobblin.runtime.JobException;
+import org.apache.gobblin.runtime.JobLauncher;
+import org.apache.gobblin.runtime.JobState;
+import org.apache.gobblin.runtime.TaskStateCollectorService;
+import org.apache.gobblin.runtime.listeners.JobListener;
+import org.apache.gobblin.runtime.util.StateStores;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaSource;
+import org.apache.gobblin.source.workunit.MultiWorkUnit;
+import org.apache.gobblin.source.workunit.WorkUnit;
+import org.apache.gobblin.util.ConfigUtils;
+import org.apache.gobblin.util.Id;
+import org.apache.gobblin.util.JobLauncherUtils;
+import org.apache.gobblin.util.ParallelRunner;
+import org.apache.gobblin.util.PropertiesUtils;
+import org.apache.gobblin.util.SerializationUtils;
+
+/**
+ * An implementation of {@link JobLauncher} that launches a Gobblin job using the Temporal task framework.
+ *
+ * <p>
+ *   Each {@link WorkUnit} of the job is persisted to the {@link FileSystem} of choice and the path to the file
+ *   storing the serialized {@link WorkUnit} is passed to the Temporal task running the {@link WorkUnit} as a
+ *   user-defined property {@link GobblinClusterConfigurationKeys#WORK_UNIT_FILE_PATH}. Upon startup, the gobblin
+ *   task reads the property for the file path and de-serializes the {@link WorkUnit} from the file.
+ * </p>
+ */
+@Alpha
+@Slf4j
+public class GobblinJobLauncher extends AbstractJobLauncher {
+  private static final Logger LOGGER = LoggerFactory.getLogger(GobblinJobLauncher.class);
+  protected static final String WORK_UNIT_FILE_EXTENSION = ".wu";
+  protected final FileSystem fs;
+  protected final Path appWorkDir;
+  protected final Path inputWorkUnitDir;
+  protected final Path outputTaskStateDir;
+
+  // Number of ParallelRunner threads to be used for state serialization/deserialization
+  protected final int stateSerDeRunnerThreads;
+
+  protected final TaskStateCollectorService taskStateCollectorService;
+  protected final ConcurrentHashMap<String, Boolean> runningMap;
+  @Getter
+  protected final StateStores stateStores;
+  protected JobListener jobListener;
+  protected volatile boolean jobSubmitted = false;
+
+
+  public GobblinJobLauncher(Properties jobProps, Path appWorkDir,
+      List<? extends Tag<?>> metadataTags, ConcurrentHashMap<String, Boolean> runningMap)
+      throws Exception {
+    super(jobProps, HelixUtils.initBaseEventTags(jobProps, metadataTags));
+    LOGGER.debug("GobblinJobLauncher: jobProps {}, appWorkDir {}", jobProps, appWorkDir);
+    this.runningMap = runningMap;
+    this.appWorkDir = appWorkDir;
+    this.inputWorkUnitDir = new Path(appWorkDir, GobblinClusterConfigurationKeys.INPUT_WORK_UNIT_DIR_NAME);
+    this.outputTaskStateDir = new Path(this.appWorkDir,
+        GobblinClusterConfigurationKeys.OUTPUT_TASK_STATE_DIR_NAME + Path.SEPARATOR + this.jobContext.getJobId());
+
+    this.jobContext.getJobState().setJobLauncherType(LauncherTypeEnum.CLUSTER);
+
+    this.stateSerDeRunnerThreads = Integer.parseInt(jobProps.getProperty(ParallelRunner.PARALLEL_RUNNER_THREADS_KEY,
+        Integer.toString(ParallelRunner.DEFAULT_PARALLEL_RUNNER_THREADS)));
+
+    Config stateStoreJobConfig = ConfigUtils.propertiesToConfig(jobProps)
+        .withValue(ConfigurationKeys.STATE_STORE_FS_URI_KEY, ConfigValueFactory.fromAnyRef(
+            new URI(appWorkDir.toUri().getScheme(), null, appWorkDir.toUri().getHost(), appWorkDir.toUri().getPort(),
+                "/", null, null).toString()));
+
+    this.stateStores =
+        new StateStores(stateStoreJobConfig, appWorkDir, GobblinClusterConfigurationKeys.OUTPUT_TASK_STATE_DIR_NAME,
+            appWorkDir, GobblinClusterConfigurationKeys.INPUT_WORK_UNIT_DIR_NAME, appWorkDir,
+            GobblinClusterConfigurationKeys.JOB_STATE_DIR_NAME);
+
+    URI fsUri = URI.create(jobProps.getProperty(ConfigurationKeys.FS_URI_KEY, ConfigurationKeys.LOCAL_FS_URI));
+    this.fs = FileSystem.get(fsUri, new Configuration());
+
+    this.taskStateCollectorService =
+        new TaskStateCollectorService(jobProps, this.jobContext.getJobState(), this.eventBus, this.eventSubmitter,
+            this.stateStores.getTaskStateStore(), this.outputTaskStateDir, this.getIssueRepository());
+  }
+
+  @Override
+  public void close() throws IOException {
+    try {
+      executeCancellation();
+    } finally {
+      super.close();
+    }
+  }
+
+  public String getJobId() {
+    return this.jobContext.getJobId();
+  }
+
+  @Override
+  protected void runWorkUnits(List<WorkUnit> workUnits) throws Exception {
+    try {
+      CountEventBuilder countEventBuilder = new CountEventBuilder(JobEvent.WORK_UNITS_CREATED, workUnits.size());
+      this.eventSubmitter.submit(countEventBuilder);
+      LOGGER.info("Emitting WorkUnitsCreated Count: " + countEventBuilder.getCount());
+
+      long workUnitStartTime = System.currentTimeMillis();
+      workUnits.forEach((k) -> k.setProp(ConfigurationKeys.WORK_UNIT_CREATION_TIME_IN_MILLIS, workUnitStartTime));
+
+      // Start the output TaskState collector service
+      this.taskStateCollectorService.startAsync().awaitRunning();
+
+      TimingEvent jobSubmissionTimer =
+          this.eventSubmitter.getTimingEvent(TimingEvent.RunJobTimings.HELIX_JOB_SUBMISSION);
+
+      synchronized (this.cancellationRequest) {
+        if (!this.cancellationRequested) {
+          submitJob(workUnits);
+          jobSubmissionTimer.stop();
+          LOGGER.info(String.format("Submitted job %s", this.jobContext.getJobId()));
+          this.jobSubmitted = true;
+        } else {
+          LOGGER.warn("Job {} not submitted as it was requested to be cancelled.", this.jobContext.getJobId());
+        }
+      }
+
+      TimingEvent jobRunTimer = this.eventSubmitter.getTimingEvent(TimingEvent.RunJobTimings.HELIX_JOB_RUN);
+      waitJob();
+      jobRunTimer.stop();
+      LOGGER.info(String.format("Job %s completed", this.jobContext.getJobId()));
+    } finally {
+      // The last iteration of output TaskState collecting will run when the collector service gets stopped
+      this.taskStateCollectorService.stopAsync().awaitTerminated();
+      cleanupWorkingDirectory();
+    }
+  }
+
+  protected void submitJob(List<WorkUnit> workUnits) throws Exception {
+  }
+
+  protected void waitJob() throws InterruptedException {
+  }
+
+  @Override
+  protected void executeCancellation() {
+  }
+
+  public void launchJob(@Nullable JobListener jobListener) throws JobException {
+    this.jobListener = jobListener;
+    LOGGER.info("Launching Job");
+    boolean isLaunched = false;
+    this.runningMap.putIfAbsent(this.jobContext.getJobName(), false);
+
+    Throwable errorInJobLaunching = null;
+    try {
+      if (this.runningMap.replace(this.jobContext.getJobName(), false, true)) {
+        LOGGER.info("Job {} will be executed, add into running map.", this.jobContext.getJobId());
+        isLaunched = true;
+        launchJobImpl(jobListener);
+      } else {
+        LOGGER.warn("Job {} will not be executed because other jobs are still running.", this.jobContext.getJobId());
+      }
+
+      // TODO: Better error handling. The current impl swallows exceptions for jobs that were started by this method call.
+      // One potential way to improve the error handling is to make this error swallowing conifgurable
+    } catch (Throwable t) {
+      errorInJobLaunching = t;
+      if (isLaunched) {
+        // Attempts to cancel workflow if an error occurs during launch
+        cancelJob(jobListener);
+      }
+    } finally {
+      if (isLaunched) {
+        if (this.runningMap.replace(this.jobContext.getJobName(), true, false)) {
+          LOGGER.info("Job {} is done, remove from running map.", this.jobContext.getJobId());
+        } else {
+          throw errorInJobLaunching == null ? new IllegalStateException(
+              "A launched job should have running state equal to true in the running map.")
+              : new RuntimeException("Failure in launching job:", errorInJobLaunching);
+        }
+      }
+    }
+  }
+
+  /**
+   * This method looks silly at first glance but exists for a reason.
+   *
+   * The method {@link GobblinJobLauncher#launchJob(JobListener)} contains boiler plate for handling exceptions and
+   * mutating the runningMap to communicate state back to the {@link GobblinJobScheduler}. The boiler plate swallows
+   * exceptions when launching the job because many use cases require that 1 job failure should not affect other jobs by causing the
+   * entire process to fail through an uncaught exception.
+   *
+   * This method is useful for unit testing edge cases where we expect {@link JobException}s during the underlying launch operation.
+   * It would be nice to not swallow exceptions, but the implications of doing that will require careful refactoring since
+   * the class {@link GobblinJobLauncher} and {@link GobblinJobScheduler} are shared for 2 quite different cases
+   * between GaaS and streaming. GaaS typically requiring many short lifetime workflows (where a failure is tolerated) and
+   * streaming requiring a small number of long running workflows (where failure to submit is unexpected and is not
+   * tolerated)
+   *
+   * @throws JobException
+   */
+  @VisibleForTesting
+  void launchJobImpl(@Nullable JobListener jobListener) throws JobException {
+    super.launchJob(jobListener);
+  }
+
+  /**
+   * Delete persisted {@link WorkUnit}s and {@link JobState} upon job completion.
+   */
+  protected void cleanupWorkingDirectory() throws IOException {
+    LOGGER.info("Deleting persisted work units for job " + this.jobContext.getJobId());
+    stateStores.getWuStateStore().delete(this.jobContext.getJobId());
+
+    // delete the directory that stores the task state files
+    stateStores.getTaskStateStore().delete(outputTaskStateDir.getName());
+
+    LOGGER.info("Deleting job state file for job " + this.jobContext.getJobId());
+
+    if (this.stateStores.haveJobStateStore()) {
+      this.stateStores.getJobStateStore().delete(this.jobContext.getJobId());
+    } else {
+      Path jobStateFilePath =
+          GobblinClusterUtils.getJobStateFilePath(false, this.appWorkDir, this.jobContext.getJobId());
+      this.fs.delete(jobStateFilePath, false);
+    }
+  }
+}
+