Kill Hadoop MR task on kill of Hadoop ingestion task #6828

ankit0811 · 2019-01-09T23:07:28Z

KillTask from overlord UI now makes sure that it terminates the underlying MR job, thus saving unnecessary compute #6803

Run in jobby is now split into 2

submitAndGetHadoopJobId followed by 2. run
submitAndGetHadoopJobId is responsible for submitting the job and returning the jobId as a string, run monitors this job for completion

JobHelper writes this jobId in the path provided by HadoopIndexTask which in turn is provided by the ForkingTaskRunner

HadoopIndexTask reads this path when kill task is clicked to get the jobId and fire the kill command via the yarn api. This is taken care in the stopGracefully method which is called in SingleTaskBackgroundRunner. Have enabled canRestore method to return true for HadoopIndexTask in order for the stopGracefully method to be called

Hadoop*Job classes have been changed to incorporate the changes to jobby

ankit0811 · 2019-01-10T22:27:20Z

@jihoonson @jon-wei can you pls look at this PR for the kill job proposal #6803
Thanks

jon-wei · 2019-01-11T00:42:15Z

indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java

@@ -229,6 +231,9 @@ public TaskStatus call()
                        final File taskDir = taskConfig.getTaskDir(task.getId());
                        final File attemptDir = new File(taskDir, attemptUUID);

+                        task.getContext().put(INDEX_TASK_DIR, taskDir.toString());


I don't think you need these context params, you can get the TaskConfig by calling toolbox.getConfig() in HadoopIndexTask.run().

jon-wei · 2019-01-11T00:45:02Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java

+    public String runTask(String[] args) throws Exception
+    {
+      int res = ToolRunner.run(new JobClient(), args);
+      return res == 0 ? "Sucess" : "Fail";


Sucess -> Success

jon-wei · 2019-01-11T00:45:18Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java

@@ -585,6 +662,7 @@ public String runTask(String[] args) throws Exception
    {
      final String schema = args[0];
      String version = args[1];
+      final String HadoopJobIdFile = args[2];


HadoopJobIdFile -> hadoopJobIdFile

jon-wei · 2019-01-11T00:46:11Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java

+            new Object[]{buildKillJobInput}
+        );
+
+        log.info(String.format(Locale.ENGLISH, "Tried killing job %s , status: %s", jobId, killStatusString));


Should use StringUtils.format instead

jon-wei · 2019-01-11T00:47:41Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java

@@ -218,6 +224,15 @@ public String getClasspathPrefix()
    return classpathPrefix;
  }

+  public String getHadoopJobIdFileName()
+  {
+    String hadoopJobIdFileName = "mapReduceJobId.json";


Let's make this a constant

jon-wei · 2019-01-11T00:48:44Z

indexing-hadoop/src/main/java/org/apache/druid/indexer/JobHelper.java

+          log.info("MR job id is written to jobId file");
+        }
+        catch (IOException e) {
+          log.error("Error wriritng job id to jobId file. Exception %s ", Throwables.getStackTraceAsString(e));


log.error accepts a Throwable directly, the getStackTraceAsString is unnecessary, likewise for runSingleJob

also, wriritng -> writing

jon-wei · 2019-01-11T00:51:02Z

indexing-hadoop/src/main/java/org/apache/druid/indexer/JobHelper.java

+          objectMapper.writeValue(new OutputStreamWriter(
+                  new FileOutputStream(new File(hadoopJobIdFileName)), StandardCharsets.UTF_8),
+              hadoopJobId);
+          log.info("MR job id is written to jobId file");


Suggest putting the full job id path in the log message

jon-wei · 2019-01-11T01:03:11Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java

@@ -412,6 +430,63 @@ private TaskStatus runInternal(TaskToolbox toolbox) throws Exception
    }
  }

+  @Override
+  public boolean canRestore()


Rather than tying the MR job cleanup to the restore functionality, I think it would be better to change the Task.stopGracefully() contract such that it's always called.

