indexing-service no longer depends on indexing-hadoop (#18247)

* indexing-service no longer depends on indexing-hadoop, instead indexing-hadoop depends on indexing-service

* fixes

* cleanup

* fix it

* fix declared dependencies

* suppress warning

* try again

* suppress check

Author: clintropolis

examples/quickstart/tutorial/hadoop/docker/Dockerfile

@@ -52,12 +52,12 @@ RUN rpm --import http://repos.azulsystems.com/RPM-GPG-KEY-azulsystems && \
     rpm -ivh zulu-repo-${ZULU_REPO_VER}.noarch.rpm && \
     yum -q -y update && \
     yum -q -y upgrade && \
-    yum -q -y install zulu17-jdk && \
+    yum -q -y install zulu11-jdk && \
     yum -q -y install nano net-tools telnet less unzip wget && \
     yum clean all && \
     rm -rf /var/cache/yum zulu-repo_${ZULU_REPO_VER}.noarch.rpm
 
-ENV JAVA_HOME=/usr/lib/jvm/zulu17
+ENV JAVA_HOME=/usr/lib/jvm/zulu11
 ENV PATH=$PATH:$JAVA_HOME/bin
 
 # hadoop
@@ -76,7 +76,7 @@ ENV YARN_CONF_DIR=$HADOOP_HOME/etc/hadoop
 # in hadoop 3 the example file is nearly empty so we can just append stuff
 RUN cat << EOT >> $HADOOP_HOME/etc/hadoop/hadoop-env.sh
 
-RUN sed -i '$ a export JAVA_HOME=/usr/lib/jvm/zulu17' $HADOOP_HOME/etc/hadoop/hadoop-env.sh
+RUN sed -i '$ a export JAVA_HOME=/usr/lib/jvm/zulu11' $HADOOP_HOME/etc/hadoop/hadoop-env.sh
 RUN sed -i '$ a export HADOOP_HOME=/usr/local/hadoop' $HADOOP_HOME/etc/hadoop/hadoop-env.sh
 RUN sed -i '$ a export HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoop/' $HADOOP_HOME/etc/hadoop/hadoop-env.sh
 RUN sed -i '$ a export HDFS_NAMENODE_USER=root' $HADOOP_HOME/etc/hadoop/hadoop-env.sh

extensions-contrib/materialized-view-maintenance/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisor.java

@@ -29,8 +29,8 @@
 import com.google.common.util.concurrent.MoreExecutors;
 import org.apache.druid.error.DruidException;
 import org.apache.druid.error.EntryAlreadyExists;
+import org.apache.druid.indexer.HadoopIndexTask;
 import org.apache.druid.indexer.TaskStatus;
-import org.apache.druid.indexing.common.task.HadoopIndexTask;
 import org.apache.druid.indexing.overlord.DataSourceMetadata;
 import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
 import org.apache.druid.indexing.overlord.Segments;

extensions-contrib/materialized-view-maintenance/src/main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisorSpec.java

@@ -29,11 +29,12 @@
 import org.apache.druid.data.input.impl.DimensionSchema;
 import org.apache.druid.data.input.impl.DimensionsSpec;
 import org.apache.druid.indexer.HadoopIOConfig;
+import org.apache.druid.indexer.HadoopIndexTask;
 import org.apache.druid.indexer.HadoopIngestionSpec;
+import org.apache.druid.indexer.HadoopTaskConfig;
 import org.apache.druid.indexer.HadoopTuningConfig;
 import org.apache.druid.indexer.granularity.ArbitraryGranularitySpec;
 import org.apache.druid.indexer.hadoop.DatasourceIngestionSpec;
-import org.apache.druid.indexing.common.task.HadoopIndexTask;
 import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
 import org.apache.druid.indexing.overlord.TaskMaster;
 import org.apache.druid.indexing.overlord.TaskStorage;
@@ -245,7 +246,8 @@ public HadoopIndexTask createTask(Interval interval, String version, List<DataSe
         objectMapper,
         context,
         authorizerMapper,
-        chatHandlerProvider
+        chatHandlerProvider,
+        new HadoopTaskConfig(null, null)
     );
 
     return task;

extensions-contrib/materialized-view-maintenance/src/test/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisorTest.java

@@ -30,10 +30,11 @@
 import org.apache.druid.data.input.impl.StringDimensionSchema;
 import org.apache.druid.error.EntryAlreadyExists;
 import org.apache.druid.indexer.HadoopIOConfig;
+import org.apache.druid.indexer.HadoopIndexTask;
 import org.apache.druid.indexer.HadoopIngestionSpec;
+import org.apache.druid.indexer.HadoopTaskConfig;
 import org.apache.druid.indexer.HadoopTuningConfig;
 import org.apache.druid.indexer.TaskStatus;
-import org.apache.druid.indexing.common.task.HadoopIndexTask;
 import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
 import org.apache.druid.indexing.overlord.TaskMaster;
 import org.apache.druid.indexing.overlord.TaskQueue;
@@ -259,7 +260,8 @@ public void testCheckSegmentsAndSubmitTasks()
         objectMapper,
         null,
         null,
-        null
+        null,
+        new HadoopTaskConfig(null, null)
     );
     runningTasks.put(Intervals.of("2015-01-01T00Z/2015-01-02T00Z"), task1);
     runningVersion.put(Intervals.of("2015-01-01T00Z/2015-01-02T00Z"), "test_version1");
