Kill Hadoop MR task on kill of Hadoop ingestion task

indexing-hadoop/pom.xml

@@ -130,6 +130,21 @@
             <version>${hadoop.compile.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-mapreduce-client-core</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>javax.servlet</groupId>
+                    <artifactId>servlet-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>org.apache.druid</groupId>
             <artifactId>druid-server</artifactId>

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

@@ -75,6 +75,7 @@ public class DetermineHashedPartitionsJob implements Jobby
   private final HadoopDruidIndexerConfig config;
   private String failureCause;
   private Job groupByJob;
+  private long startTime;
 
   public DetermineHashedPartitionsJob(
       HadoopDruidIndexerConfig config
@@ -91,7 +92,7 @@ public boolean run()
        * Group by (timestamp, dimensions) so we can correctly count dimension values as they would appear
        * in the final segment.
        */
-      final long startTime = System.currentTimeMillis();
+      startTime = System.currentTimeMillis();
       groupByJob = Job.getInstance(
           new Configuration(),
           StringUtils.format("%s-determine_partitions_hashed-%s", config.getDataSource(), config.getIntervals())
@@ -125,6 +126,11 @@ public boolean run()
       groupByJob.submit();
       log.info("Job %s submitted, status available at: %s", groupByJob.getJobName(), groupByJob.getTrackingURL());
 
+      // Store the jobId in the file
+      if (groupByJob.getJobID() != null) {
+        JobHelper.writeJobIdToFile(config.getHadoopJobIdFileName(), groupByJob.getJobID().toString());
+      }
+
       if (!groupByJob.waitForCompletion(true)) {
         log.error("Job failed: %s", groupByJob.getJobID());
         failureCause = Utils.getFailureMessage(groupByJob, config.JSON_MAPPER);

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

@@ -161,6 +161,12 @@ public boolean run()
         groupByJob.submit();
         log.info("Job %s submitted, status available at: %s", groupByJob.getJobName(), groupByJob.getTrackingURL());
 
+        // Store the jobId in the file
+        if (groupByJob.getJobID() != null) {
+          JobHelper.writeJobIdToFile(config.getHadoopJobIdFileName(), groupByJob.getJobID().toString());
+        }
+
+
         if (!groupByJob.waitForCompletion(true)) {
           log.error("Job failed: %s", groupByJob.getJobID());
           failureCause = Utils.getFailureMessage(groupByJob, config.JSON_MAPPER);
@@ -218,6 +224,12 @@ public boolean run()
           dimSelectionJob.getTrackingURL()
       );
 
+      // Store the jobId in the file
+      if (dimSelectionJob.getJobID() != null) {
+        JobHelper.writeJobIdToFile(config.getHadoopJobIdFileName(), dimSelectionJob.getJobID().toString());
+      }
+
+
       if (!dimSelectionJob.waitForCompletion(true)) {
         log.error("Job failed: %s", dimSelectionJob.getJobID().toString());
         failureCause = Utils.getFailureMessage(dimSelectionJob, config.JSON_MAPPER);

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

@@ -39,6 +39,7 @@ public class HadoopDruidDetermineConfigurationJob implements Jobby
   private static final Logger log = new Logger(HadoopDruidDetermineConfigurationJob.class);
   private final HadoopDruidIndexerConfig config;
   private Jobby job;
+  private String hadoopJobIdFile;
 
   @Inject
   public HadoopDruidDetermineConfigurationJob(
@@ -55,6 +56,7 @@ public boolean run()
 
     if (config.isDeterminingPartitions()) {
       job = config.getPartitionsSpec().getPartitionJob(config);
+      config.setHadoopJobIdFileName(hadoopJobIdFile);
       return JobHelper.runSingleJob(job, config);
     } else {
       int shardsPerInterval = config.getPartitionsSpec().getNumShards();
@@ -109,4 +111,9 @@ public String getErrorMessage()
 
     return job.getErrorMessage();
   }
+
+  public void setHadoopJobIdFile(String hadoopJobIdFile)
+  {
+    this.hadoopJobIdFile = hadoopJobIdFile;
+  }
 }

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

@@ -212,6 +212,7 @@ public static HadoopDruidIndexerConfig fromConfiguration(Configuration conf)
 
   private HadoopIngestionSpec schema;
   private PathSpec pathSpec;
+  private String hadoopJobIdFileName;
   private final Map<Long, ShardSpecLookup> shardSpecLookups = new HashMap<>();
   private final Map<Long, Map<ShardSpec, HadoopyShardSpec>> hadoopShardSpecLookup = new HashMap<>();
   private final Granularity rollupGran;
@@ -375,6 +376,16 @@ public int getMaxParseExceptions()
     return schema.getTuningConfig().getMaxParseExceptions();
   }
 
+  public void setHadoopJobIdFileName(String hadoopJobIdFileName)
+  {
+    this.hadoopJobIdFileName = hadoopJobIdFileName;
+  }
+
+  public String getHadoopJobIdFileName()
+  {
+    return hadoopJobIdFileName;
+  }
+
   /**
    * Job instance should have Configuration set (by calling {@link #addJobProperties(Job)}
    * or via injected system properties) before this method is called.  The {@link PathSpec} may

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

@@ -38,6 +38,7 @@ public class HadoopDruidIndexerJob implements Jobby
   private final MetadataStorageUpdaterJob metadataStorageUpdaterJob;
   private IndexGeneratorJob indexJob;
   private volatile List<DataSegment> publishedSegments = null;
+  private String hadoopJobIdFile;
 
   @Inject
   public HadoopDruidIndexerJob(
@@ -92,7 +93,7 @@ public boolean run()
         }
     );
 
-
+    config.setHadoopJobIdFileName(hadoopJobIdFile);
     return JobHelper.runJobs(jobs, config);
   }
 
@@ -124,4 +125,9 @@ public List<DataSegment> getPublishedSegments()
     }
     return publishedSegments;
   }
+
+  public void setHadoopJobIdFile(String hadoopJobIdFile)
+  {
+    this.hadoopJobIdFile = hadoopJobIdFile;
+  }
 }

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

@@ -97,6 +97,7 @@
 import java.util.concurrent.TimeoutException;
 
 /**
+ *
  */
 public class IndexGeneratorJob implements Jobby
 {
@@ -207,6 +208,11 @@ public boolean run()
       job.submit();
       log.info("Job %s submitted, status available at %s", job.getJobName(), job.getTrackingURL());
 
+      // Store the jobId in the file
+      if (job.getJobID() != null) {
+        JobHelper.writeJobIdToFile(config.getHadoopJobIdFileName(), job.getJobID().toString());
+      }
+
       boolean success = job.waitForCompletion(true);
 
       Counters counters = job.getCounters();

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

@@ -59,8 +59,10 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.io.OutputStreamWriter;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -347,6 +349,24 @@ public static void ensurePaths(HadoopDruidIndexerConfig config)
     }
   }
 
+  public static void writeJobIdToFile(String hadoopJobIdFileName, String hadoopJobId)
+  {
+    if (hadoopJobId != null && hadoopJobIdFileName != null) {
+      try {
+        HadoopDruidIndexerConfig.JSON_MAPPER.writeValue(
+            new OutputStreamWriter(new FileOutputStream(new File(hadoopJobIdFileName)), StandardCharsets.UTF_8),
+            hadoopJobId
+        );
+        log.info("MR job id [%s] is written to the file [%s]", hadoopJobId, hadoopJobIdFileName);
+      }
+      catch (IOException e) {
+        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);
+    }
+  }
+
   public static boolean runSingleJob(Jobby job, HadoopDruidIndexerConfig config)
   {
     boolean succeeded = job.run();

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.
+   * If resource cleaning is required, concrete class should override this method
+   */
   @Override
   public void stopGracefully()
   {
-    // Should not be called when canRestore = false.
-    throw new UnsupportedOperationException("Cannot stop gracefully");
+    // Do nothing and let the concrete class handle it
   }
 
   @Override