minor: [JDK upgrade] Update forking task runner to use bin/run-java

docs/configuration/index.md

@@ -1315,7 +1315,7 @@ Middle Managers pass their configurations down to their child peons. The Middle
 |--------|-----------|-------|
 |`druid.indexer.runner.allowedPrefixes`|Whitelist of prefixes for configs that can be passed down to child peons.|`com.metamx`, `druid`, `org.apache.druid`, `user.timezone`, `file.encoding`, `java.io.tmpdir`, `hadoop`|
 |`druid.indexer.runner.classpath`|Java classpath for the peon.|`System.getProperty("java.class.path")`|
-|`druid.indexer.runner.javaCommand`|Command required to execute java.|java|
+|`druid.indexer.runner.javaCommand`| Command required to execute java. If not specified, Druid uses the bundled `bin/run-java` script when it is present in the working directory (it honors the `DRUID_JAVA_HOME`/`JAVA_HOME` environment variables), otherwise it falls back to `java` on the `PATH`.|`bin/run-java` if present, else `java`|
 |`druid.indexer.runner.javaOpts`|_DEPRECATED_ A string of -X Java options to pass to the peon's JVM. Quotable parameters or parameters with spaces are encouraged to use javaOptsArray|`''`|
 |`druid.indexer.runner.javaOptsArray`|A JSON array of strings to be passed in as options to the peon's JVM. This is additive to `druid.indexer.runner.javaOpts` and is recommended for properly handling arguments which contain quotes or spaces like `["-XX:OnOutOfMemoryError=kill -9 %p"]`|`[]`|
 |`druid.indexer.runner.startPort`|Starting port used for Peon services, should be greater than 1023 and less than 65536.|8100|

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

@@ -100,6 +100,7 @@ public class ForkingTaskRunner
   private static final EmittingLogger LOGGER = new EmittingLogger(ForkingTaskRunner.class);
   private static final String CHILD_PROPERTY_PREFIX = "druid.indexer.fork.property.";
 