@@ -273,7 +275,8 @@ public void testCheckSegmentsAndSubmitTasks()
         objectMapper,
         null,
         null,
-        null
+        null,
+        new HadoopTaskConfig(null, null)
     );
     runningTasks.put(Intervals.of("2015-01-02T00Z/2015-01-03T00Z"), task2);
     runningVersion.put(Intervals.of("2015-01-02T00Z/2015-01-03T00Z"), "test_version2");

indexing-hadoop/pom.xml

@@ -48,6 +48,11 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.apache.druid</groupId>
+            <artifactId>druid-indexing-service</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
 
         <dependency>
             <groupId>commons-io</groupId>
@@ -89,6 +94,18 @@
             <groupId>org.eclipse.jetty</groupId>
             <artifactId>jetty-client</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>javax.servlet</groupId>
+            <artifactId>javax.servlet-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>javax.ws.rs</groupId>
+            <artifactId>jsr311-api</artifactId>
+        </dependency>
 
         <!-- Tests -->
         <dependency>
@@ -130,6 +147,13 @@
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.druid</groupId>
+            <artifactId>druid-indexing-service</artifactId>
+            <version>${project.parent.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.derby</groupId>
             <artifactId>derbyclient</artifactId>
@@ -152,7 +176,6 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
-
     <profiles>
         <profile>
             <id>hadoop3</id>
@@ -197,6 +220,11 @@
                     <scope>test</scope>
                 </dependency>
             </dependencies>
+            <properties>
+                <hadoop-task-libs>
+                    org.apache.hadoop:hadoop-client-api:${hadoop.compile.version},org.apache.hadoop:hadoop-client-runtime:${hadoop.compile.version}
+                </hadoop-task-libs>
+            </properties>
         </profile>
     </profiles>
     <build>
@@ -224,7 +252,25 @@
                     </execution>
                 </executions>
             </plugin>
+            <plugin>
+                <artifactId>maven-resources-plugin</artifactId>
+                <groupId>org.apache.maven.plugins</groupId>
+            </plugin>
         </plugins>
+        <resources>
+            <resource>
+                <directory>src/main/resources</directory>
+                <includes>
+                    <include>hadoop.indexer.libs.version</include>
+                </includes>
+                <filtering>true</filtering>
+            </resource>
+        </resources>
+        <testResources>
+            <testResource>
+                <directory>src/test/resources</directory>
+                <filtering>false</filtering>
+            </testResource>
+        </testResources>
     </build>
-
 </project>

…ndexing/common/task/HadoopIndexTask.java …pache/druid/indexer/HadoopIndexTask.javaindexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java renamed to indexing-hadoop/src/main/java/org/apache/druid/indexer/HadoopIndexTask.java indexing-service/src/main/java/org/apache/druid/indexing/common/task/HadoopIndexTask.java renamed to indexing-hadoop/src/main/java/org/apache/druid/indexer/HadoopIndexTask.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.druid.indexing.common.task;
+package org.apache.druid.indexer;
 
 import com.fasterxml.jackson.annotation.JacksonInject;
 import com.fasterxml.jackson.annotation.JsonCreator;
@@ -32,16 +32,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterables;
 import org.apache.commons.lang3.BooleanUtils;
-import org.apache.druid.indexer.DataSegmentAndIndexZipFilePath;
-import org.apache.druid.indexer.HadoopDruidDetermineConfigurationJob;
-import org.apache.druid.indexer.HadoopDruidIndexerConfig;
-import org.apache.druid.indexer.HadoopDruidIndexerJob;
-import org.apache.druid.indexer.HadoopIngestionSpec;
-import org.apache.druid.indexer.IngestionState;
-import org.apache.druid.indexer.JobHelper;
-import org.apache.druid.indexer.TaskMetricsGetter;
-import org.apache.druid.indexer.TaskMetricsUtils;
-import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.annotations.SuppressFBWarnings;
 import org.apache.druid.indexer.granularity.ArbitraryGranularitySpec;
 import org.apache.druid.indexer.granularity.GranularitySpec;
 import org.apache.druid.indexer.path.SegmentMetadataPublisher;
@@ -53,7 +44,8 @@
 import org.apache.druid.indexing.common.actions.TimeChunkLockAcquireAction;
 import org.apache.druid.indexing.common.actions.TimeChunkLockTryAcquireAction;
 import org.apache.druid.indexing.common.config.TaskConfig;
-import org.apache.druid.indexing.hadoop.OverlordActionBasedUsedSegmentsRetriever;
+import org.apache.druid.indexing.common.task.AbstractTask;
+import org.apache.druid.indexing.common.task.Tasks;
 import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
 import org.apache.druid.java.util.common.JodaUtils;
 import org.apache.druid.java.util.common.StringUtils;
@@ -90,6 +82,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+@SuppressFBWarnings({"NP_NONNULL_PARAM_VIOLATION", "NP_STORE_INTO_NONNULL_FIELD"})
 public class HadoopIndexTask extends HadoopTask implements ChatHandler
 {
   public static final String TYPE = "index_hadoop";
@@ -115,6 +108,9 @@ private static String getTheDataSource(HadoopIngestionSpec spec)
   @JsonIgnore
   private final AuthorizerMapper authorizerMapper;
 
+  @JsonIgnore
+  private final HadoopTaskConfig hadoopTaskConfig;
+
   @JsonIgnore
   private final Optional<ChatHandlerProvider> chatHandlerProvider;
 
@@ -136,6 +132,7 @@ private static String getTheDataSource(HadoopIngestionSpec spec)
   @JsonIgnore
   private String errorMsg;
 
+
   /**
    * @param spec is used by the HadoopDruidIndexerJob to set up the appropriate parameters
    *             for creating Druid index segments. It may be modified.
@@ -156,16 +153,18 @@ public HadoopIndexTask(
       @JacksonInject ObjectMapper jsonMapper,
       @JsonProperty("context") Map<String, Object> context,
       @JacksonInject AuthorizerMapper authorizerMapper,
-      @JacksonInject ChatHandlerProvider chatHandlerProvider
+      @JacksonInject ChatHandlerProvider chatHandlerProvider,
+      @JacksonInject HadoopTaskConfig hadoopTaskConfig
   )
   {
     super(
-        getOrMakeId(id, TYPE, getTheDataSource(spec)),
+        AbstractTask.getOrMakeId(id, TYPE, getTheDataSource(spec)),
         getTheDataSource(spec),
         hadoopDependencyCoordinates == null
         ? (hadoopCoordinates == null ? null : ImmutableList.of(hadoopCoordinates))
         : hadoopDependencyCoordinates,
-        context
+        context,
+        hadoopTaskConfig
     );
     this.authorizerMapper = authorizerMapper;
     this.chatHandlerProvider = Optional.fromNullable(chatHandlerProvider);
@@ -181,6 +180,7 @@ public HadoopIndexTask(
         this.spec.getIOConfig().getMetadataUpdateSpec() == null,
         "metadataUpdateSpec must be absent"
     );
+    this.hadoopTaskConfig = hadoopTaskConfig;
 
     this.classpathPrefix = classpathPrefix;
     this.jsonMapper = Preconditions.checkNotNull(jsonMapper, "null ObjectMappper");
@@ -338,8 +338,8 @@ private TaskStatus runInternal(TaskToolbox toolbox) throws Exception
     try {
       registerResourceCloserOnAbnormalExit(config -> killHadoopJob());
       String hadoopJobIdFile = getHadoopJobIdFileName();
-      logExtensionsConfig();
-      final ClassLoader loader = buildClassLoader(toolbox);
+      HadoopTask.logExtensionsConfig();
+      final ClassLoader loader = buildClassLoader();
       boolean determineIntervals = spec.getDataSchema().getGranularitySpec().inputIntervals().isEmpty();
 
       HadoopIngestionSpec.updateSegmentListIfDatasourcePathSpecIsUsed(
@@ -348,15 +348,15 @@ private TaskStatus runInternal(TaskToolbox toolbox) throws Exception
           new OverlordActionBasedUsedSegmentsRetriever(toolbox)
       );
 
-      Object determinePartitionsInnerProcessingRunner = getForeignClassloaderObject(
-          "org.apache.druid.indexing.common.task.HadoopIndexTask$HadoopDetermineConfigInnerProcessingRunner",
+      Object determinePartitionsInnerProcessingRunner = HadoopTask.getForeignClassloaderObject(
+          "org.apache.druid.indexer.HadoopIndexTask$HadoopDetermineConfigInnerProcessingRunner",
           loader
       );
       determinePartitionsStatsGetter = new InnerProcessingStatsGetter(determinePartitionsInnerProcessingRunner);
 
       String[] determinePartitionsInput = new String[]{
           toolbox.getJsonMapper().writeValueAsString(spec),
-          toolbox.getConfig().getHadoopWorkingPath(),
+          hadoopTaskConfig.getHadoopWorkingPath(),
           toolbox.getSegmentPusher().getPathForHadoop(),
           hadoopJobIdFile
       };
@@ -418,7 +418,7 @@ private TaskStatus runInternal(TaskToolbox toolbox) throws Exception
         lock.assertNotRevoked();
         version = lock.getVersion();
       } else {
-        Iterable<TaskLock> locks = getTaskLocks(toolbox.getTaskActionClient());
+        Iterable<TaskLock> locks = AbstractTask.getTaskLocks(toolbox.getTaskActionClient());
         final TaskLock myLock = Iterables.getOnlyElement(locks);
         version = myLock.getVersion();
       }
@@ -442,8 +442,8 @@ private TaskStatus runInternal(TaskToolbox toolbox) throws Exception
 
       log.info("Setting version to: %s", version);
 
-      Object innerProcessingRunner = getForeignClassloaderObject(
-          "org.apache.druid.indexing.common.task.HadoopIndexTask$HadoopIndexGeneratorInnerProcessingRunner",
+      Object innerProcessingRunner = HadoopTask.getForeignClassloaderObject(
+          "org.apache.druid.indexer.HadoopIndexTask$HadoopIndexGeneratorInnerProcessingRunner",
           loader
       );
       buildSegmentsStatsGetter = new InnerProcessingStatsGetter(innerProcessingRunner);
@@ -535,11 +535,11 @@ private void killHadoopJob()
       try {
         ClassLoader loader = HadoopTask.buildClassLoader(
             getHadoopDependencyCoordinates(),
-            taskConfig.getDefaultHadoopCoordinates()
+            hadoopTaskConfig.getDefaultHadoopCoordinates()
         );
 
-        Object killMRJobInnerProcessingRunner = getForeignClassloaderObject(
-            "org.apache.druid.indexing.common.task.HadoopIndexTask$HadoopKillMRJobIdProcessingRunner",
+        Object killMRJobInnerProcessingRunner = HadoopTask.getForeignClassloaderObject(
+            "org.apache.druid.indexer.HadoopIndexTask$HadoopKillMRJobIdProcessingRunner",
             loader
         );
 
@@ -576,7 +576,7 @@ private void renameSegmentIndexFilesJob(
     final ClassLoader loader = Thread.currentThread().getContextClassLoader();
     try {
       final Class<?> clazz = loader.loadClass(
-          "org.apache.druid.indexing.common.task.HadoopIndexTask$HadoopRenameSegmentIndexFilesRunner"
+          "org.apache.druid.indexer.HadoopIndexTask$HadoopRenameSegmentIndexFilesRunner"
       );
       Object renameSegmentIndexFilesRunner = clazz.newInstance();
 
@@ -616,11 +616,11 @@ private void indexerGeneratorCleanupJob(
     try {
       ClassLoader loader = HadoopTask.buildClassLoader(
           getHadoopDependencyCoordinates(),
-          taskConfig.getDefaultHadoopCoordinates()
+          hadoopTaskConfig.getDefaultHadoopCoordinates()
       );
 
-      Object indexerGeneratorCleanupRunner = getForeignClassloaderObject(
-          "org.apache.druid.indexing.common.task.HadoopIndexTask$HadoopIndexerGeneratorCleanupRunner",
+      Object indexerGeneratorCleanupRunner = HadoopTask.getForeignClassloaderObject(
+          "org.apache.druid.indexer.HadoopIndexTask$HadoopIndexerGeneratorCleanupRunner",
           loader
       );