Regardless of whether the task would be restored, I think it makes sense to call stopGracefully() in case the task wants to attempt to clean up any open resources, like in this situation.

Also, as is, the MR job termination won't run unless the user has enabled restoreTasksOnRestart which is false by default:

if (taskConfig.isRestoreTasksOnRestart() && task.canRestore()) {

…lying MR job, thus saving unnecessary compute Run in jobby is now split into 2 1. submitAndGetHadoopJobId followed by 2. run submitAndGetHadoopJobId is responsible for submitting the job and returning the jobId as a string, run monitors this job for completion JobHelper writes this jobId in the path provided by HadoopIndexTask which in turn is provided by the ForkingTaskRunner HadoopIndexTask reads this path when kill task is clicked to get hte jobId and fire the kill command via the yarn api. This is taken care in the stopGracefully method which is called in SingleTaskBackgroundRunner. Have enabled `canRestore` method to return `true` for HadoopIndexTask in order for the stopGracefully method to be called Hadoop*Job files have been changed to incorporate the changes to jobby

…ully() `SingleTaskBackgroundRunner` calls stopGracefully in stop() and then checks for canRestore condition to return the status of the task

ankit0811 · 2019-01-11T23:27:10Z

@jon-wei Thanks for reviewing
Have tried to make the necessary changes suggested
And apologies for accidentally using force-push to merge my changes

jihoonson · 2019-01-15T21:16:13Z

@ankit0811 thanks, I'll take a look today.

jihoonson · 2019-01-16T05:58:49Z

core/src/main/java/org/apache/druid/indexer/Jobby.java

+   *
+   * @return A string represtenting the jobId of the actual MR job.
+   * Run method is now divided into two parts. The first one being submitAndGetHadoopJobId which just submits the job and returns the job ID
+   * Run then monitors this job for completion


Looks that the last two lines are descriptions for run method. Please move it to the proper method.

Also, please annotate this method with @Nullable and add a description about when this is null and how null is checked in where.

jihoonson · 2019-01-16T06:02:43Z

indexing-hadoop/src/main/java/org/apache/druid/indexer/JobHelper.java

@@ -349,6 +352,22 @@ public static void ensurePaths(HadoopDruidIndexerConfig config)

  public static boolean runSingleJob(Jobby job, HadoopDruidIndexerConfig config)
  {
+    String hadoopJobId = job.submitAndGetHadoopJobId();
+    ObjectMapper objectMapper = new ObjectMapper();


Please use HadoopDruidIndexerConfig.JSON_MAPPER instead.

jihoonson · 2019-01-16T06:02:47Z

indexing-hadoop/src/main/java/org/apache/druid/indexer/JobHelper.java

@@ -372,7 +391,23 @@ public static boolean runSingleJob(Jobby job, HadoopDruidIndexerConfig config)
  public static boolean runJobs(List<Jobby> jobs, HadoopDruidIndexerConfig config)
  {
    boolean succeeded = true;
+    ObjectMapper objectMapper = new ObjectMapper();


Please use HadoopDruidIndexerConfig.JSON_MAPPER instead.

jihoonson · 2019-01-16T06:03:39Z

indexing-hadoop/src/main/java/org/apache/druid/indexer/DetermineHashedPartitionsJob.java

+      return groupByJob.getJobID().toString();
+    }
+    catch (Exception e) {
+      throw Throwables.propagate(e);


Please throw new RuntimeException(e) instead.

jihoonson · 2019-01-16T06:11:13Z

core/src/main/java/org/apache/druid/indexer/Jobby.java

+   * Run method is now divided into two parts. The first one being submitAndGetHadoopJobId which just submits the job and returns the job ID
+   * Run then monitors this job for completion
+   */
+  default String submitAndGetHadoopJobId()


DeterminePartitionsJob should implement this method and use JobHelper.

Thanks @jihoonson
That was a miss from my side
I see DeterminePartitionJob basically runs 2 job

determine_partitions_groupby

determine_partitions_dimselection

Is it alright to split this into two separate method and then handle their job Id in JobHelper.singleRunJob() bu casting the job

Sorry, I'm not sure what you mean. Would you tell me more details for how to split DeterminPartitionJob?

So currently the run() issues two MR jobs

determine_partitions_groupby

determine_partitions_dimselection

I need these job to return their jobIds first and then check for their status, hence planning to split run to two parts

Basically, case when job is an instance of DetermineJobPartition JobHelper.runSingleJob() will look like this

if (job instanceof DeterminePartitionsJob) { String hadoopJobId = ((DeterminePartitionsJob) job).submitAndGetHadoopJobIdForDeterminePartitionsGroupBy(); ((DeterminePartitionsJob) job).RunDeterminePartitionsGroupBy();

the above will take care of the determine_partitions_groupby job
followed by submitAndGetHadoopJobId and run() (the normal code flow) which will take care of determine_partitions_dimselection job

Let me know if that makes sense
Thanks

Thanks, but I think it's too specific for an implementation of DeterminePartitionsJob which can't handle any other custom Jobby implementations running two or more Hadoop jobs.

I think maybe it's not a good idea to add submitAndGetHadoopJobId to Jobby because a single Jobby can run 0, 1, or more Hadoop jobs. How about removing submitAndGetHadoopJobId from Jobby but adding writeHadoopJobId to JobHelper? Every Jobby running one or more Hadoop jobs should use this method.

Yes
That simplifies things 👍
Thanks. Will make the changes

jihoonson · 2019-01-16T06:26:09Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java

+    }
+    catch (Exception e) {
+      log.info("Exeption while reading json file from path: " + hadoopJobIdFile);
+      log.error(Throwables.getStackTraceAsString(e));


You don't need to print two logs. Please replace them with log.warn(e, "Exeption while reading Hadoop Job ID from: %s", hadoopJobIdFile);.

jihoonson · 2019-01-16T06:27:32Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java

+        ClassLoader loader = HadoopTask.buildClassLoader(getHadoopDependencyCoordinates(),
+            taskConfig.getDefaultHadoopCoordinates());
+
+        Object killMRJobInnerProcessingRunner = getForeignClassloaderObject("org.apache.druid.indexing.common.task.HadoopIndexTask$HadoopKillMRJobIdProcessingRunner",


Format:

Object killMRJobInnerProcessingRunner = getForeignClassloaderObject( "org.apache.druid.indexing.common.task.HadoopIndexTask$HadoopKillMRJobIdProcessingRunner", loader );

jihoonson · 2019-01-16T06:29:39Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/Task.java

@@ -163,8 +163,8 @@ default int getPriority()
  boolean canRestore();

  /**
-   * Asks a task to arrange for its "run" method to exit promptly. This method will only be called if
-   * {@link #canRestore()} returns true. Tasks that take too long to stop gracefully will be terminated with
+   * Asks a task to arrange for its "run" method to exit promptly. This method will be called, whether


Let's just remove This method will be called, whether {@link #canRestore()} returns true/false.

jihoonson · 2019-01-16T06:31:12Z

...ing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java

-        log.info("Starting graceful shutdown of task[%s].", task.getId());
+      // stopGracefully for resource cleaning, independent of the fact whether the task is restorable or not
+      // Attempt graceful shutdown.
+      graceful = true;


graceful is always true. Please remove it and set true for metric like below.

final ServiceMetricEvent.Builder metricBuilder = ServiceMetricEvent .builder() .setDimension("task", task.getId()) .setDimension("dataSource", task.getDataSource()) .setDimension("graceful", "true") // for backward compatibility .setDimension("error", String.valueOf(error));

jihoonson · 2019-01-16T06:40:32Z

indexing-service/pom.xml

@@ -90,6 +90,16 @@
            <type>test-jar</type>
            <scope>test</scope>
        </dependency>
+      <dependency>
+        <groupId>org.apache.hadoop</groupId>
+        <artifactId>hadoop-common</artifactId>


We really shouldn't add hadoop dependencies to druid core because 1) it's unnecessary if we don't use hadoop and 2) it might cause some errors because of the version mismatch when we use a different version of Hadoop.

Please move them to indexing-hadoop. The version must not be specified and the scope should be provided.

1. Formatting 2. Removing `submitAndGetHadoopJobId` from `Jobby` and calling writeJobIdToFile in the job itself

1. POM change. Moving hadoop dependency to indexing-hadoop

jon-wei

Generally LGTM, but I think this needs one more major change:

Since stopGracefully is always called now, we should adjust the tasks that previously implemented graceful shutdown so that they continue to only do graceful shutdown when restore is enabled.

I also left some minor comments re: log messages

jon-wei · 2019-01-17T20:58:39Z

indexing-hadoop/src/main/java/org/apache/druid/indexer/JobHelper.java

+        log.warn(e, "Error writing job id [%s] to the file [%s]", hadoopJobId, hadoopJobIdFileName);
+      }
+    } else {
+      log.info("Either job Id or File Name is null for the submitted job. Skipping writing the file [%s]", hadoopJobIdFileName);


nit: suggest "job id" and "filename" without capitalization

jon-wei · 2019-01-17T20:59:24Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractTask.java

@@ -153,11 +153,14 @@ public boolean canRestore()
    return false;
  }

+  /**
+   * Should be called independent of canRestore so that Resource cleaning can be achieved.


Resource -> resource

jon-wei · 2019-01-17T21:07:00Z

indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java

+      }
+    }
+    catch (Exception e) {
+      log.warn(e, "Exeption while reading Hadoop Job ID from: %s", hadoopJobIdFile);


Exception -> exception

jon-wei · 2019-01-17T21:08:32Z

...ing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java

-        // Attempt graceful shutdown.
-        graceful = true;
-        log.info("Starting graceful shutdown of task[%s].", task.getId());
+      // stopGracefully for resource cleaning, independent of the fact whether the task is restorable or not


nit: can delete the "independent of the fact" portion and the later "Attempt graceful shutdown" comment

jon-wei · 2019-01-17T21:09:05Z

...ing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java

@@ -223,7 +221,7 @@ public void stop()
          .builder()
          .setDimension("task", task.getId())
          .setDimension("dataSource", task.getDataSource())
-          .setDimension("graceful", String.valueOf(graceful))
+          .setDimension("graceful", "true") // for backword compatibility


backword -> backward

ankit0811 · 2019-01-17T23:24:58Z

Generally LGTM, but I think this needs one more major change:

Since stopGracefully is always called now, we should adjust the tasks that previously implemented graceful shutdown so that they continue to only do graceful shutdown when restore is enabled

Just wanted to be sure,
The only case which is not handled here is when isRestoreOnRestart = False and canRestore = True ?
As realtime tasks have canRestore set to True and Index task have canRestore set to False always

jon-wei · 2019-01-17T23:45:49Z

@ankit0811

I would change AbstractTask so that stopGracefully() no longer throws an exception but just does nothing:

  public void stopGracefully()
  {
    // Should not be called when canRestore = false.
    throw new UnsupportedOperationException("Cannot stop gracefully");
  }

For the tasks that did have a stopGracefully() implementation (AppenderatorDriverRealtimeIndexTask, RealtimeIndexTask, SeekableStreamIndexTask), this used to be called only when taskConfig. isRestoreOnRestart is true, so that check should be moved now into the stopGracefully() implementations there.

You could change the stopGracefully() method to have the task runner pass in the TaskConfig object for access to restoreTasksOnRestart

jihoonson

@jon-wei thanks for catching it!

jihoonson · 2019-01-18T00:06:02Z

indexing-hadoop/pom.xml

+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-mapreduce-client-core</artifactId>


Please add provided scope here too.

1. stopGracefully now accepts TaskConfig as a param Handling isRestoreOnRestart in stopGracefully for `AppenderatorDriverRealtimeIndexTask, RealtimeIndexTask, SeekableStreamIndexTask` Changing tests to make TaskConfig param isRestoreOnRestart to true

ankit0811 · 2019-01-18T20:51:29Z

@jihoonson @jon-wei
Thanks for the reviews
As per your suggestions, have made the necessary changes to stopGracefully() method
Can u pls review the same
Thanks :)

jon-wei

LGTM

ankit0811 changed the title ~~Kill Hadoop MR task on kill of ingestion task and resume ability for Hadoop ingestion tasks~~ Kill Hadoop MR task on kill of Hadoop ingestion task Jan 10, 2019

jon-wei reviewed Jan 11, 2019

View reviewed changes

ankit0811 added 2 commits January 10, 2019 23:14

Addressing PR comments

9b79c3a

ankit0811 force-pushed the feature-killMRJob branch from 97cf628 to 9b79c3a Compare January 11, 2019 07:18

ankit0811 added 2 commits January 11, 2019 00:11

Addressing PR comments - Fix taskDir

ca46739

Addressing PR comments - For changing the contract of Task.stopGracef…

8152f87

…ully() `SingleTaskBackgroundRunner` calls stopGracefully in stop() and then checks for canRestore condition to return the status of the task

gianm assigned jon-wei Jan 15, 2019

jihoonson reviewed Jan 16, 2019

View reviewed changes

ankit0811 added 2 commits January 16, 2019 15:55

Addressing PR comments

1f598e0

1. Formatting 2. Removing `submitAndGetHadoopJobId` from `Jobby` and calling writeJobIdToFile in the job itself

Addressing PR comments

fca6e1b

1. POM change. Moving hadoop dependency to indexing-hadoop

jon-wei reviewed Jan 17, 2019

View reviewed changes

jihoonson reviewed Jan 18, 2019

View reviewed changes

Addressing PR comments

d9d93e3

1. stopGracefully now accepts TaskConfig as a param Handling isRestoreOnRestart in stopGracefully for `AppenderatorDriverRealtimeIndexTask, RealtimeIndexTask, SeekableStreamIndexTask` Changing tests to make TaskConfig param isRestoreOnRestart to true

jon-wei approved these changes Jan 22, 2019

View reviewed changes

jihoonson approved these changes Jan 25, 2019

View reviewed changes

jihoonson merged commit 8492d94 into apache:master Jan 25, 2019

jon-wei mentioned this pull request Jan 31, 2019

NoClassDefFoundError when using druid-hdfs-storage #6967

Closed

jon-wei added this to the 0.14.0 milestone Feb 20, 2019

jon-wei mentioned this pull request Feb 22, 2019

0.14.0-incubating release notes #7126

Closed

a2l007 mentioned this pull request Jun 25, 2019

Killing hadoop ingestion task does not kill spawned Hadoop MR task #7962

Closed

This was referenced Aug 19, 2019

Missing hadoop client dependency #8338

Closed

Don't inadvertently exclude hadoop-client-core dependency #8339

Merged

jihoonson mentioned this pull request Oct 22, 2019

Fix graceful shutdown for tasks #8718

Open

Kill Hadoop MR task on kill of Hadoop ingestion task #6828

Kill Hadoop MR task on kill of Hadoop ingestion task #6828

Conversation

ankit0811 commented Jan 9, 2019

ankit0811 commented Jan 10, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ankit0811 commented Jan 11, 2019

jihoonson commented Jan 15, 2019

Choose a reason for hiding this comment

jihoonson Jan 16, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jihoonson Jan 16, 2019 • edited Loading

Choose a reason for hiding this comment

ankit0811 Jan 16, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jihoonson Jan 16, 2019 • edited Loading

Choose a reason for hiding this comment

jon-wei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ankit0811 commented Jan 17, 2019

jon-wei commented Jan 17, 2019

jihoonson left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ankit0811 commented Jan 18, 2019

jon-wei left a comment

Choose a reason for hiding this comment

jihoonson Jan 16, 2019 •

edited

Loading

jihoonson Jan 16, 2019 •

edited

Loading

ankit0811 Jan 16, 2019 •

edited

Loading

jihoonson Jan 16, 2019 •

edited

Loading