Fix task report streaming in https setups. (#11739)

* Fix task report streaming in https setups.

* Trivial change to re-trigger ITs.

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

@@ -21,8 +21,11 @@ package org.apache.druid.indexer;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.java.util.common.IAE;
 
 import javax.annotation.Nullable;
+import java.net.MalformedURLException;
+import java.net.URL;
 import java.util.Objects;
 
 public class TaskLocation
@@ -75,6 +78,27 @@ public class TaskLocation
     return tlsPort;
   }
 
+  public URL makeURL(final String encodedPathAndQueryString) throws MalformedURLException
+  {
+    final String scheme;
+    final int portToUse;
+
+    if (tlsPort > 0) {
+      scheme = "https";
+      portToUse = tlsPort;
+    } else {
+      scheme = "http";
+      portToUse = port;
+    }
+
+    if (!encodedPathAndQueryString.startsWith("/")) {
+      throw new IAE("Path must start with '/'");
+    }
+
+    // Use URL constructor, not URI, since the path is already encoded.
+    return new URL(scheme, host, portToUse, encodedPathAndQueryString);
+  }
+
   @Override
   public String toString()
   {

core/src/test/java/org/apache/druid/indexer/TaskLocationTest.java

@@ -21,10 +21,28 @@ package org.apache.druid.indexer;
 
 
 import nl.jqno.equalsverifier.EqualsVerifier;
+import org.junit.Assert;
 import org.junit.Test;
 
+import java.net.MalformedURLException;
+import java.net.URL;
+
 public class TaskLocationTest
 {
+  @Test
+  @SuppressWarnings("HttpUrlsUsage")
+  public void testMakeURL() throws MalformedURLException
+  {
+    Assert.assertEquals(new URL("http://abc:80/foo"), new TaskLocation("abc", 80, 0).makeURL("/foo"));
+    Assert.assertEquals(new URL("http://abc:80/foo"), new TaskLocation("abc", 80, -1).makeURL("/foo"));
+    Assert.assertEquals(new URL("https://abc:443/foo"), new TaskLocation("abc", 80, 443).makeURL("/foo"));
+    Assert.assertThrows(
+        "URL that does not start with '/'",
+        IllegalArgumentException.class,
+        () -> new TaskLocation("abc", 80, 443).makeURL("foo")
+    );
+  }
+
   @Test
   public void testEqualsAndHashCode()
   {

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

@@ -276,16 +276,8 @@ public abstract class IndexTaskClient implements AutoCloseable
       byte[] content
   ) throws MalformedURLException
   {
-    final String host = location.getHost();
-    final String scheme = location.getTlsPort() >= 0 ? "https" : "http";
-    final int port = location.getTlsPort() >= 0 ? location.getTlsPort() : location.getPort();
-
-    // Use URL constructor, not URI, since the path is already encoded.
     // The below line can throw a MalformedURLException, and this method should return immediately without rety.
-    final URL serviceUrl = new URL(
-        scheme,
-        host,
-        port,
+    final URL serviceUrl = location.makeURL(
         encodedQueryString == null ? path : StringUtils.format("%s?%s", path, encodedQueryString)
     );
 

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

@@ -105,14 +105,9 @@ public class TaskRunnerUtils
     );
 
     try {
-      return new URI(StringUtils.format(
-          "http://%s:%s%s",
-          taskLocation.getHost(),
-          taskLocation.getPort(),
-          path
-      )).toURL();
+      return taskLocation.makeURL(path);
     }
-    catch (URISyntaxException | MalformedURLException e) {
+    catch (MalformedURLException e) {
       throw new RuntimeException(e);
     }
   }

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

@@ -19,6 +19,7 @@
 
 package org.apache.druid.indexing.overlord;
 
+import org.apache.druid.indexer.TaskLocation;
 import org.apache.druid.indexing.worker.Worker;
 import org.apache.druid.indexing.worker.config.WorkerConfig;
 import org.junit.Assert;
@@ -40,4 +41,15 @@ public class TaskRunnerUtilsTest
     Assert.assertEquals("1.2.3.4:8290", url.getAuthority());
     Assert.assertEquals("/druid/worker/v1/task/foo%20bar%26/log", url.getPath());
   }
+
+  @Test
+  public void testMakeTaskLocationURL()
+  {
+    final URL url = TaskRunnerUtils.makeTaskLocationURL(
+        new TaskLocation("1.2.3.4", 8090, 8290),
+        "/druid/worker/v1/task/%s/log",
+        "foo bar&"
+    );
+    Assert.assertEquals("https://1.2.3.4:8290/druid/worker/v1/task/foo%20bar%26/log", url.toString());
+  }
 }