+  private static final String RUN_JAVA_COMMAND = "bin/run-java";
   /**
    * Properties to add on Java 11+. When updating this list, update all four:
    *  1) ForkingTaskRunner#STRONG_ENCAPSULATION_PROPERTIES (here) -->
@@ -245,7 +246,7 @@ public TaskStatus call()
                           taskClasspath = config.getClasspath();
                         }
 
-                        command.add(config.getJavaCommand());
+                        command.add(getJavaCommand());
 
                         if (JvmUtils.majorVersion() >= 11) {
                           command.addAll(STRONG_ENCAPSULATION_PROPERTIES);
@@ -506,6 +507,33 @@ public TaskStatus call()
     }
   }
 
+  private String getJavaCommand()
+  {
+    return getJavaCommand(config.getJavaCommand(), new File("."));
+  }
+
+  /**
+   * Resolves the command used to launch the peon JVM, in order of precedence:
+   * <ol>
+   *   <li>the operator-specified {@code druid.indexer.runner.javaCommand}, if set;</li>
+   *   <li>the bundled {@link #RUN_JAVA_COMMAND} script if present in {@code workingDir}, so that
+   *   {@code DRUID_JAVA_HOME} / {@code JAVA_HOME} are honored;</li>
+   *   <li>otherwise plain {@code java} on the {@code PATH}.</li>
+   * </ol>
+   * The existence check against {@code workingDir} is reliable because peons inherit this process's working
+   * directory.
+   */
+  public static String getJavaCommand(@Nullable String configuredJavaCommand, File workingDir)
+  {
+    if (configuredJavaCommand != null) {
+      return configuredJavaCommand;
+    }
+    if (new File(workingDir, RUN_JAVA_COMMAND).exists()) {
+      return RUN_JAVA_COMMAND;
+    }
+    return ForkingTaskRunnerConfig.DEFAULT_JAVA_COMMAND;
+  }
+
   @VisibleForTesting
   ProcessHolder runTaskProcess(List<String> command, File logFile, TaskLocation taskLocation) throws IOException
   {

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

@@ -24,9 +24,11 @@
 import com.google.common.collect.Lists;
 import org.apache.druid.guice.IndexingServiceModuleHelper;
 
+import javax.annotation.Nullable;
 import javax.validation.constraints.Max;
 import javax.validation.constraints.Min;
 import javax.validation.constraints.NotNull;
+import java.io.File;
 import java.util.List;
 
 public class ForkingTaskRunnerConfig
@@ -36,9 +38,14 @@ public class ForkingTaskRunnerConfig
   public static final String JAVA_OPTS_ARRAY_PROPERTY = IndexingServiceModuleHelper.INDEXER_RUNNER_PROPERTY_PREFIX
                                                   + ".javaOptsArray";
 
+  public static final String DEFAULT_JAVA_COMMAND = "java";
+
+  /**
+   * Intentionally has no default. A null value means the operator has not overridden it, which lets
+   * {@code ForkingTaskRunner} decide between the bundled {@code bin/run-java} script and plain {@code java}.
+   */
   @JsonProperty
-  @NotNull
-  private String javaCommand = "java";
+  private String javaCommand = null;
 
   /**
    * This is intended for setting -X parameters on the underlying java.  It is used by first splitting on whitespace,
@@ -87,6 +94,12 @@ public class ForkingTaskRunnerConfig
       "hadoop"
   );
 
+  /**
+   * See method {@link org.apache.druid.indexing.overlord.ForkingTaskRunner#getJavaCommand(String, File)} method to see
+   * how this config is used.
+   * @return
+   */
+  @Nullable
   public String getJavaCommand()
   {
     return javaCommand;

indexing-service/src/test/java/org/apache/druid/indexing/overlord/ForkingTaskRunnerTest.java

@@ -41,6 +41,7 @@
 import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorStateTest;
 import org.apache.druid.indexing.worker.config.WorkerConfig;
 import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.FileUtils;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.java.util.common.RE;
 import org.apache.druid.java.util.common.granularity.AllGranularity;
@@ -79,6 +80,52 @@ public class ForkingTaskRunnerTest
   @Rule
   public TemporaryFolder temporaryFolder = new TemporaryFolder();
 
+  @Test
+  public void testGetJavaCommandPrefersRunJavaScriptWhenPresent() throws IOException
+  {
+    final File workingDir = temporaryFolder.newFolder();
+    final File binDir = new File(workingDir, "bin");
+    FileUtils.mkdirp(binDir);
+    Assert.assertTrue(new File(binDir, "run-java").createNewFile());
+
+    // No configured command (null) and the run-java script is present -> use it.
+    Assert.assertEquals(
+        "bin/run-java",
+        ForkingTaskRunner.getJavaCommand(null, workingDir)
+    );
+  }
+
+  @Test
+  public void testGetJavaCommandFallsBackToJavaWhenScriptAbsent() throws IOException
+  {
+    final File workingDir = temporaryFolder.newFolder();
+    Assert.assertEquals(
+        "java",
+        ForkingTaskRunner.getJavaCommand(null, workingDir)
+    );
+  }
+
+  @Test
+  public void testConfigHasNoDefaultJavaCommand()
+  {
+    Assert.assertNull(new ForkingTaskRunnerConfig().getJavaCommand());
+  }
+
+  @Test
+  public void testGetJavaCommandRespectsExplicitOverride() throws IOException
+  {
+    final File workingDir = temporaryFolder.newFolder();
+    final File binDir = new File(workingDir, "bin");
+    FileUtils.mkdirp(binDir);
+    Assert.assertTrue(new File(binDir, "run-java").createNewFile());
+
+    // An explicit, non-default javaCommand must be used verbatim even when the run-java script is present.
+    Assert.assertEquals(
+        "/opt/jdk/bin/java",
+        ForkingTaskRunner.getJavaCommand("/opt/jdk/bin/java", workingDir)
+    );
+  }
+
   // This tests the test to make sure the test fails when it should.
   @Test(expected = AssertionError.class)
   public void testPatternMatcherFailureForJavaOptions()