diff --git a/.travis.yml b/.travis.yml
index c0e093dab78..8750ca48dd9 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -93,8 +93,8 @@ matrix:
       services:
         - docker
       env:
-        - NAME="integration test"
-        - DOCKER_IP=172.17.0.1
+        - NAME="integration test part 1"
+        - DOCKER_IP=127.0.0.1
       install:
         # Only errors will be shown with the -q option. This is to avoid generating too many logs which make travis build failed.
         - mvn install -q -ff -DskipTests -B
@@ -108,3 +108,24 @@ matrix:
           echo $v dmesg ======================== ;
           docker exec -it druid-$v sh -c 'dmesg | tail -3' ;
           done
+
+      # run integration tests
+    - sudo: required
+      services:
+        - docker
+      env:
+        - NAME="integration test part 2"
+        - DOCKER_IP=127.0.0.1
+      install:
+        # Only errors will be shown with the -q option. This is to avoid generating too many logs which make travis build failed.
+        - mvn install -q -ff -DskipTests -B
+      script:
+        - $TRAVIS_BUILD_DIR/ci/travis_script_integration_part2.sh
+      after_failure:
+        - for v in ~/shared/logs/*.log ; do
+          echo $v logtail ======================== ; tail -100 $v ;
+          done
+        - for v in broker middlemanager overlord router coordinator historical ; do
+          echo $v dmesg ======================== ;
+          docker exec -it druid-$v sh -c 'dmesg | tail -3' ;
+          done
diff --git a/ci/travis_script_integration.sh b/ci/travis_script_integration.sh
index 3d0ca621963..b4bb1af342d 100755
--- a/ci/travis_script_integration.sh
+++ b/ci/travis_script_integration.sh
@@ -21,6 +21,6 @@ set -e
 
 pushd $TRAVIS_BUILD_DIR/integration-tests
 
-mvn verify -P integration-tests
+mvn verify -P integration-tests -Dit.test=ITAppenderatorDriverRealtimeIndexTaskTest,ITCompactionTaskTest,ITIndexerTest,ITKafkaIndexingServiceTest,ITKafkaTest,ITParallelIndexTest,ITRealtimeIndexTaskTest
 
 popd
diff --git a/ci/travis_script_integration_part2.sh b/ci/travis_script_integration_part2.sh
new file mode 100755
index 00000000000..61b3b9b7b6f
--- /dev/null
+++ b/ci/travis_script_integration_part2.sh
@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+set -e
+
+pushd $TRAVIS_BUILD_DIR/integration-tests
+
+mvn verify -P integration-tests -Dit.test=ITUnionQueryTest,ITTwitterQueryTest,ITWikipediaQueryTest,ITBasicAuthConfigurationTest,ITTLSTest
+
+popd
diff --git a/docs/content/development/extensions-core/simple-client-sslcontext.md b/docs/content/development/extensions-core/simple-client-sslcontext.md
index de7dfb1d664..31ae3d6eae7 100644
--- a/docs/content/development/extensions-core/simple-client-sslcontext.md
+++ b/docs/content/development/extensions-core/simple-client-sslcontext.md
@@ -18,5 +18,16 @@ Java's SSL support, please refer to [this](http://docs.oracle.com/javase/8/docs/
 |`druid.client.https.trustStoreAlgorithm`|Algorithm to be used by TrustManager to validate certificate chains|`javax.net.ssl.TrustManagerFactory.getDefaultAlgorithm()`|no|
 |`druid.client.https.trustStorePassword`|The [Password Provider](../../operations/password-provider.html) or String password for the Trust Store.|none|yes|
 
+The following table contains optional parameters for supporting client certificate authentication:
+
+|Property|Description|Default|Required|
+|--------|-----------|-------|--------|
+|`druid.client.https.keyStorePath`|The file path or URL of the TLS/SSL Key store containing the client certificate that Druid will use when communicating with other Druid services. If this is null, the other properties in this table are ignored.|none|yes|
+|`druid.client.https.keyStoreType`|The type of the key store.|none|yes|
+|`druid.client.https.certAlias`|Alias of TLS client certificate in the keystore.|none|yes|
+|`druid.client.https.keyStorePassword`|The [Password Provider](../operations/password-provider.html) or String password for the Key Store.|none|no|
+|`druid.client.https.keyManagerFactoryAlgorithm`|Algorithm to use for creating KeyManager, more details [here](https://docs.oracle.com/javase/7/docs/technotes/guides/security/jsse/JSSERefGuide.html#KeyManager).|`javax.net.ssl.KeyManagerFactory.getDefaultAlgorithm()`|no|
+|`druid.client.https.keyManagerPassword`|The [Password Provider](../operations/password-provider.html) or String password for the Key Manager.|none|no|
+
 This [document](http://docs.oracle.com/javase/8/docs/technotes/guides/security/StandardNames.html) lists all the possible
 values for the above mentioned configs among others provided by Java implementation.
\ No newline at end of file
diff --git a/docs/content/operations/tls-support.md b/docs/content/operations/tls-support.md
index bc004b1da62..63da285a315 100644
--- a/docs/content/operations/tls-support.md
+++ b/docs/content/operations/tls-support.md
@@ -31,7 +31,19 @@ values for the below mentioned configs among others provided by Java implementat
 |`druid.server.https.certAlias`|Alias of TLS/SSL certificate for the connector.|none|yes|
 |`druid.server.https.keyStorePassword`|The [Password Provider](../operations/password-provider.html) or String password for the Key Store.|none|yes|
 
-Following table contains non-mandatory advanced configuration options, use caution.
+The following table contains configuration options related to client certificate authentication.
+
+|Property|Description|Default|Required|
+|--------|-----------|-------|--------|
+|`druid.server.https.requireClientCertificate`|If set to true, clients must identify themselves by providing a TLS certificate.  If `requireClientCertificate` is false, the rest of the options in this table are ignored.|false|no|
+|`druid.server.https.trustStoreType`|The type of the trust store containing certificates used to validate client certificates. Not needed if `requireClientCertificate` is false.|`java.security.KeyStore.getDefaultType()`|no|
+|`druid.server.https.trustStorePath`|The file path or URL of the trust store containing certificates used to validate client certificates. Not needed if `requireClientCertificate` is false.|none|yes, only if `requireClientCertificate` is true|
+|`druid.server.https.trustStoreAlgorithm`|Algorithm to be used by TrustManager to validate client certificate chains. Not needed if `requireClientCertificate` is false.|`javax.net.ssl.TrustManagerFactory.getDefaultAlgorithm()`|no|
+|`druid.server.https.trustStorePassword`|The [Password Provider](../../operations/password-provider.html) or String password for the Trust Store.  Not needed if `requireClientCertificate` is false.|none|no|
+|`druid.server.https.validateHostnames`|If set to true, check that the client's hostname matches the CN/subjectAltNames in the client certificate.  Not used if `requireClientCertificate` is false.|true|no|
+|`druid.server.https.crlPath`|Specifies a path to a file containing static [Certificate Revocation Lists](https://en.wikipedia.org/wiki/Certificate_revocation_list), used to check if a client certificate has been revoked. Not used if `requireClientCertificate` is false.|null|no|
+
+The following table contains non-mandatory advanced configuration options, use caution.
 
 |Property|Description|Default|Required|
 |--------|-----------|-------|--------|
diff --git a/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLClientConfig.java b/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLClientConfig.java
index 27761effb12..b7061cb84f3 100644
--- a/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLClientConfig.java
+++ b/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLClientConfig.java
@@ -39,6 +39,24 @@ public class SSLClientConfig
   @JsonProperty("trustStorePassword")
   private PasswordProvider trustStorePasswordProvider;
 
+  @JsonProperty
+  private String keyStorePath;
+
+  @JsonProperty
+  private String keyStoreType;
+
+  @JsonProperty
+  private String certAlias;
+
+  @JsonProperty("keyStorePassword")
+  private PasswordProvider keyStorePasswordProvider;
+
+  @JsonProperty("keyManagerPassword")
+  private PasswordProvider keyManagerPasswordProvider;
+
+  @JsonProperty
+  private String keyManagerFactoryAlgorithm;
+
   public String getProtocol()
   {
     return protocol;
@@ -64,6 +82,36 @@ public class SSLClientConfig
     return trustStorePasswordProvider;
   }
 
+  public String getKeyStorePath()
+  {
+    return keyStorePath;
+  }
+
+  public String getKeyStoreType()
+  {
+    return keyStoreType;
+  }
+
+  public PasswordProvider getKeyStorePasswordProvider()
+  {
+    return keyStorePasswordProvider;
+  }
+
+  public String getCertAlias()
+  {
+    return certAlias;
+  }
+
+  public PasswordProvider getKeyManagerPasswordProvider()
+  {
+    return keyManagerPasswordProvider;
+  }
+
+  public String getKeyManagerFactoryAlgorithm()
+  {
+    return keyManagerFactoryAlgorithm;
+  }
+
   @Override
   public String toString()
   {
@@ -72,6 +120,10 @@ public class SSLClientConfig
            ", trustStoreType='" + trustStoreType + '\'' +
            ", trustStorePath='" + trustStorePath + '\'' +
            ", trustStoreAlgorithm='" + trustStoreAlgorithm + '\'' +
+           ", keyStorePath='" + keyStorePath + '\'' +
+           ", keyStoreType='" + keyStoreType + '\'' +
+           ", certAlias='" + certAlias + '\'' +
+           ", keyManagerFactoryAlgorithm='" + keyManagerFactoryAlgorithm + '\'' +
            '}';
   }
 }
diff --git a/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLContextProvider.java b/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLContextProvider.java
index fc5dba4af12..826791deb64 100644
--- a/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLContextProvider.java
+++ b/extensions-core/simple-client-sslcontext/src/main/java/org/apache/druid/https/SSLContextProvider.java
@@ -43,12 +43,18 @@ public class SSLContextProvider implements Provider<SSLContext>
   {
     log.info("Creating SslContext for https client using config [%s]", config);
 
-    return TLSUtils.createSSLContext(
-        config.getProtocol(),
-        config.getTrustStoreType(),
-        config.getTrustStorePath(),
-        config.getTrustStoreAlgorithm(),
-        config.getTrustStorePasswordProvider()
-    );
+    return new TLSUtils.ClientSSLContextBuilder()
+        .setProtocol(config.getProtocol())
+        .setTrustStoreType(config.getTrustStoreType())
+        .setTrustStorePath(config.getTrustStorePath())
+        .setTrustStoreAlgorithm(config.getTrustStoreAlgorithm())
+        .setTrustStorePasswordProvider(config.getTrustStorePasswordProvider())
+        .setKeyStoreType(config.getKeyStoreType())
+        .setKeyStorePath(config.getKeyStorePath())
+        .setKeyStoreAlgorithm(config.getKeyManagerFactoryAlgorithm())
+        .setCertAlias(config.getCertAlias())
+        .setKeyStorePasswordProvider(config.getKeyStorePasswordProvider())
+        .setKeyManagerFactoryPasswordProvider(config.getKeyManagerPasswordProvider())
+        .build();
   }
 }
diff --git a/integration-tests/.gitignore b/integration-tests/.gitignore
new file mode 100644
index 00000000000..aa0b0953609
--- /dev/null
+++ b/integration-tests/.gitignore
@@ -0,0 +1,6 @@
+client_tls/
+docker/docker_ip
+docker/tls/root.key
+docker/tls/root.pem
+docker/tls/untrusted_root.key
+docker/tls/untrusted_root.pem
\ No newline at end of file
diff --git a/integration-tests/docker/Dockerfile b/integration-tests/docker/Dockerfile
index e0f104f1d15..5cbb54c4d8b 100644
--- a/integration-tests/docker/Dockerfile
+++ b/integration-tests/docker/Dockerfile
@@ -39,32 +39,43 @@ RUN find /var/lib/mysql -type f -exec touch {} \; && service mysql start \
 # Setup supervisord
 ADD supervisord.conf /etc/supervisor/conf.d/supervisord.conf
 
+# mysql
+ADD run-mysql.sh /run-mysql.sh
+
 # internal docker_ip:9092 endpoint is used to access Kafka from other Docker containers
 # external docker ip:9093 endpoint is used to access Kafka from test code
 # run this last to avoid rebuilding the image every time the ip changes
 ADD docker_ip docker_ip
-RUN perl -pi -e "s/#listeners=.*/listeners=INTERNAL:\/\/$(resolveip -s $HOSTNAME):9092,EXTERNAL:\/\/$(resolveip -s $HOSTNAME):9093/" /usr/local/kafka/config/server.properties
-RUN perl -pi -e "s/#advertised.listeners=.*/advertised.listeners=INTERNAL:\/\/$(resolveip -s $HOSTNAME):9092,EXTERNAL:\/\/$(cat docker_ip):9093/" /usr/local/kafka/config/server.properties
+RUN perl -pi -e "s/#listeners=.*/listeners=INTERNAL:\/\/172.172.172.2:9092,EXTERNAL:\/\/172.172.172.2:9093/" /usr/local/kafka/config/server.properties
+RUN perl -pi -e "s/#advertised.listeners=.*/advertised.listeners=INTERNAL:\/\/172.172.172.2:9092,EXTERNAL:\/\/$(cat docker_ip):9093/" /usr/local/kafka/config/server.properties
 RUN perl -pi -e "s/#listener.security.protocol.map=.*/listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT\ninter.broker.listener.name=INTERNAL/" /usr/local/kafka/config/server.properties
 RUN perl
 
+# Add directory with TLS support files
+ADD tls tls
+
+ADD client_tls client_tls
+
 # Expose ports:
-# - 8081: HTTP (coordinator)
-# - 8082: HTTP (broker)
-# - 8083: HTTP (historical)
-# - 8090: HTTP (overlord)
-# - 8091: HTTP (middlemanager)
+# - 8081, 8281: HTTP, HTTPS (coordinator)
+# - 8082, 8282: HTTP, HTTPS (broker)
+# - 8083, 8283: HTTP, HTTPS (historical)
+# - 8090, 8290: HTTP, HTTPS (overlord)
+# - 8091, 8291: HTTP, HTTPS (middlemanager)
 # - 3306: MySQL
 # - 2181 2888 3888: ZooKeeper
-# - 8100 8101 8102 8103 8104 : peon ports
-EXPOSE 8081
-EXPOSE 8082
-EXPOSE 8083
-EXPOSE 8090
-EXPOSE 8091
+# - 8100 8101 8102 8103 8104 8105 : peon ports
+# - 8300 8301 8302 8303 8304 8305 : peon HTTPS ports
+EXPOSE 8081 8281
+EXPOSE 8082 8282
+EXPOSE 8083 8283
+EXPOSE 8090 8290
+EXPOSE 8091 8291
 EXPOSE 3306
 EXPOSE 2181 2888 3888
-EXPOSE 8100 8101 8102 8103 8104
+EXPOSE 8100 8101 8102 8103 8104 8105
+EXPOSE 8300 8301 8302 8303 8304 8305
+EXPOSE 9092 9093
 
 WORKDIR /var/lib/druid
-ENTRYPOINT export HOST_IP="$(resolveip -s $HOSTNAME)" && exec /usr/bin/supervisord -c /etc/supervisor/conf.d/supervisord.conf
+ENTRYPOINT export HOST_IP="$(resolveip -s $HOSTNAME)" && /tls/generate-server-certs-and-keystores.sh && exec /usr/bin/supervisord -c /etc/supervisor/conf.d/supervisord.conf
diff --git a/integration-tests/docker/broker.conf b/integration-tests/docker/broker.conf
index 4510c6ec692..a83eb1cd106 100644
--- a/integration-tests/docker/broker.conf
+++ b/integration-tests/docker/broker.conf
@@ -36,6 +36,26 @@ command=java
   -Ddruid.auth.authorizer.basic.type=basic
   -Ddruid.sql.enable=true
   -Ddruid.sql.avatica.enable=true
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
   -cp /shared/docker/lib/*
   org.apache.druid.cli.Main server broker
 redirect_stderr=true
diff --git a/integration-tests/docker/coordinator.conf b/integration-tests/docker/coordinator.conf
index 099abf52838..d6b6b591591 100644
--- a/integration-tests/docker/coordinator.conf
+++ b/integration-tests/docker/coordinator.conf
@@ -9,6 +9,7 @@ command=java
   -Duser.timezone=UTC
   -Dfile.encoding=UTF-8
   -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.server.http.numThreads=100
   -Ddruid.metadata.storage.type=mysql
   -Ddruid.metadata.storage.connector.connectURI=jdbc:mysql://druid-metadata-storage/druid
   -Ddruid.metadata.storage.connector.user=druid
@@ -29,6 +30,26 @@ command=java
   -Ddruid.auth.authorizers="[\"basic\"]"
   -Ddruid.auth.authorizer.basic.type=basic
   -Ddruid.auth.unsecuredPaths="[\"/druid/coordinator/v1/loadqueue\"]"
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
   -cp /shared/docker/lib/*
   org.apache.druid.cli.Main server coordinator
 redirect_stderr=true
diff --git a/integration-tests/docker/historical.conf b/integration-tests/docker/historical.conf
index 6ff564c1306..5b3fd0e6024 100644
--- a/integration-tests/docker/historical.conf
+++ b/integration-tests/docker/historical.conf
@@ -32,6 +32,26 @@ command=java
   -Ddruid.escalator.authorizerName=basic
   -Ddruid.auth.authorizers="[\"basic\"]"
   -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
   -cp /shared/docker/lib/*
   org.apache.druid.cli.Main server historical
 redirect_stderr=true
diff --git a/integration-tests/docker/metadata-storage.conf b/integration-tests/docker/metadata-storage.conf
index eb60e214665..828377403dd 100644
--- a/integration-tests/docker/metadata-storage.conf
+++ b/integration-tests/docker/metadata-storage.conf
@@ -1,6 +1,5 @@
 [program:mysql]
-command=/usr/bin/pidproxy /var/run/mysqld/mysqld.pid /usr/bin/mysqld_safe
-         --bind-address=0.0.0.0
+command=/run-mysql.sh
 user=mysql
 priority=0
 stdout_logfile=/shared/logs/mysql.log
diff --git a/integration-tests/docker/middlemanager.conf b/integration-tests/docker/middlemanager.conf
index e2f4f06ebb5..cf09ae82006 100644
--- a/integration-tests/docker/middlemanager.conf
+++ b/integration-tests/docker/middlemanager.conf
@@ -9,11 +9,12 @@ command=java
   -Duser.timezone=UTC
   -Dfile.encoding=UTF-8
   -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.server.http.numThreads=100
   -Ddruid.zk.service.host=druid-zookeeper-kafka
   -Ddruid.worker.capacity=3
   -Ddruid.indexer.logs.directory=/shared/tasklogs
   -Ddruid.storage.storageDirectory=/shared/storage
-  -Ddruid.indexer.runner.javaOpts=-server -Xmx256m -Xms256m -XX:NewSize=128m -XX:MaxNewSize=128m -XX:+UseConcMarkSweepGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps
+  -Ddruid.indexer.runner.javaOpts="-server -Xmx256m -Xms256m -XX:NewSize=128m -XX:MaxNewSize=128m -XX:+UseConcMarkSweepGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps"
   -Ddruid.indexer.fork.property.druid.processing.buffer.sizeBytes=25000000
   -Ddruid.indexer.fork.property.druid.processing.numThreads=1
   -Ddruid.indexer.fork.server.http.numThreads=100
@@ -35,6 +36,27 @@ command=java
   -Ddruid.escalator.authorizerName=basic
   -Ddruid.auth.authorizers="[\"basic\"]"
   -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
+  -Ddruid.startup.logging.logProperties=true
   -cp /shared/docker/lib/*
   org.apache.druid.cli.Main server middleManager
 redirect_stderr=true
diff --git a/integration-tests/docker/overlord.conf b/integration-tests/docker/overlord.conf
index 97a48ccaeb4..293835a08a7 100644
--- a/integration-tests/docker/overlord.conf
+++ b/integration-tests/docker/overlord.conf
@@ -9,6 +9,7 @@ command=java
   -Duser.timezone=UTC
   -Dfile.encoding=UTF-8
   -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.server.http.numThreads=100
   -Ddruid.metadata.storage.type=mysql
   -Ddruid.metadata.storage.connector.connectURI=jdbc:mysql://druid-metadata-storage/druid
   -Ddruid.metadata.storage.connector.user=druid
@@ -30,6 +31,26 @@ command=java
   -Ddruid.escalator.authorizerName=basic
   -Ddruid.auth.authorizers="[\"basic\"]"
   -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
   -cp /shared/docker/lib/*
   org.apache.druid.cli.Main server overlord
 redirect_stderr=true
diff --git a/integration-tests/docker/router-no-client-auth-tls.conf b/integration-tests/docker/router-no-client-auth-tls.conf
new file mode 100644
index 00000000000..f947aa35d54
--- /dev/null
+++ b/integration-tests/docker/router-no-client-auth-tls.conf
@@ -0,0 +1,54 @@
+[program:druid-router-no-client-auth-tls]
+command=java
+  -server
+  -Xmx128m
+  -XX:+UseConcMarkSweepGC
+  -XX:+PrintGCDetails
+  -XX:+PrintGCTimeStamps
+  -Duser.timezone=UTC
+  -Dfile.encoding=UTF-8
+  -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.plaintextPort=8890
+  -Ddruid.tlsPort=9090
+  -Ddruid.zk.service.host=druid-zookeeper-kafka
+  -Ddruid.server.http.numThreads=100
+  -Ddruid.lookup.numLookupLoadingThreads=1
+  -Ddruid.auth.authenticatorChain="[\"basic\"]"
+  -Ddruid.auth.authenticator.basic.type=basic
+  -Ddruid.auth.authenticator.basic.initialAdminPassword=priest
+  -Ddruid.auth.authenticator.basic.initialInternalClientPassword=warlock
+  -Ddruid.auth.authenticator.basic.authorizerName=basic
+  -Ddruid.auth.basic.common.cacheDirectory=/tmp/authCache/router-no-client-auth-tls
+  -Ddruid.escalator.type=basic
+  -Ddruid.escalator.internalClientUsername=druid_system
+  -Ddruid.escalator.internalClientPassword=warlock
+  -Ddruid.escalator.authorizerName=basic
+  -Ddruid.auth.authorizers="[\"basic\"]"
+  -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.sql.enable=true
+  -Ddruid.sql.avatica.enable=true
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=false
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=false
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
+  -cp /shared/docker/lib/*
+  org.apache.druid.cli.Main server router
+redirect_stderr=true
+priority=100
+autorestart=false
+stdout_logfile=/shared/logs/router-no-client-auth-tls.log
diff --git a/integration-tests/docker/router-permissive-tls.conf b/integration-tests/docker/router-permissive-tls.conf
new file mode 100644
index 00000000000..1b6e1591643
--- /dev/null
+++ b/integration-tests/docker/router-permissive-tls.conf
@@ -0,0 +1,54 @@
+[program:druid-router-permissive-tls]
+command=java
+  -server
+  -Xmx128m
+  -XX:+UseConcMarkSweepGC
+  -XX:+PrintGCDetails
+  -XX:+PrintGCTimeStamps
+  -Duser.timezone=UTC
+  -Dfile.encoding=UTF-8
+  -Ddruid.host=%(ENV_HOST_IP)s
+  -Ddruid.plaintextPort=8889
+  -Ddruid.tlsPort=9089
+  -Ddruid.zk.service.host=druid-zookeeper-kafka
+  -Ddruid.server.http.numThreads=100
+  -Ddruid.lookup.numLookupLoadingThreads=1
+  -Ddruid.auth.authenticatorChain="[\"basic\"]"
+  -Ddruid.auth.authenticator.basic.type=basic
+  -Ddruid.auth.authenticator.basic.initialAdminPassword=priest
+  -Ddruid.auth.authenticator.basic.initialInternalClientPassword=warlock
+  -Ddruid.auth.authenticator.basic.authorizerName=basic
+  -Ddruid.auth.basic.common.cacheDirectory=/tmp/authCache/router-permissive-tls
+  -Ddruid.escalator.type=basic
+  -Ddruid.escalator.internalClientUsername=druid_system
+  -Ddruid.escalator.internalClientPassword=warlock
+  -Ddruid.escalator.authorizerName=basic
+  -Ddruid.auth.authorizers="[\"basic\"]"
+  -Ddruid.auth.authorizer.basic.type=basic
+  -Ddruid.sql.enable=true
+  -Ddruid.sql.avatica.enable=true
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=false
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
+  -cp /shared/docker/lib/*
+  org.apache.druid.cli.Main server router
+redirect_stderr=true
+priority=100
+autorestart=false
+stdout_logfile=/shared/logs/router-permissive-tls.log
diff --git a/integration-tests/docker/router.conf b/integration-tests/docker/router.conf
index 39923d9a8d2..a65c25ef031 100644
--- a/integration-tests/docker/router.conf
+++ b/integration-tests/docker/router.conf
@@ -25,6 +25,26 @@ command=java
   -Ddruid.auth.authorizer.basic.type=basic
   -Ddruid.sql.enable=true
   -Ddruid.sql.avatica.enable=true
+  -Ddruid.enableTlsPort=true
+  -Ddruid.server.https.keyStorePath=/tls/server.jks
+  -Ddruid.server.https.keyStoreType=jks
+  -Ddruid.server.https.certAlias=druid
+  -Ddruid.server.https.keyManagerPassword=druid123
+  -Ddruid.server.https.keyStorePassword=druid123
+  -Ddruid.server.https.requireClientCertificate=true
+  -Ddruid.server.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.server.https.trustStorePassword=druid123
+  -Ddruid.server.https.trustStoreAlgorithm=PKIX
+  -Ddruid.server.https.validateHostnames=true
+  -Ddruid.server.https.crlPath=/tls/revocations.crl
+  -Ddruid.client.https.trustStoreAlgorithm=PKIX
+  -Ddruid.client.https.protocol=TLSv1.2
+  -Ddruid.client.https.trustStorePath=/tls/truststore.jks
+  -Ddruid.client.https.trustStorePassword=druid123
+  -Ddruid.client.https.keyStorePath=/tls/server.jks
+  -Ddruid.client.https.certAlias=druid
+  -Ddruid.client.https.keyManagerPassword=druid123
+  -Ddruid.client.https.keyStorePassword=druid123
   -cp /shared/docker/lib/*
   org.apache.druid.cli.Main server router
 redirect_stderr=true
diff --git a/integration-tests/docker/run-mysql.sh b/integration-tests/docker/run-mysql.sh
new file mode 100755
index 00000000000..266b6526bb7
--- /dev/null
+++ b/integration-tests/docker/run-mysql.sh
@@ -0,0 +1,3 @@
+#!/bin/bash
+
+find /var/lib/mysql -type f -exec touch {} \; && /usr/bin/pidproxy /var/run/mysqld/mysqld.pid /usr/bin/mysqld_safe --bind-address=0.0.0.0
\ No newline at end of file
diff --git a/integration-tests/docker/tls/generate-client-certs-and-keystores.sh b/integration-tests/docker/tls/generate-client-certs-and-keystores.sh
new file mode 100755
index 00000000000..0fad8012b79
--- /dev/null
+++ b/integration-tests/docker/tls/generate-client-certs-and-keystores.sh
@@ -0,0 +1,21 @@
+#!/bin/bash -eu
+
+./docker/tls/generate-root-certs.sh
+
+mkdir -p client_tls
+rm -f client_tls/*
+cp docker/tls/root.key client_tls/root.key
+cp docker/tls/root.pem client_tls/root.pem
+cp docker/tls/untrusted_root.key client_tls/untrusted_root.key
+cp docker/tls/untrusted_root.pem client_tls/untrusted_root.pem
+cd client_tls
+
+../docker/tls/generate-expired-client-cert.sh
+../docker/tls/generate-good-client-cert.sh
+../docker/tls/generate-incorrect-hostname-client-cert.sh
+../docker/tls/generate-invalid-intermediate-client-cert.sh
+../docker/tls/generate-to-be-revoked-client-cert.sh
+../docker/tls/generate-untrusted-root-client-cert.sh
+../docker/tls/generate-valid-intermediate-client-cert.sh
+
+
diff --git a/integration-tests/docker/tls/generate-expired-client-cert.sh b/integration-tests/docker/tls/generate-expired-client-cert.sh
new file mode 100755
index 00000000000..f4d7236a6d0
--- /dev/null
+++ b/integration-tests/docker/tls/generate-expired-client-cert.sh
@@ -0,0 +1,41 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > expired_csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out expired_client.key 1024 -sha256
+openssl req -new -out expired_client.csr -key expired_client.key -reqexts req_ext -config expired_csr.conf
+openssl x509 -req -days -3650 -in expired_client.csr -CA root.pem -CAkey root.key -set_serial 0x11111115 -out expired_client.pem -sha256 -extfile expired_csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in expired_client.pem -inkey expired_client.key -out expired_client.p12 -name expired_client -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore expired_client.p12 -srcstoretype PKCS12 -destkeystore expired_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
diff --git a/integration-tests/docker/tls/generate-good-client-cert.sh b/integration-tests/docker/tls/generate-good-client-cert.sh
new file mode 100755
index 00000000000..96a1bc0aaf8
--- /dev/null
+++ b/integration-tests/docker/tls/generate-good-client-cert.sh
@@ -0,0 +1,44 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out client.key 1024 -sha256
+openssl req -new -out client.csr -key client.key -reqexts req_ext -config csr.conf
+openssl x509 -req -days 3650 -in client.csr -CA root.pem -CAkey root.key -set_serial 0x11111111 -out client.pem -sha256 -extfile csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in client.pem -inkey client.key -out client.p12 -name druid -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore client.p12 -srcstoretype PKCS12 -destkeystore client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
+
+# Create a Java truststore with the imply test cluster root CA
+keytool -import -alias druid-it-root -keystore truststore.jks -file root.pem -storepass druid123 -noprompt
diff --git a/integration-tests/docker/tls/generate-incorrect-hostname-client-cert.sh b/integration-tests/docker/tls/generate-incorrect-hostname-client-cert.sh
new file mode 100755
index 00000000000..5cb3022468e
--- /dev/null
+++ b/integration-tests/docker/tls/generate-incorrect-hostname-client-cert.sh
@@ -0,0 +1,40 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+
+# Generate a client cert with an incorrect hostname for testing
+cat <<EOT > invalid_hostname_csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = thisisprobablynottherighthostname
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+DNS.1 = thisisprobablywrongtoo
+
+EOT
+
+openssl genrsa -out invalid_hostname_client.key 1024 -sha256
+openssl req -new -out invalid_hostname_client.csr -key invalid_hostname_client.key -reqexts req_ext -config invalid_hostname_csr.conf
+openssl x509 -req -days 3650 -in invalid_hostname_client.csr -CA root.pem -CAkey root.key -set_serial 0x11111112 -out invalid_hostname_client.pem -sha256 -extfile invalid_hostname_csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in invalid_hostname_client.pem -inkey invalid_hostname_client.key -out invalid_hostname_client.p12 -name invalid_hostname_client -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore invalid_hostname_client.p12 -srcstoretype PKCS12 -destkeystore invalid_hostname_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
+
diff --git a/integration-tests/docker/tls/generate-invalid-intermediate-client-cert.sh b/integration-tests/docker/tls/generate-invalid-intermediate-client-cert.sh
new file mode 100755
index 00000000000..134898b27c7
--- /dev/null
+++ b/integration-tests/docker/tls/generate-invalid-intermediate-client-cert.sh
@@ -0,0 +1,76 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > invalid_ca_intermediate.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=bad-intermediate@druid.io
+CN = badintermediate
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = 9.9.9.9
+EOT
+
+# Generate a bad intermediate certificate
+openssl genrsa -out invalid_ca_intermediate.key 1024 -sha256
+openssl req -new -out invalid_ca_intermediate.csr -key invalid_ca_intermediate.key -reqexts req_ext -config invalid_ca_intermediate.conf
+openssl x509 -req -days 3650 -in invalid_ca_intermediate.csr -CA root.pem -CAkey root.key -set_serial 0x33333331 -out invalid_ca_intermediate.pem -sha256 -extfile invalid_ca_intermediate.conf -extensions req_ext
+
+
+cat <<EOT > invalid_ca_client.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=basic-constraint-fail@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out invalid_ca_client.key 1024 -sha256
+openssl req -new -out invalid_ca_client.csr -key invalid_ca_client.key -reqexts req_ext -config invalid_ca_client.conf
+openssl x509 -req -days 3650 -in invalid_ca_client.csr -CA invalid_ca_intermediate.pem -CAkey invalid_ca_intermediate.key -set_serial 0x33333333 -out invalid_ca_client.pem -sha256 -extfile invalid_ca_client.conf -extensions req_ext
+
+# Append the signing cert
+printf "\n" >> invalid_ca_client.pem
+cat invalid_ca_intermediate.pem >> invalid_ca_client.pem
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in invalid_ca_client.pem -inkey invalid_ca_client.key -out invalid_ca_client.p12 -name invalid_ca_client -CAfile invalid_ca_intermediate.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore invalid_ca_client.p12 -srcstoretype PKCS12 -destkeystore invalid_ca_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
\ No newline at end of file
diff --git a/integration-tests/docker/tls/generate-root-certs.sh b/integration-tests/docker/tls/generate-root-certs.sh
new file mode 100755
index 00000000000..8b14f2783bd
--- /dev/null
+++ b/integration-tests/docker/tls/generate-root-certs.sh
@@ -0,0 +1,13 @@
+#!/bin/bash -eu
+
+rm -f root.key
+rm -f untrusted_root.key
+rm -f root.pem
+rm -f untrusted_root.pem
+
+openssl genrsa -out docker/tls/root.key 4096
+openssl genrsa -out docker/tls/untrusted_root.key 4096
+
+openssl req -config docker/tls/root.cnf -key docker/tls/root.key -new -x509 -days 3650 -sha256 -extensions v3_ca -out docker/tls/root.pem
+openssl req -config docker/tls/root.cnf -key docker/tls/untrusted_root.key -new -x509 -days 3650 -sha256 -extensions v3_ca -out docker/tls/untrusted_root.pem
+
diff --git a/integration-tests/docker/tls/generate-server-certs-and-keystores.sh b/integration-tests/docker/tls/generate-server-certs-and-keystores.sh
new file mode 100755
index 00000000000..940d44436fc
--- /dev/null
+++ b/integration-tests/docker/tls/generate-server-certs-and-keystores.sh
@@ -0,0 +1,66 @@
+#!/bin/bash -eu
+
+cd /tls
+
+rm -f cert_db.txt
+touch cert_db.txt
+
+export DOCKER_IP=$(cat /docker_ip)
+export MY_HOSTNAME=$(hostname)
+export MY_IP=$(hostname -i)
+
+cat <<EOT > csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = ${MY_IP}
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_IP}
+IP.2 = ${MY_IP}
+IP.3 = 127.0.0.1
+DNS.1 = ${MY_HOSTNAME}
+DNS.2 = localhost
+
+EOT
+
+# Generate a server certificate for this machine
+openssl genrsa -out server.key 1024 -sha256
+openssl req -new -out server.csr -key server.key -reqexts req_ext -config csr.conf
+openssl x509 -req -days 3650 -in server.csr -CA root.pem -CAkey root.key -set_serial 0x22222222 -out server.pem -sha256 -extfile csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in server.pem -inkey server.key -out server.p12 -name druid -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore server.p12 -srcstoretype PKCS12 -destkeystore server.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
+
+# Create a Java truststore with the imply test cluster root CA
+keytool -import -alias druid-it-root -keystore truststore.jks -file root.pem -storepass druid123 -noprompt
+
+# Revoke one of the client certs
+openssl ca -revoke /client_tls/revoked_client.pem -config root.cnf -cert root.pem -keyfile root.key
+
+# Create the CRL
+openssl ca -gencrl -config root.cnf -cert root.pem -keyfile root.key -out /tls/revocations.crl
+
+# Generate empty CRLs for the intermediate cert test case
+rm -f cert_db2.txt
+touch cert_db2.txt
+openssl ca -gencrl -config root2.cnf -cert /client_tls/ca_intermediate.pem -keyfile /client_tls/ca_intermediate.key -out /tls/empty-revocations-intermediate.crl
+
+# Append CRLs
+cat empty-revocations-intermediate.crl >> revocations.crl
diff --git a/integration-tests/docker/tls/generate-to-be-revoked-client-cert.sh b/integration-tests/docker/tls/generate-to-be-revoked-client-cert.sh
new file mode 100755
index 00000000000..ecb49bd640a
--- /dev/null
+++ b/integration-tests/docker/tls/generate-to-be-revoked-client-cert.sh
@@ -0,0 +1,43 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+# Generate a client cert that will be revoked
+cat <<EOT > revoked_csr.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=RevokedIntegrationTests
+emailAddress=revoked-it-cert@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out revoked_client.key 1024 -sha256
+openssl req -new -out revoked_client.csr -key revoked_client.key -reqexts req_ext -config revoked_csr.conf
+openssl x509 -req -days 3650 -in revoked_client.csr -CA root.pem -CAkey root.key -set_serial 0x11111113 -out revoked_client.pem -sha256 -extfile csr.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in revoked_client.pem -inkey revoked_client.key -out revoked_client.p12 -name revoked_druid -CAfile root.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore revoked_client.p12 -srcstoretype PKCS12 -destkeystore revoked_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
diff --git a/integration-tests/docker/tls/generate-untrusted-root-client-cert.sh b/integration-tests/docker/tls/generate-untrusted-root-client-cert.sh
new file mode 100755
index 00000000000..08ff13d1d98
--- /dev/null
+++ b/integration-tests/docker/tls/generate-untrusted-root-client-cert.sh
@@ -0,0 +1,42 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > csr_another_root.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out client_another_root.key 1024 -sha256
+openssl req -new -out client_another_root.csr -key client_another_root.key -reqexts req_ext -config csr_another_root.conf
+openssl x509 -req -days 3650 -in client_another_root.csr -CA untrusted_root.pem -CAkey untrusted_root.key -set_serial 0x11111114 -out client_another_root.pem -sha256 -extfile csr_another_root.conf -extensions req_ext
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in client_another_root.pem -inkey client_another_root.key -out client_another_root.p12 -name druid_another_root -CAfile untrusted_root.pem -caname druid-it-untrusted-root -password pass:druid123
+keytool -importkeystore -srckeystore client_another_root.p12 -srcstoretype PKCS12 -destkeystore client_another_root.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
+
diff --git a/integration-tests/docker/tls/generate-valid-intermediate-client-cert.sh b/integration-tests/docker/tls/generate-valid-intermediate-client-cert.sh
new file mode 100755
index 00000000000..d76a4fc2c90
--- /dev/null
+++ b/integration-tests/docker/tls/generate-valid-intermediate-client-cert.sh
@@ -0,0 +1,76 @@
+#!/bin/bash -eu
+
+export DOCKER_HOST_IP=$(resolveip -s $HOSTNAME)
+
+cat <<EOT > ca_intermediate.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=intermediate@druid.io
+CN = intermediate
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:TRUE,pathlen:1
+
+[ alt_names ]
+IP.1 = 9.9.9.9
+EOT
+
+# Generate an intermediate certificate
+openssl genrsa -out ca_intermediate.key 1024 -sha256
+openssl req -new -out ca_intermediate.csr -key ca_intermediate.key -reqexts req_ext -config ca_intermediate.conf
+openssl x509 -req -days 3650 -in ca_intermediate.csr -CA root.pem -CAkey root.key -set_serial 0x33333332 -out ca_intermediate.pem -sha256 -extfile ca_intermediate.conf -extensions req_ext
+
+
+cat <<EOT > intermediate_ca_client.conf
+[req]
+default_bits = 1024
+prompt = no
+default_md = sha256
+req_extensions = req_ext
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=intermediate-client@druid.io
+CN = localhost
+
+[ req_ext ]
+subjectAltName = @alt_names
+basicConstraints=CA:FALSE,pathlen:0
+
+[ alt_names ]
+IP.1 = ${DOCKER_HOST_IP}
+IP.2 = 127.0.0.1
+IP.3 = 172.172.172.1
+DNS.1 = ${HOSTNAME}
+DNS.2 = localhost
+EOT
+
+# Generate a client certificate for this machine
+openssl genrsa -out intermediate_ca_client.key 1024 -sha256
+openssl req -new -out intermediate_ca_client.csr -key intermediate_ca_client.key -reqexts req_ext -config intermediate_ca_client.conf
+openssl x509 -req -days 3650 -in intermediate_ca_client.csr -CA ca_intermediate.pem -CAkey ca_intermediate.key -set_serial 0x33333333 -out intermediate_ca_client.pem -sha256 -extfile intermediate_ca_client.conf -extensions req_ext
+
+# Append the signing cert
+printf "\n" >> intermediate_ca_client.pem
+cat ca_intermediate.pem >> intermediate_ca_client.pem
+
+# Create a Java keystore containing the generated certificate
+openssl pkcs12 -export -in intermediate_ca_client.pem -inkey intermediate_ca_client.key -out intermediate_ca_client.p12 -name intermediate_ca_client -CAfile ca_intermediate.pem -caname druid-it-root -password pass:druid123
+keytool -importkeystore -srckeystore intermediate_ca_client.p12 -srcstoretype PKCS12 -destkeystore intermediate_ca_client.jks -deststoretype JKS -srcstorepass druid123 -deststorepass druid123
\ No newline at end of file
diff --git a/integration-tests/docker/tls/root.cnf b/integration-tests/docker/tls/root.cnf
new file mode 100644
index 00000000000..dd664bf0ca8
--- /dev/null
+++ b/integration-tests/docker/tls/root.cnf
@@ -0,0 +1,35 @@
+[ ca ]
+default_ca = CA_default
+
+[ CA_default ]
+database = /tls/cert_db.txt
+x509_extensions	= usr_cert
+name_opt = ca_default
+cert_opt = ca_default
+default_days = 365
+default_crl_days= 30
+default_md = default
+preserve = no
+policy = policy_match
+
+[req]
+default_bits = 4096
+prompt = no
+default_md = sha256
+req_extensions = v3_ca
+distinguished_name = dn
+
+[ dn ]
+C=DR
+ST=DR
+L=Druid City
+O=Druid
+OU=IntegrationTests
+emailAddress=integration-test@druid.io
+CN = itroot
+
+[ v3_ca ]
+subjectKeyIdentifier = hash
+authorityKeyIdentifier = keyid:always,issuer
+basicConstraints = critical, CA:true
+keyUsage = critical, digitalSignature, cRLSign, keyCertSign
\ No newline at end of file
diff --git a/integration-tests/docker/tls/root2.cnf b/integration-tests/docker/tls/root2.cnf
new file mode 100644
index 00000000000..5173a872c67
--- /dev/null
+++ b/integration-tests/docker/tls/root2.cnf
@@ -0,0 +1,13 @@
+[ ca ]
+default_ca = CA_default
+
+[ CA_default ]
+database = /tls/cert_db2.txt
+x509_extensions	= usr_cert
+name_opt = ca_default
+cert_opt = ca_default
+default_days = 365
+default_crl_days= 30
+default_md = default
+preserve = no
+policy = policy_match
diff --git a/integration-tests/pom.xml b/integration-tests/pom.xml
index 9c67b3ec147..2682e76ab0c 100644
--- a/integration-tests/pom.xml
+++ b/integration-tests/pom.xml
@@ -88,6 +88,11 @@
             <artifactId>druid-basic-security</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.druid.extensions</groupId>
+            <artifactId>simple-client-sslcontext</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.druid</groupId>
             <artifactId>druid-services</artifactId>
@@ -227,6 +232,12 @@
                                 -Ddruid.test.config.dockerIp=${env.DOCKER_IP}
                                 -Ddruid.test.config.hadoopDir=${env.HADOOP_DIR}
                                 -Ddruid.zk.service.host=${env.DOCKER_IP}
+                                -Ddruid.client.https.trustStorePath=client_tls/truststore.jks
+                                -Ddruid.client.https.trustStorePassword=druid123
+                                -Ddruid.client.https.keyStorePath=client_tls/client.jks
+                                -Ddruid.client.https.certAlias=druid
+                                -Ddruid.client.https.keyManagerPassword=druid123
+                                -Ddruid.client.https.keyStorePassword=druid123
                             </argLine>
                             <suiteXmlFiles>
                                 <suiteXmlFile>src/test/resources/testng.xml</suiteXmlFile>
@@ -276,6 +287,12 @@
                                 -Dfile.encoding=UTF-8
                                 -Ddruid.test.config.type=configFile
                                 -Ddruid.test.config.configFile=${env.CONFIG_FILE}
+                                -Ddruid.client.https.trustStorePath=client_tls/truststore.jks
+                                -Ddruid.client.https.trustStorePassword=druid123
+                                -Ddruid.client.https.keyStorePath=client_tls/client.jks
+                                -Ddruid.client.https.certAlias=druid
+                                -Ddruid.client.https.keyManagerPassword=druid123
+                                -Ddruid.client.https.keyStorePassword=druid123
                             </argLine>
                             <suiteXmlFiles>
                                 <suiteXmlFile>src/test/resources/testng.xml</suiteXmlFile>
diff --git a/integration-tests/run_cluster.sh b/integration-tests/run_cluster.sh
index e5d3a259009..a41a5b8b0ea 100755
--- a/integration-tests/run_cluster.sh
+++ b/integration-tests/run_cluster.sh
@@ -15,12 +15,14 @@
 # limitations under the License.
 
 # cleanup
-for node in druid-historical druid-coordinator druid-overlord druid-router druid-broker druid-middlemanager druid-zookeeper-kafka druid-metadata-storage;
+for node in druid-historical druid-coordinator druid-overlord druid-router druid-router-permissive-tls druid-router-no-client-auth-tls druid-broker druid-middlemanager druid-zookeeper-kafka druid-metadata-storage;
 do
 docker stop $node
 docker rm $node
 done
 
+docker network rm druid-it-net
+
 # environment variables
 DIR=$(cd $(dirname $0) && pwd)
 DOCKERDIR=$DIR/docker
@@ -31,6 +33,11 @@ RESOURCEDIR=$DIR/src/test/resources
 # so docker IP addr will be known during docker build
 echo ${DOCKER_IP:=127.0.0.1} > $DOCKERDIR/docker_ip
 
+# setup client keystore
+./docker/tls/generate-client-certs-and-keystores.sh
+rm -rf docker/client_tls
+cp -r client_tls docker/client_tls
+
 # Make directories if they dont exist
 mkdir -p $SHARED_DIR/logs
 mkdir -p $SHARED_DIR/tasklogs
@@ -40,29 +47,37 @@ rm -rf $SHARED_DIR/docker
 cp -R docker $SHARED_DIR/docker
 mvn -B dependency:copy-dependencies -DoutputDirectory=$SHARED_DIR/docker/lib
 
+docker network create --subnet=172.172.172.0/24 druid-it-net
+
 # Build Druid Cluster Image
 docker build -t druid/cluster $SHARED_DIR/docker
 
 # Start zookeeper and kafka
-docker run -d --privileged --name druid-zookeeper-kafka -p 2181:2181 -p 9092:9092 -p 9093:9093 -v $SHARED_DIR:/shared -v $DOCKERDIR/zookeeper.conf:$SUPERVISORDIR/zookeeper.conf -v $DOCKERDIR/kafka.conf:$SUPERVISORDIR/kafka.conf druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.2 --name druid-zookeeper-kafka -p 2181:2181 -p 9092:9092 -p 9093:9093 -v $SHARED_DIR:/shared -v $DOCKERDIR/zookeeper.conf:$SUPERVISORDIR/zookeeper.conf -v $DOCKERDIR/kafka.conf:$SUPERVISORDIR/kafka.conf druid/cluster
 
 # Start MYSQL 
-docker run -d --privileged --name druid-metadata-storage -v $SHARED_DIR:/shared -v $DOCKERDIR/metadata-storage.conf:$SUPERVISORDIR/metadata-storage.conf druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.3 --name druid-metadata-storage -v $SHARED_DIR:/shared -v $DOCKERDIR/metadata-storage.conf:$SUPERVISORDIR/metadata-storage.conf druid/cluster
 
 # Start Overlord
-docker run -d --privileged --name druid-overlord -p 8090:8090 -v $SHARED_DIR:/shared -v $DOCKERDIR/overlord.conf:$SUPERVISORDIR/overlord.conf --link druid-metadata-storage:druid-metadata-storage --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.4 --name druid-overlord -p 8090:8090 -p 8290:8290 -v $SHARED_DIR:/shared -v $DOCKERDIR/overlord.conf:$SUPERVISORDIR/overlord.conf --link druid-metadata-storage:druid-metadata-storage --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster
 
 # Start Coordinator
-docker run -d --privileged --name druid-coordinator -p 8081:8081 -v $SHARED_DIR:/shared -v $DOCKERDIR/coordinator.conf:$SUPERVISORDIR/coordinator.conf --link druid-overlord:druid-overlord --link druid-metadata-storage:druid-metadata-storage --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.5 --name druid-coordinator -p 8081:8081 -p 8281:8281 -v $SHARED_DIR:/shared -v $DOCKERDIR/coordinator.conf:$SUPERVISORDIR/coordinator.conf --link druid-overlord:druid-overlord --link druid-metadata-storage:druid-metadata-storage --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster
 
 # Start Historical 
-docker run -d --privileged --name druid-historical -p 8083:8083 -v $SHARED_DIR:/shared -v $DOCKERDIR/historical.conf:$SUPERVISORDIR/historical.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.6 --name druid-historical -p 8083:8083 -p 8283:8283 -v $SHARED_DIR:/shared -v $DOCKERDIR/historical.conf:$SUPERVISORDIR/historical.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka druid/cluster
 
 # Start Middlemanger
-docker run -d --privileged --name druid-middlemanager -p 8091:8091 -p 8100:8100 -p 8101:8101 -p 8102:8102 -p 8103:8103 -p 8104:8104 -p 8105:8105 -v $RESOURCEDIR:/resources -v $SHARED_DIR:/shared -v $DOCKERDIR/middlemanager.conf:$SUPERVISORDIR/middlemanager.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-overlord:druid-overlord druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.7 --name druid-middlemanager -p 8091:8091 -p 8291:8291 -p 8100:8100 -p 8101:8101 -p 8102:8102 -p 8103:8103 -p 8104:8104 -p 8105:8105 -p 8300:8300 -p 8301:8301 -p 8302:8302 -p 8303:8303 -p 8304:8304 -p 8305:8305 -v $RESOURCEDIR:/resources -v $SHARED_DIR:/shared -v $DOCKERDIR/middlemanager.conf:$SUPERVISORDIR/middlemanager.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-overlord:druid-overlord druid/cluster
 
 # Start Broker 
-docker run -d --privileged --name druid-broker -p 8082:8082 -v $SHARED_DIR:/shared -v $DOCKERDIR/broker.conf:$SUPERVISORDIR/broker.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-middlemanager:druid-middlemanager --link druid-historical:druid-historical druid/cluster
+docker run -d --privileged --net druid-it-net --ip 172.172.172.8 --name druid-broker -p 8082:8082 -p 8282:8282 -v $SHARED_DIR:/shared -v $DOCKERDIR/broker.conf:$SUPERVISORDIR/broker.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-middlemanager:druid-middlemanager --link druid-historical:druid-historical druid/cluster
 
-# Start Router 
-docker run -d --privileged --name druid-router -p 8888:8888 -v $SHARED_DIR:/shared -v $DOCKERDIR/router.conf:$SUPERVISORDIR/router.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-coordinator:druid-coordinator --link druid-broker:druid-broker druid/cluster
+# Start Router
+docker run -d --privileged --net druid-it-net --ip 172.172.172.9 --name druid-router -p 8888:8888 -p 9088:9088 -v $SHARED_DIR:/shared -v $DOCKERDIR/router.conf:$SUPERVISORDIR/router.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-coordinator:druid-coordinator --link druid-broker:druid-broker druid/cluster
+
+# Start Router with permissive TLS settings (client auth enabled, no hostname verification, no revocation check)
+docker run -d --privileged --net druid-it-net --ip 172.172.172.10 --name druid-router-permissive-tls -p 8889:8889 -p 9089:9089 -v $SHARED_DIR:/shared -v $DOCKERDIR/router-permissive-tls.conf:$SUPERVISORDIR/router-permissive-tls.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-coordinator:druid-coordinator --link druid-broker:druid-broker druid/cluster
+
+# Start Router with TLS but no client auth
+docker run -d --privileged --net druid-it-net --ip 172.172.172.11 --name druid-router-no-client-auth-tls -p 8890:8890 -p 9090:9090 -v $SHARED_DIR:/shared -v $DOCKERDIR/router-no-client-auth-tls.conf:$SUPERVISORDIR/router-no-client-auth-tls.conf --link druid-zookeeper-kafka:druid-zookeeper-kafka --link druid-coordinator:druid-coordinator --link druid-broker:druid-broker druid/cluster
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/ConfigFileConfigProvider.java b/integration-tests/src/main/java/org/apache/druid/testing/ConfigFileConfigProvider.java
index 584e1189842..27805278d47 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/ConfigFileConfigProvider.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/ConfigFileConfigProvider.java
@@ -38,6 +38,15 @@ public class ConfigFileConfigProvider implements IntegrationTestingConfigProvide
   private String historicalUrl;
   private String coordinatorUrl;
   private String indexerUrl;
+  private String permissiveRouterUrl;
+  private String noClientAuthRouterUrl;
+  private String routerTLSUrl;
+  private String brokerTLSUrl;
+  private String historicalTLSUrl;
+  private String coordinatorTLSUrl;
+  private String indexerTLSUrl;
+  private String permissiveRouterTLSUrl;
+  private String noClientAuthRouterTLSUrl;
   private String middleManagerHost;
   private String zookeeperHosts;        // comma-separated list of host:port
   private String kafkaHost;
@@ -70,25 +79,89 @@ public class ConfigFileConfigProvider implements IntegrationTestingConfigProvide
         routerUrl = StringUtils.format("http://%s:%s", routerHost, props.get("router_port"));
       }
     }
+    routerTLSUrl = props.get("router_tls_url");
+    if (routerTLSUrl == null) {
+      String routerHost = props.get("router_host");
+      if (null != routerHost) {
+        routerTLSUrl = StringUtils.format("https://%s:%s", routerHost, props.get("router_tls_port"));
+      }
+    }
+    permissiveRouterUrl = props.get("router_permissive_url");
+    if (permissiveRouterUrl == null) {
+      String permissiveRouterHost = props.get("router_permissive_host");
+      if (null != permissiveRouterHost) {
+        permissiveRouterUrl = StringUtils.format("http://%s:%s", permissiveRouterHost, props.get("router_permissive_port"));
+      }
+    }
+    permissiveRouterTLSUrl = props.get("router_permissive_tls_url");
+    if (permissiveRouterTLSUrl == null) {
+      String permissiveRouterHost = props.get("router_permissive_host");
+      if (null != permissiveRouterHost) {
+        permissiveRouterTLSUrl = StringUtils.format("https://%s:%s", permissiveRouterHost, props.get("router_permissive_tls_port"));
+      }
+    }
+    noClientAuthRouterUrl = props.get("router_no_client_auth_url");
+    if (noClientAuthRouterUrl == null) {
+      String noClientAuthRouterHost = props.get("router_no_client_auth_host");
+      if (null != noClientAuthRouterHost) {
+        noClientAuthRouterUrl = StringUtils.format("http://%s:%s", noClientAuthRouterHost, props.get("router_no_client_auth_port"));
+      }
+    }
+    noClientAuthRouterTLSUrl = props.get("router_no_client_auth_tls_url");
+    if (noClientAuthRouterTLSUrl == null) {
+      String noClientAuthRouterHost = props.get("router_no_client_auth_host");
+      if (null != noClientAuthRouterHost) {
+        noClientAuthRouterTLSUrl = StringUtils.format("https://%s:%s", noClientAuthRouterHost, props.get("router_no_client_auth_tls_port"));
+      }
+    }
     brokerUrl = props.get("broker_url");
     if (brokerUrl == null) {
       brokerUrl = StringUtils.format("http://%s:%s", props.get("broker_host"), props.get("broker_port"));
     }
-
+    brokerTLSUrl = props.get("broker_tls_url");
+    if (brokerTLSUrl == null) {
+      String brokerHost = props.get("broker_host");
+      if (null != brokerHost) {
+        brokerTLSUrl = StringUtils.format("https://%s:%s", brokerHost, props.get("broker_tls_port"));
+      }
+    }
+    
     historicalUrl = props.get("historical_url");
     if (historicalUrl == null) {
       historicalUrl = StringUtils.format("http://%s:%s", props.get("historical_host"), props.get("historical_port"));
     }
+    historicalTLSUrl = props.get("historical_tls_url");
+    if (historicalTLSUrl == null) {
+      String historicalHost = props.get("historical_host");
+      if (null != historicalHost) {
+        historicalTLSUrl = StringUtils.format("https://%s:%s", historicalHost, props.get("historical_tls_port"));
+      }
+    }
 
     coordinatorUrl = props.get("coordinator_url");
     if (coordinatorUrl == null) {
       coordinatorUrl = StringUtils.format("http://%s:%s", props.get("coordinator_host"), props.get("coordinator_port"));
     }
+    coordinatorTLSUrl = props.get("coordinator_tls_url");
+    if (coordinatorTLSUrl == null) {
+      String coordinatorHost = props.get("coordinator_host");
+      if (null != coordinatorHost) {
+        coordinatorTLSUrl = StringUtils.format("https://%s:%s", coordinatorHost, props.get("coordinator_tls_port"));
+      }
+    }
 
     indexerUrl = props.get("indexer_url");
     if (indexerUrl == null) {
       indexerUrl = StringUtils.format("http://%s:%s", props.get("indexer_host"), props.get("indexer_port"));
     }
+    indexerTLSUrl = props.get("indexer_tls_url");
+    if (indexerTLSUrl == null) {
+      String indexerHost = props.get("indexer_host");
+      if (null != indexerHost) {
+        indexerTLSUrl = StringUtils.format("https://%s:%s", indexerHost, props.get("indexer_tls_port"));
+      }
+    }
+    
     middleManagerHost = props.get("middlemanager_host");
 
     zookeeperHosts = props.get("zookeeper_hosts");
@@ -98,10 +171,11 @@ public class ConfigFileConfigProvider implements IntegrationTestingConfigProvide
 
     password = props.get("password");
 
-    LOG.info("router: [%s]", routerUrl);
-    LOG.info("broker: [%s]", brokerUrl);
-    LOG.info("coordinator: [%s]", coordinatorUrl);
-    LOG.info("overlord: [%s]", indexerUrl);
+    LOG.info("router: [%s], [%s]", routerUrl, routerTLSUrl);
+    LOG.info("broker: [%s], [%s]", brokerUrl, brokerTLSUrl);
+    LOG.info("historical: [%s], [%s]", historicalUrl, historicalTLSUrl);
+    LOG.info("coordinator: [%s], [%s]", coordinatorUrl, coordinatorTLSUrl);
+    LOG.info("overlord: [%s], [%s]", indexerUrl, indexerTLSUrl);
     LOG.info("middle manager: [%s]", middleManagerHost);
     LOG.info("zookeepers: [%s]", zookeeperHosts);
     LOG.info("kafka: [%s]", kafkaHost);
@@ -120,30 +194,84 @@ public class ConfigFileConfigProvider implements IntegrationTestingConfigProvide
         return coordinatorUrl;
       }
 
+      @Override
+      public String getCoordinatorTLSUrl()
+      {
+        return coordinatorTLSUrl;
+      }
+
       @Override
       public String getIndexerUrl()
       {
         return indexerUrl;
       }
 
+      @Override
+      public String getIndexerTLSUrl()
+      {
+        return indexerTLSUrl;
+      }
+
       @Override
       public String getRouterUrl()
       {
         return routerUrl;
       }
 
+      @Override
+      public String getRouterTLSUrl()
+      {
+        return routerTLSUrl;
+      }
+
+      @Override
+      public String getPermissiveRouterUrl()
+      {
+        return permissiveRouterUrl;
+      }
+
+      @Override
+      public String getPermissiveRouterTLSUrl()
+      {
+        return permissiveRouterTLSUrl;
+      }
+
+      @Override
+      public String getNoClientAuthRouterUrl()
+      {
+        return noClientAuthRouterUrl;
+      }
+
+      @Override
+      public String getNoClientAuthRouterTLSUrl()
+      {
+        return noClientAuthRouterTLSUrl;
+      }
+
       @Override
       public String getBrokerUrl()
       {
         return brokerUrl;
       }
 
+      @Override
+      public String getBrokerTLSUrl()
+      {
+        return brokerTLSUrl;
+      }
+
       @Override
       public String getHistoricalUrl()
       {
         return historicalUrl;
       }
 
+      @Override
+      public String getHistoricalTLSUrl()
+      {
+        return historicalTLSUrl;
+      }
+
       @Override
       public String getMiddleManagerHost()
       {
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/DockerConfigProvider.java b/integration-tests/src/main/java/org/apache/druid/testing/DockerConfigProvider.java
index 3cb333f1b04..bb71bc3bf4c 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/DockerConfigProvider.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/DockerConfigProvider.java
@@ -48,30 +48,84 @@ public class DockerConfigProvider implements IntegrationTestingConfigProvider
         return "http://" + dockerIp + ":8081";
       }
 
+      @Override
+      public String getCoordinatorTLSUrl()
+      {
+        return "https://" + dockerIp + ":8281";
+      }
+
       @Override
       public String getIndexerUrl()
       {
         return "http://" + dockerIp + ":8090";
       }
 
+      @Override
+      public String getIndexerTLSUrl()
+      {
+        return "https://" + dockerIp + ":8290";
+      }
+
       @Override
       public String getRouterUrl()
       {
         return "http://" + dockerIp + ":8888";
       }
 
+      @Override
+      public String getRouterTLSUrl()
+      {
+        return "https://" + dockerIp + ":9088";
+      }
+
+      @Override
+      public String getPermissiveRouterUrl()
+      {
+        return "http://" + dockerIp + ":8889";
+      }
+
+      @Override
+      public String getPermissiveRouterTLSUrl()
+      {
+        return "https://" + dockerIp + ":9089";
+      }
+
+      @Override
+      public String getNoClientAuthRouterUrl()
+      {
+        return "http://" + dockerIp + ":8890";
+      }
+
+      @Override
+      public String getNoClientAuthRouterTLSUrl()
+      {
+        return "https://" + dockerIp + ":9090";
+      }
+
       @Override
       public String getBrokerUrl()
       {
         return "http://" + dockerIp + ":8082";
       }
 
+      @Override
+      public String getBrokerTLSUrl()
+      {
+        return "https://" + dockerIp + ":8282";
+      }
+
       @Override
       public String getHistoricalUrl()
       {
         return "http://" + dockerIp + ":8083";
       }
 
+      @Override
+      public String getHistoricalTLSUrl()
+      {
+        return "https://" + dockerIp + ":8283";
+      }
+
       @Override
       public String getMiddleManagerHost()
       {
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/IntegrationTestingConfig.java b/integration-tests/src/main/java/org/apache/druid/testing/IntegrationTestingConfig.java
index de05013a36f..2b72e58c9d6 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/IntegrationTestingConfig.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/IntegrationTestingConfig.java
@@ -27,14 +27,32 @@ public interface IntegrationTestingConfig
 {
   String getCoordinatorUrl();
 
+  String getCoordinatorTLSUrl();
+
   String getIndexerUrl();
 
+  String getIndexerTLSUrl();
+
   String getRouterUrl();
 
+  String getRouterTLSUrl();
+
+  String getPermissiveRouterUrl();
+
+  String getPermissiveRouterTLSUrl();
+
+  String getNoClientAuthRouterUrl();
+
+  String getNoClientAuthRouterTLSUrl();
+
   String getBrokerUrl();
 
+  String getBrokerTLSUrl();
+
   String getHistoricalUrl();
 
+  String getHistoricalTLSUrl();
+
   String getMiddleManagerHost();
 
   String getZookeeperHosts();
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java b/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java
index 667febc6be4..eff8989b23a 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/clients/CoordinatorResourceTestClient.java
@@ -70,6 +70,11 @@ public class CoordinatorResourceTestClient
     );
   }
 
+  private String getMetadataSegmentsURL(String dataSource)
+  {
+    return StringUtils.format("%smetadata/datasources/%s/segments", getCoordinatorURL(), dataSource);
+  }
+
   private String getIntervalsURL(String dataSource)
   {
     return StringUtils.format("%sdatasources/%s/intervals", getCoordinatorURL(), dataSource);
@@ -80,6 +85,25 @@ public class CoordinatorResourceTestClient
     return StringUtils.format("%s%s", getCoordinatorURL(), "loadstatus");
   }
 
+  // return a list of the segment dates for the specified datasource
+  public List<String> getMetadataSegments(final String dataSource)
+  {
+    ArrayList<String> segments = null;
+    try {
+      StatusResponseHolder response = makeRequest(HttpMethod.GET, getMetadataSegmentsURL(dataSource));
+
+      segments = jsonMapper.readValue(
+          response.getContent(), new TypeReference<ArrayList<String>>()
+          {
+          }
+      );
+    }
+    catch (Exception e) {
+      throw Throwables.propagate(e);
+    }
+    return segments;
+  }
+
   // return a list of the segment dates for the specified datasource
   public List<String> getSegmentIntervals(final String dataSource)
   {
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/clients/EventReceiverFirehoseTestClient.java b/integration-tests/src/main/java/org/apache/druid/testing/clients/EventReceiverFirehoseTestClient.java
index a7b4d9abc44..5b01c9ecf3c 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/clients/EventReceiverFirehoseTestClient.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/clients/EventReceiverFirehoseTestClient.java
@@ -71,7 +71,7 @@ public class EventReceiverFirehoseTestClient
   private String getURL()
   {
     return StringUtils.format(
-        "http://%s/druid/worker/v1/chat/%s/push-events/",
+        "https://%s/druid/worker/v1/chat/%s/push-events/",
         host,
         chatID
     );
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/clients/OverlordResourceTestClient.java b/integration-tests/src/main/java/org/apache/druid/testing/clients/OverlordResourceTestClient.java
index 1f5d8888032..eee9eaa75d9 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/clients/OverlordResourceTestClient.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/clients/OverlordResourceTestClient.java
@@ -175,7 +175,7 @@ public class OverlordResourceTestClient
 
   public void waitUntilTaskCompletes(final String taskID)
   {
-    waitUntilTaskCompletes(taskID, 60000, 10);
+    waitUntilTaskCompletes(taskID, 10000, 60);
   }
 
   public void waitUntilTaskCompletes(final String taskID, final int millisEach, final int numTimes)
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/utils/RetryUtil.java b/integration-tests/src/main/java/org/apache/druid/testing/utils/RetryUtil.java
index 6f3d70a68c9..53888dbcfd1 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/utils/RetryUtil.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/utils/RetryUtil.java
@@ -32,9 +32,9 @@ public class RetryUtil
 
   private static final Logger LOG = new Logger(RetryUtil.class);
 
-  public static int DEFAULT_RETRY_COUNT = 10;
+  public static int DEFAULT_RETRY_COUNT = 30;
 
-  public static long DEFAULT_RETRY_SLEEP = TimeUnit.SECONDS.toMillis(30);
+  public static long DEFAULT_RETRY_SLEEP = TimeUnit.SECONDS.toMillis(10);
 
   public static void retryUntilTrue(Callable<Boolean> callable, String task)
   {
diff --git a/integration-tests/src/main/java/org/apache/druid/testing/utils/TestQueryHelper.java b/integration-tests/src/main/java/org/apache/druid/testing/utils/TestQueryHelper.java
index 93692e4c04d..649557d043a 100644
--- a/integration-tests/src/main/java/org/apache/druid/testing/utils/TestQueryHelper.java
+++ b/integration-tests/src/main/java/org/apache/druid/testing/utils/TestQueryHelper.java
@@ -42,6 +42,9 @@ public class TestQueryHelper
   private final QueryResourceTestClient queryClient;
   private final ObjectMapper jsonMapper;
   private final String broker;
+  private final String brokerTLS;
+  private final String router;
+  private final String routerTLS;
 
   @Inject
   TestQueryHelper(
@@ -53,11 +56,17 @@ public class TestQueryHelper
     this.jsonMapper = jsonMapper;
     this.queryClient = queryClient;
     this.broker = config.getBrokerUrl();
+    this.brokerTLS = config.getBrokerTLSUrl();
+    this.router = config.getRouterUrl();
+    this.routerTLS = config.getRouterTLSUrl();
   }
 
   public void testQueriesFromFile(String filePath, int timesToRun) throws Exception
   {
-    testQueriesFromFile(getBrokerURL(), filePath, timesToRun);
+    testQueriesFromFile(getQueryURL(broker), filePath, timesToRun);
+    testQueriesFromFile(getQueryURL(brokerTLS), filePath, timesToRun);
+    testQueriesFromFile(getQueryURL(router), filePath, timesToRun);
+    testQueriesFromFile(getQueryURL(routerTLS), filePath, timesToRun);
   }
 
   public void testQueriesFromFile(String url, String filePath, int timesToRun) throws Exception
@@ -75,7 +84,10 @@ public class TestQueryHelper
 
   public void testQueriesFromString(String str, int timesToRun) throws Exception
   {
-    testQueriesFromString(getBrokerURL(), str, timesToRun);
+    testQueriesFromString(getQueryURL(broker), str, timesToRun);
+    testQueriesFromString(getQueryURL(brokerTLS), str, timesToRun);
+    testQueriesFromString(getQueryURL(router), str, timesToRun);
+    testQueriesFromString(getQueryURL(routerTLS), str, timesToRun);
   }
 
   public void testQueriesFromString(String url, String str, int timesToRun) throws Exception
@@ -93,6 +105,7 @@ public class TestQueryHelper
 
   private void testQueries(String url, List<QueryWithResults> queries, int timesToRun) throws Exception
   {
+    LOG.info("Running queries, url [%s]", url);
     for (int i = 0; i < timesToRun; i++) {
       LOG.info("Starting Iteration %d", i);
 
@@ -119,9 +132,9 @@ public class TestQueryHelper
     }
   }
 
-  private String getBrokerURL()
+  private String getQueryURL(String schemeAndHost)
   {
-    return StringUtils.format("%s/druid/v2?pretty", broker);
+    return StringUtils.format("%s/druid/v2?pretty", schemeAndHost);
   }
 
   @SuppressWarnings("unchecked")
@@ -138,7 +151,7 @@ public class TestQueryHelper
                                   .intervals(interval)
                                   .build();
 
-    List<Map<String, Object>> results = queryClient.query(getBrokerURL(), query);
+    List<Map<String, Object>> results = queryClient.query(getQueryURL(broker), query);
     if (results.isEmpty()) {
       return 0;
     } else {
diff --git a/integration-tests/src/test/java/io/druid/tests/security/ITTLSTest.java b/integration-tests/src/test/java/io/druid/tests/security/ITTLSTest.java
new file mode 100644
index 00000000000..7759eff627f
--- /dev/null
+++ b/integration-tests/src/test/java/io/druid/tests/security/ITTLSTest.java
@@ -0,0 +1,482 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.tests.security;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Throwables;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Client;
+import org.apache.druid.guice.http.DruidHttpClientConfig;
+import org.apache.druid.guice.http.LifecycleUtils;
+import org.apache.druid.https.SSLClientConfig;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.lifecycle.Lifecycle;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.http.client.CredentialedHttpClient;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.HttpClientConfig;
+import org.apache.druid.java.util.http.client.HttpClientInit;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.auth.BasicCredentials;
+import org.apache.druid.java.util.http.client.response.StatusResponseHandler;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.server.security.TLSUtils;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.guice.DruidTestModuleFactory;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.joda.time.Duration;
+import org.testng.Assert;
+import org.testng.annotations.Guice;
+import org.testng.annotations.Test;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLException;
+import javax.ws.rs.core.MediaType;
+import java.io.IOException;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+
+@Guice(moduleFactory = DruidTestModuleFactory.class)
+public class ITTLSTest
+{
+  private static final Logger LOG = new Logger(ITTLSTest.class);
+
+  private static final Duration SSL_HANDSHAKE_TIMEOUT = new Duration(30 * 1000);
+
+  private static final int MAX_BROKEN_PIPE_RETRIES = 30;
+
+  @Inject
+  IntegrationTestingConfig config;
+
+  @Inject
+  ObjectMapper jsonMapper;
+
+  @Inject
+  SSLClientConfig sslClientConfig;
+
+  @Inject
+  @Client
+  HttpClient httpClient;
+
+  @Inject
+  @Client
+  DruidHttpClientConfig httpClientConfig;
+
+  StatusResponseHandler responseHandler = new StatusResponseHandler(StandardCharsets.UTF_8);
+
+  @Test
+  public void testPlaintextAccess()
+  {
+    LOG.info("---------Testing resource access without TLS---------");
+    HttpClient adminClient = new CredentialedHttpClient(
+        new BasicCredentials("admin", "priest"),
+        httpClient
+    );
+    makeRequest(adminClient, HttpMethod.GET, config.getCoordinatorUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getIndexerUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getBrokerUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getHistoricalUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getRouterUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getPermissiveRouterUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getNoClientAuthRouterUrl() + "/status", null);
+  }
+
+  @Test
+  public void testTLSNodeAccess()
+  {
+    LOG.info("---------Testing resource access with TLS enabled---------");
+    HttpClient adminClient = new CredentialedHttpClient(
+        new BasicCredentials("admin", "priest"),
+        httpClient
+    );
+    makeRequest(adminClient, HttpMethod.GET, config.getCoordinatorTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getIndexerTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getBrokerTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getHistoricalTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getRouterTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl() + "/status", null);
+    makeRequest(adminClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void testTLSNodeAccessWithIntermediate()
+  {
+    LOG.info("---------Testing TLS resource access with 3-part cert chain---------");
+    HttpClient intermediateCertClient = makeCustomHttpClient(
+        "client_tls/intermediate_ca_client.jks",
+        "intermediate_ca_client"
+    );
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getCoordinatorTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getIndexerTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getBrokerTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getHistoricalTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getRouterTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl() + "/status", null);
+    makeRequest(intermediateCertClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithNoCert()
+  {
+    LOG.info("---------Testing TLS resource access without a certificate---------");
+    HttpClient certlessClient = makeCertlessClient();
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getRouterTLSUrl());
+    checkFailedAccessNoCert(certlessClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl());
+    makeRequest(certlessClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithWrongHostname()
+  {
+    LOG.info("---------Testing TLS resource access when client certificate has non-matching hostnames---------");
+    HttpClient wrongHostnameClient = makeCustomHttpClient(
+        "client_tls/invalid_hostname_client.jks",
+        "invalid_hostname_client"
+    );
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessWrongHostname(wrongHostnameClient, HttpMethod.GET, config.getRouterTLSUrl());
+    makeRequest(wrongHostnameClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl() + "/status", null);
+    makeRequest(wrongHostnameClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithWrongRoot()
+  {
+    LOG.info("---------Testing TLS resource access when client certificate is signed by a non-trusted root CA---------");
+    HttpClient wrongRootClient = makeCustomHttpClient(
+        "client_tls/client_another_root.jks",
+        "druid_another_root"
+    );
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getRouterTLSUrl());
+    checkFailedAccessWrongRoot(wrongRootClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl());
+    makeRequest(wrongRootClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithRevokedCert()
+  {
+    LOG.info("---------Testing TLS resource access when client certificate has been revoked---------");
+    HttpClient revokedClient = makeCustomHttpClient(
+        "client_tls/revoked_client.jks",
+        "revoked_druid"
+    );
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessRevoked(revokedClient, HttpMethod.GET, config.getRouterTLSUrl());
+    makeRequest(revokedClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl() + "/status", null);
+    makeRequest(revokedClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithExpiredCert()
+  {
+    LOG.info("---------Testing TLS resource access when client certificate has expired---------");
+    HttpClient expiredClient = makeCustomHttpClient(
+        "client_tls/expired_client.jks",
+        "expired_client"
+    );
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getRouterTLSUrl());
+    checkFailedAccessExpired(expiredClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl());
+    makeRequest(expiredClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  @Test
+  public void checkAccessWithNotCASignedCert()
+  {
+    LOG.info(
+        "---------Testing TLS resource access when client certificate is signed by a non-CA intermediate cert---------");
+    HttpClient notCAClient = makeCustomHttpClient(
+        "client_tls/invalid_ca_client.jks",
+        "invalid_ca_client"
+    );
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getCoordinatorTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getIndexerTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getBrokerTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getHistoricalTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getRouterTLSUrl());
+    checkFailedAccessNotCA(notCAClient, HttpMethod.GET, config.getPermissiveRouterTLSUrl());
+    makeRequest(notCAClient, HttpMethod.GET, config.getNoClientAuthRouterTLSUrl() + "/status", null);
+  }
+
+  private void checkFailedAccessNoCert(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Certless",
+        SSLException.class,
+        "Received fatal alert: bad_certificate"
+    );
+  }
+
+  private void checkFailedAccessWrongHostname(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Wrong hostname",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private void checkFailedAccessWrongRoot(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Wrong root cert",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private void checkFailedAccessRevoked(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Revoked cert",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private void checkFailedAccessExpired(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Expired cert",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private void checkFailedAccessNotCA(HttpClient httpClient, HttpMethod method, String url)
+  {
+    checkFailedAccess(
+        httpClient,
+        method,
+        url + "/status",
+        "Cert signed by non-CA",
+        SSLException.class,
+        "Received fatal alert: certificate_unknown"
+    );
+  }
+
+  private HttpClientConfig.Builder getHttpClientConfigBuilder(SSLContext sslContext)
+  {
+    return HttpClientConfig
+        .builder()
+        .withNumConnections(httpClientConfig.getNumConnections())
+        .withReadTimeout(httpClientConfig.getReadTimeout())
+        .withWorkerCount(httpClientConfig.getNumMaxThreads())
+        .withCompressionCodec(
+            HttpClientConfig.CompressionCodec.valueOf(StringUtils.toUpperCase(httpClientConfig.getCompressionCodec()))
+        )
+        .withUnusedConnectionTimeoutDuration(httpClientConfig.getUnusedConnectionTimeout())
+        .withSslHandshakeTimeout(SSL_HANDSHAKE_TIMEOUT)
+        .withSslContext(sslContext);
+  }
+
+  private HttpClient makeCustomHttpClient(String keystorePath, String certAlias)
+  {
+    SSLContext intermediateClientSSLContext = new TLSUtils.ClientSSLContextBuilder()
+        .setProtocol(sslClientConfig.getProtocol())
+        .setTrustStoreType(sslClientConfig.getTrustStoreType())
+        .setTrustStorePath(sslClientConfig.getTrustStorePath())
+        .setTrustStoreAlgorithm(sslClientConfig.getTrustStoreAlgorithm())
+        .setTrustStorePasswordProvider(sslClientConfig.getTrustStorePasswordProvider())
+        .setKeyStoreType(sslClientConfig.getKeyStoreType())
+        .setKeyStorePath(keystorePath)
+        .setKeyStoreAlgorithm(sslClientConfig.getKeyManagerFactoryAlgorithm())
+        .setCertAlias(certAlias)
+        .setKeyStorePasswordProvider(sslClientConfig.getKeyStorePasswordProvider())
+        .setKeyManagerFactoryPasswordProvider(sslClientConfig.getKeyManagerPasswordProvider())
+        .build();
+
+    final HttpClientConfig.Builder builder = getHttpClientConfigBuilder(intermediateClientSSLContext);
+
+    final Lifecycle lifecycle = new Lifecycle();
+
+    HttpClient client = HttpClientInit.createClient(
+        builder.build(),
+        LifecycleUtils.asMmxLifecycle(lifecycle)
+    );
+
+    HttpClient adminClient = new CredentialedHttpClient(
+        new BasicCredentials("admin", "priest"),
+        client
+    );
+    return adminClient;
+  }
+
+  private HttpClient makeCertlessClient()
+  {
+    SSLContext certlessClientSSLContext = new TLSUtils.ClientSSLContextBuilder()
+        .setProtocol(sslClientConfig.getProtocol())
+        .setTrustStoreType(sslClientConfig.getTrustStoreType())
+        .setTrustStorePath(sslClientConfig.getTrustStorePath())
+        .setTrustStoreAlgorithm(sslClientConfig.getTrustStoreAlgorithm())
+        .setTrustStorePasswordProvider(sslClientConfig.getTrustStorePasswordProvider())
+        .build();
+
+    final HttpClientConfig.Builder builder = getHttpClientConfigBuilder(certlessClientSSLContext);
+
+    final Lifecycle lifecycle = new Lifecycle();
+
+    HttpClient client = HttpClientInit.createClient(
+        builder.build(),
+        LifecycleUtils.asMmxLifecycle(lifecycle)
+    );
+
+    HttpClient adminClient = new CredentialedHttpClient(
+        new BasicCredentials("admin", "priest"),
+        client
+    );
+    return adminClient;
+  }
+
+  private void checkFailedAccess(
+      HttpClient httpClient,
+      HttpMethod method,
+      String url,
+      String clientDesc,
+      Class expectedException,
+      String expectedExceptionMsg
+  )
+  {
+    int retries = 0;
+    while (true) {
+      try {
+        makeRequest(httpClient, method, url, null, -1);
+      }
+      catch (RuntimeException re) {
+        Throwable rootCause = Throwables.getRootCause(re);
+
+        if (rootCause instanceof IOException && "Broken pipe".equals(rootCause.getMessage())) {
+          if (retries > MAX_BROKEN_PIPE_RETRIES) {
+            Assert.fail(StringUtils.format(
+                "Broken pipe retries exhausted, test failed, did not get %s.",
+                expectedException
+            ));
+          } else {
+            retries += 1;
+            continue;
+          }
+        }
+
+        Assert.assertTrue(
+            expectedException.isInstance(rootCause),
+            StringUtils.format("Expected %s but found %s instead.", expectedException, rootCause)
+        );
+
+        Assert.assertEquals(
+            rootCause.getMessage(),
+            expectedExceptionMsg
+        );
+
+        LOG.info("%s client [%s] request failed as expected when accessing [%s]", clientDesc, method, url);
+        return;
+      }
+      Assert.fail(StringUtils.format("Test failed, did not get %s.", expectedException));
+    }
+  }
+
+  private StatusResponseHolder makeRequest(HttpClient httpClient, HttpMethod method, String url, byte[] content)
+  {
+    return makeRequest(httpClient, method, url, content, 4);
+  }
+
+  private StatusResponseHolder makeRequest(
+      HttpClient httpClient,
+      HttpMethod method,
+      String url,
+      byte[] content,
+      int maxRetries
+  )
+  {
+    try {
+      Request request = new Request(method, new URL(url));
+      if (content != null) {
+        request.setContent(MediaType.APPLICATION_JSON, content);
+      }
+      int retryCount = 0;
+
+      StatusResponseHolder response;
+
+      while (true) {
+        response = httpClient.go(
+            request,
+            responseHandler
+        ).get();
+
+        if (!response.getStatus().equals(HttpResponseStatus.OK)) {
+          String errMsg = StringUtils.format(
+              "Error while making request to url[%s] status[%s] content[%s]",
+              url,
+              response.getStatus(),
+              response.getContent()
+          );
+          if (retryCount > maxRetries) {
+            throw new ISE(errMsg);
+          } else {
+            LOG.error(errMsg);
+            LOG.error("retrying in 3000ms, retryCount: " + retryCount);
+            retryCount++;
+            Thread.sleep(3000);
+          }
+        } else {
+          LOG.info("[%s] request to [%s] succeeded.", method, url);
+          break;
+        }
+      }
+      return response;
+    }
+    catch (Exception e) {
+      throw Throwables.propagate(e);
+    }
+  }
+}
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/hadoop/ITHadoopIndexTest.java b/integration-tests/src/test/java/org/apache/druid/tests/hadoop/ITHadoopIndexTest.java
index 5a8124e689f..ebfe22d9595 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/hadoop/ITHadoopIndexTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/hadoop/ITHadoopIndexTest.java
@@ -75,7 +75,7 @@ public class ITHadoopIndexTest extends AbstractIndexerTest
     try {
       final String taskID = indexer.submitTask(indexerSpec);
       LOG.info("TaskID for loading index task %s", taskID);
-      indexer.waitUntilTaskCompletes(taskID, 60000, 20);
+      indexer.waitUntilTaskCompletes(taskID, 10000, 120);
       RetryUtil.retryUntil(
           new Callable<Boolean>()
           {
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITRealtimeIndexTaskTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITRealtimeIndexTaskTest.java
index eb728396101..0995dba495b 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITRealtimeIndexTaskTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/AbstractITRealtimeIndexTaskTest.java
@@ -142,8 +142,8 @@ public abstract class AbstractITRealtimeIndexTaskTest extends AbstractIndexerTes
             }
           },
           true,
-          60000,
-          10,
+          10000,
+          60,
           "Real-time generated segments loaded"
       );
 
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java
index 8ea14edda26..2f9078aeb16 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITCompactionTaskTest.java
@@ -42,6 +42,7 @@ public class ITCompactionTaskTest extends AbstractIndexerTest
   {
     loadData();
     final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
+    intervalsBeforeCompaction.sort(null);
     final String compactedInterval = "2013-08-31T00:00:00.000Z/2013-09-02T00:00:00.000Z";
     if (intervalsBeforeCompaction.contains(compactedInterval)) {
       throw new ISE("Containing a segment for the compacted interval[%s] before compaction", compactedInterval);
@@ -49,12 +50,14 @@ public class ITCompactionTaskTest extends AbstractIndexerTest
     try {
       queryHelper.testQueriesFromFile(INDEX_QUERIES_RESOURCE, 2);
       compactData(false);
+
+      // 4 segments across 2 days, compacted into 1 new segment (5 total)
+      checkCompactionFinished(5);
       queryHelper.testQueriesFromFile(INDEX_QUERIES_RESOURCE, 2);
 
-      final List<String> intervalsAfterCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
-      if (!intervalsAfterCompaction.contains(compactedInterval)) {
-        throw new ISE("Compacted segment for interval[%s] does not exist", compactedInterval);
-      }
+      intervalsBeforeCompaction.add(compactedInterval);
+      intervalsBeforeCompaction.sort(null);
+      checkCompactionIntervals(intervalsBeforeCompaction);
     }
     finally {
       unloadAndKillData(INDEX_DATASOURCE);
@@ -66,21 +69,16 @@ public class ITCompactionTaskTest extends AbstractIndexerTest
   {
     loadData();
     final List<String> intervalsBeforeCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
+    intervalsBeforeCompaction.sort(null);
     try {
       queryHelper.testQueriesFromFile(INDEX_QUERIES_RESOURCE, 2);
       compactData(true);
+
+      // 4 segments across 2 days, compacted into 2 new segments (6 total)
+      checkCompactionFinished(6);
       queryHelper.testQueriesFromFile(INDEX_QUERIES_RESOURCE, 2);
 
-      final List<String> intervalsAfterCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
-      intervalsBeforeCompaction.sort(null);
-      intervalsAfterCompaction.sort(null);
-      if (!intervalsBeforeCompaction.equals(intervalsAfterCompaction)) {
-        throw new ISE(
-            "Intervals before compaction[%s] should be same with those after compaction[%s]",
-            intervalsBeforeCompaction,
-            intervalsAfterCompaction
-        );
-      }
+      checkCompactionIntervals(intervalsBeforeCompaction);
     }
     finally {
       unloadAndKillData(INDEX_DATASOURCE);
@@ -112,4 +110,30 @@ public class ITCompactionTaskTest extends AbstractIndexerTest
         "Segment Compaction"
     );
   }
+
+  private void checkCompactionFinished(int numExpectedSegments)
+  {
+    RetryUtil.retryUntilTrue(
+        () -> {
+          int metadataSegmentCount = coordinator.getMetadataSegments(INDEX_DATASOURCE).size();
+          LOG.info("Current metadata segment count: %d, expected: %d", metadataSegmentCount, numExpectedSegments);
+          return metadataSegmentCount == numExpectedSegments;
+        },
+        "Compaction segment count check"
+    );
+  }
+
+  private void checkCompactionIntervals(List<String> expectedIntervals)
+  {
+    RetryUtil.retryUntilTrue(
+        () -> {
+          final List<String> intervalsAfterCompaction = coordinator.getSegmentIntervals(INDEX_DATASOURCE);
+          intervalsAfterCompaction.sort(null);
+          System.out.println("AFTER: " + intervalsAfterCompaction);
+          System.out.println("EXPECTED: " + expectedIntervals);
+          return intervalsAfterCompaction.equals(expectedIntervals);
+        },
+        "Compaction interval check"
+    );
+  }
 }
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaIndexingServiceTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaIndexingServiceTest.java
index 38b508c4539..7b59721277a 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaIndexingServiceTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaIndexingServiceTest.java
@@ -278,8 +278,8 @@ public class ITKafkaIndexingServiceTest extends AbstractIndexerTest
             }
           },
           true,
-          30000,
-          10,
+          10000,
+          30,
           "Real-time generated segments loaded"
       );
     }
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaTest.java
index efbf2c29592..b9ddcfc394d 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITKafkaTest.java
@@ -226,7 +226,7 @@ public class ITKafkaTest extends AbstractIndexerTest
     LOG.info("-------------SUBMITTED TASK");
 
     // wait for the task to finish
-    indexer.waitUntilTaskCompletes(taskID, 20000, 30);
+    indexer.waitUntilTaskCompletes(taskID, 10000, 60);
 
     // wait for segments to be handed off
     try {
@@ -240,8 +240,8 @@ public class ITKafkaTest extends AbstractIndexerTest
             }
           },
           true,
-          30000,
-          10,
+          10000,
+          30,
           "Real-time generated segments loaded"
       );
     }
diff --git a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITUnionQueryTest.java b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITUnionQueryTest.java
index 748da882ee2..a1bfb4c2fc1 100644
--- a/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITUnionQueryTest.java
+++ b/integration-tests/src/test/java/org/apache/druid/tests/indexer/ITUnionQueryTest.java
@@ -25,18 +25,26 @@ import com.google.inject.Inject;
 import org.apache.druid.curator.discovery.ServerDiscoveryFactory;
 import org.apache.druid.curator.discovery.ServerDiscoverySelector;
 import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.response.StatusResponseHandler;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
 import org.apache.druid.testing.IntegrationTestingConfig;
 import org.apache.druid.testing.clients.EventReceiverFirehoseTestClient;
 import org.apache.druid.testing.guice.DruidTestModuleFactory;
 import org.apache.druid.testing.guice.TestClient;
 import org.apache.druid.testing.utils.RetryUtil;
 import org.apache.druid.testing.utils.ServerDiscoveryUtil;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
 import org.joda.time.DateTime;
 import org.testng.annotations.Guice;
 import org.testng.annotations.Test;
 
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
@@ -173,6 +181,26 @@ public class ITUnionQueryTest extends AbstractIndexerTest
 
       LOG.info("Event Receiver Found at host [%s]", host);
 
+      LOG.info("Checking worker /status/health for [%s]", host);
+      final StatusResponseHandler handler = new StatusResponseHandler(StandardCharsets.UTF_8);
+      RetryUtil.retryUntilTrue(
+          () -> {
+            try {
+              StatusResponseHolder response = httpClient.go(
+                  new Request(HttpMethod.GET, new URL(StringUtils.format("https://%s/status/health", host))),
+                  handler
+              ).get();
+              return response.getStatus().equals(HttpResponseStatus.OK);
+            }
+            catch (Throwable e) {
+              LOG.error(e, "");
+              return false;
+            }
+          },
+          StringUtils.format("Checking /status/health for worker [%s]", host)
+      );
+      LOG.info("Finished checking worker /status/health for [%s], success", host);
+
       EventReceiverFirehoseTestClient client = new EventReceiverFirehoseTestClient(
           host,
           EVENT_RECEIVER_SERVICE_PREFIX + id,
diff --git a/integration-tests/src/test/resources/indexer/kafka_index_task.json b/integration-tests/src/test/resources/indexer/kafka_index_task.json
index f580d46a500..55e28c7c47a 100644
--- a/integration-tests/src/test/resources/indexer/kafka_index_task.json
+++ b/integration-tests/src/test/resources/indexer/kafka_index_task.json
@@ -61,7 +61,7 @@
       "type" : "realtime",
       "maxRowsInMemory": 500000,
       "intermediatePersistPeriod": "PT3M",
-      "windowPeriod": "PT1M",
+      "windowPeriod": "PT150S",
       "basePersistDirectory": "/home/y/var/druid_state/kafka_test/realtime/basePersist"
     }
   }
diff --git a/integration-tests/stop_cluster.sh b/integration-tests/stop_cluster.sh
index fb09ec2067b..fe4ad202fe5 100755
--- a/integration-tests/stop_cluster.sh
+++ b/integration-tests/stop_cluster.sh
@@ -14,8 +14,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-for node in druid-historical druid-coordinator druid-overlord druid-router druid-broker druid-middlemanager druid-zookeeper-kafka druid-metadata-storage;
-do 
+for node in druid-historical druid-coordinator druid-overlord druid-router druid-router-permissive-tls druid-router-no-client-auth-tls druid-broker druid-middlemanager druid-zookeeper-kafka druid-metadata-storage;
+
+do
 docker stop $node
 docker rm $node
 done
+
+docker network rm druid-it-net
diff --git a/pom.xml b/pom.xml
index 4438a0de052..ec3e5938f00 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1106,6 +1106,7 @@
                         <exclude>**/tutorial/conf/**</exclude>
                         <exclude>**/derby.log</exclude>
                         <exclude>**/docker/**</exclude>
+                        <exclude>**/client_tls/**</exclude>
                         <!--IDE MODULE FILES-->
                         <exclude>**/*.iml</exclude>
                         <!--CRASH LOGS-->
diff --git a/server/src/main/java/org/apache/druid/server/StatusResource.java b/server/src/main/java/org/apache/druid/server/StatusResource.java
index 5d14ca7f334..b96d2bf6b60 100644
--- a/server/src/main/java/org/apache/druid/server/StatusResource.java
+++ b/server/src/main/java/org/apache/druid/server/StatusResource.java
@@ -31,9 +31,11 @@ import org.apache.druid.server.http.security.ConfigResourceFilter;
 import org.apache.druid.server.http.security.StateResourceFilter;
 
 import javax.inject.Inject;
+import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -47,7 +49,6 @@ import java.util.Set;
 @Path("/status")
 public class StatusResource
 {
-
   private final Properties properties;
 
   private final DruidServerConfig druidServerConfig;
@@ -73,7 +74,9 @@ public class StatusResource
   @GET
   @ResourceFilters(StateResourceFilter.class)
   @Produces(MediaType.APPLICATION_JSON)
-  public Status doGet()
+  public Status doGet(
+      @Context final HttpServletRequest req
+  )
   {
     return new Status(Initialization.getLoadedImplementations(DruidModule.class));
   }
diff --git a/server/src/main/java/org/apache/druid/server/emitter/HttpEmitterModule.java b/server/src/main/java/org/apache/druid/server/emitter/HttpEmitterModule.java
index 3c3bca1df7e..35170140f51 100644
--- a/server/src/main/java/org/apache/druid/server/emitter/HttpEmitterModule.java
+++ b/server/src/main/java/org/apache/druid/server/emitter/HttpEmitterModule.java
@@ -126,13 +126,13 @@ public class HttpEmitterModule implements Module
       }
     } else if (sslConfig.getTrustStorePath() != null) {
       log.info("Creating SSLContext for HttpEmitter client using config [%s]", sslConfig);
-      effectiveSSLContext = TLSUtils.createSSLContext(
-          sslConfig.getProtocol(),
-          sslConfig.getTrustStoreType(),
-          sslConfig.getTrustStorePath(),
-          sslConfig.getTrustStoreAlgorithm(),
-          sslConfig.getTrustStorePasswordProvider()
-      );
+      effectiveSSLContext = new TLSUtils.ClientSSLContextBuilder()
+          .setProtocol(sslConfig.getProtocol())
+          .setTrustStoreType(sslConfig.getTrustStoreType())
+          .setTrustStorePath(sslConfig.getTrustStorePath())
+          .setTrustStoreAlgorithm(sslConfig.getTrustStoreAlgorithm())
+          .setTrustStorePasswordProvider(sslConfig.getTrustStorePasswordProvider())
+          .build();
     } else {
       effectiveSSLContext = sslContext;
     }
diff --git a/server/src/main/java/org/apache/druid/server/http/OverlordProxyServlet.java b/server/src/main/java/org/apache/druid/server/http/OverlordProxyServlet.java
index 9f6b3eeaab9..0b4a12dd276 100644
--- a/server/src/main/java/org/apache/druid/server/http/OverlordProxyServlet.java
+++ b/server/src/main/java/org/apache/druid/server/http/OverlordProxyServlet.java
@@ -23,12 +23,17 @@ import com.google.common.base.Throwables;
 import com.google.inject.Inject;
 import org.apache.druid.client.indexing.IndexingService;
 import org.apache.druid.discovery.DruidLeaderClient;
+import org.apache.druid.guice.annotations.Global;
+import org.apache.druid.guice.http.DruidHttpClientConfig;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.server.security.AuthConfig;
+import com.google.inject.Provider;
+import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.client.api.Request;
 import org.eclipse.jetty.proxy.ProxyServlet;
 
+import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.net.URI;
@@ -40,13 +45,19 @@ import java.net.URISyntaxException;
 public class OverlordProxyServlet extends ProxyServlet
 {
   private final DruidLeaderClient druidLeaderClient;
+  private final Provider<HttpClient> httpClientProvider;
+  private final DruidHttpClientConfig httpClientConfig;
 
   @Inject
   OverlordProxyServlet(
-      @IndexingService DruidLeaderClient druidLeaderClient
+      @IndexingService DruidLeaderClient druidLeaderClient,
+      @Global Provider<HttpClient> httpClientProvider,
+      @Global DruidHttpClientConfig httpClientConfig
   )
   {
     this.druidLeaderClient = druidLeaderClient;
+    this.httpClientProvider = httpClientProvider;
+    this.httpClientConfig = httpClientConfig;
   }
 
   @Override
@@ -71,6 +82,22 @@ public class OverlordProxyServlet extends ProxyServlet
     }
   }
 
+  @Override
+  protected HttpClient newHttpClient()
+  {
+    return httpClientProvider.get();
+  }
+
+  @Override
+  protected HttpClient createHttpClient() throws ServletException
+  {
+    HttpClient client = super.createHttpClient();
+    // override timeout set in ProxyServlet.createHttpClient
+    setTimeout(httpClientConfig.getReadTimeout().getMillis());
+    return client;
+  }
+
+
   @Override
   protected void sendProxyRequest(
       HttpServletRequest clientRequest,
diff --git a/server/src/main/java/org/apache/druid/server/initialization/TLSServerConfig.java b/server/src/main/java/org/apache/druid/server/initialization/TLSServerConfig.java
index 6ea2c1258d7..4e8a625023b 100644
--- a/server/src/main/java/org/apache/druid/server/initialization/TLSServerConfig.java
+++ b/server/src/main/java/org/apache/druid/server/initialization/TLSServerConfig.java
@@ -55,6 +55,27 @@ public class TLSServerConfig
   @JsonProperty
   private List<String> excludeProtocols;
 
+  @JsonProperty
+  private boolean requireClientCertificate = false;
+
+  @JsonProperty
+  private String trustStoreType;
+
+  @JsonProperty
+  private String trustStorePath;
+
+  @JsonProperty
+  private String trustStoreAlgorithm;
+
+  @JsonProperty("trustStorePassword")
+  private PasswordProvider trustStorePasswordProvider;
+
+  @JsonProperty
+  private boolean validateHostnames = true;
+
+  @JsonProperty
+  private String crlPath;
+
   public String getKeyStorePath()
   {
     return keyStorePath;
@@ -105,6 +126,41 @@ public class TLSServerConfig
     return excludeProtocols;
   }
 
+  public boolean isRequireClientCertificate()
+  {
+    return requireClientCertificate;
+  }
+
+  public String getTrustStoreType()
+  {
+    return trustStoreType;
+  }
+
+  public String getTrustStorePath()
+  {
+    return trustStorePath;
+  }
+
+  public String getTrustStoreAlgorithm()
+  {
+    return trustStoreAlgorithm;
+  }
+
+  public PasswordProvider getTrustStorePasswordProvider()
+  {
+    return trustStorePasswordProvider;
+  }
+
+  public boolean isValidateHostnames()
+  {
+    return validateHostnames;
+  }
+
+  public String getCrlPath()
+  {
+    return crlPath;
+  }
+
   @Override
   public String toString()
   {
@@ -117,6 +173,12 @@ public class TLSServerConfig
            ", excludeCipherSuites=" + excludeCipherSuites +
            ", includeProtocols=" + includeProtocols +
            ", excludeProtocols=" + excludeProtocols +
+           ", requireClientCertificate=" + requireClientCertificate +
+           ", trustStoreType='" + trustStoreType + '\'' +
+           ", trustStorePath='" + trustStorePath + '\'' +
+           ", trustStoreAlgorithm='" + trustStoreAlgorithm + '\'' +
+           ", validateHostnames='" + validateHostnames + '\'' +
+           ", crlPath='" + crlPath + '\'' +
            '}';
   }
 }
diff --git a/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyServerModule.java b/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyServerModule.java
index ee613d15334..269bb2c554e 100644
--- a/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyServerModule.java
+++ b/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyServerModule.java
@@ -75,6 +75,8 @@ import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
 
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLEngine;
+import javax.net.ssl.TrustManagerFactory;
+import java.security.KeyStore;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -231,6 +233,7 @@ public class JettyServerModule extends JerseyServletModule
       if (sslContextFactoryBinding == null) {
         // Never trust all certificates by default
         sslContextFactory = new SslContextFactory(false);
+
         sslContextFactory.setKeyStorePath(tlsServerConfig.getKeyStorePath());
         sslContextFactory.setKeyStoreType(tlsServerConfig.getKeyStoreType());
         sslContextFactory.setKeyStorePassword(tlsServerConfig.getKeyStorePasswordProvider().getPassword());
@@ -256,6 +259,37 @@ public class JettyServerModule extends JerseyServletModule
           sslContextFactory.setExcludeProtocols(
               tlsServerConfig.getExcludeProtocols().toArray(new String[0]));
         }
+
+        sslContextFactory.setNeedClientAuth(tlsServerConfig.isRequireClientCertificate());
+        if (tlsServerConfig.isRequireClientCertificate()) {
+          if (tlsServerConfig.getCrlPath() != null) {
+            // setValidatePeerCerts is used just to enable revocation checking using a static CRL file.
+            // Certificate validation is always performed when client certificates are required.
+            sslContextFactory.setValidatePeerCerts(true);
+            sslContextFactory.setCrlPath(tlsServerConfig.getCrlPath());
+          }
+          if (tlsServerConfig.isValidateHostnames()) {
+            sslContextFactory.setEndpointIdentificationAlgorithm("HTTPS");
+          }
+          if (tlsServerConfig.getTrustStorePath() != null) {
+            sslContextFactory.setTrustStorePath(tlsServerConfig.getTrustStorePath());
+            sslContextFactory.setTrustStoreType(
+                tlsServerConfig.getTrustStoreType() == null
+                ? KeyStore.getDefaultType()
+                : tlsServerConfig.getTrustStoreType()
+            );
+            sslContextFactory.setTrustManagerFactoryAlgorithm(
+                tlsServerConfig.getTrustStoreAlgorithm() == null
+                ? TrustManagerFactory.getDefaultAlgorithm()
+                : tlsServerConfig.getTrustStoreAlgorithm()
+            );
+            sslContextFactory.setTrustStorePassword(
+                tlsServerConfig.getTrustStorePasswordProvider() == null
+                ? null
+                : tlsServerConfig.getTrustStorePasswordProvider().getPassword()
+            );
+          }
+        }
       } else {
         sslContextFactory = sslContextFactoryBinding.getProvider().get();
       }
diff --git a/server/src/main/java/org/apache/druid/server/security/TLSUtils.java b/server/src/main/java/org/apache/druid/server/security/TLSUtils.java
index 733af5ef34d..691a0ae1c75 100644
--- a/server/src/main/java/org/apache/druid/server/security/TLSUtils.java
+++ b/server/src/main/java/org/apache/druid/server/security/TLSUtils.java
@@ -19,48 +19,205 @@
 
 package org.apache.druid.server.security;
 
+import com.google.common.base.Preconditions;
 import com.google.common.base.Throwables;
 import org.apache.druid.metadata.PasswordProvider;
 
+import org.eclipse.jetty.util.ssl.AliasedX509ExtendedKeyManager;
+
+import javax.annotation.Nullable;
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509ExtendedKeyManager;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.security.KeyManagementException;
 import java.security.KeyStore;
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
 import java.security.cert.CertificateException;
 
 public class TLSUtils
 {
+  public static class ClientSSLContextBuilder
+  {
+    private String protocol;
+    private String trustStoreType;
+    private String trustStorePath;
+    private String trustStoreAlgorithm;
+    private PasswordProvider trustStorePasswordProvider;
+    private String keyStoreType;
+    private String keyStorePath;
+    private String keyStoreAlgorithm;
+    private String certAlias;
+    private PasswordProvider keyStorePasswordProvider;
+    private PasswordProvider keyManagerFactoryPasswordProvider;
+
+    public ClientSSLContextBuilder setProtocol(String protocol)
+    {
+      this.protocol = protocol;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setTrustStoreType(String trustStoreType)
+    {
+      this.trustStoreType = trustStoreType;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setTrustStorePath(String trustStorePath)
+    {
+      this.trustStorePath = trustStorePath;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setTrustStoreAlgorithm(String trustStoreAlgorithm)
+    {
+      this.trustStoreAlgorithm = trustStoreAlgorithm;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setTrustStorePasswordProvider(PasswordProvider trustStorePasswordProvider)
+    {
+      this.trustStorePasswordProvider = trustStorePasswordProvider;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyStoreType(String keyStoreType)
+    {
+      this.keyStoreType = keyStoreType;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyStorePath(String keyStorePath)
+    {
+      this.keyStorePath = keyStorePath;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyStoreAlgorithm(String keyStoreAlgorithm)
+    {
+      this.keyStoreAlgorithm = keyStoreAlgorithm;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setCertAlias(String certAlias)
+    {
+      this.certAlias = certAlias;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyStorePasswordProvider(PasswordProvider keyStorePasswordProvider)
+    {
+      this.keyStorePasswordProvider = keyStorePasswordProvider;
+      return this;
+    }
+
+    public ClientSSLContextBuilder setKeyManagerFactoryPasswordProvider(PasswordProvider keyManagerFactoryPasswordProvider)
+    {
+      this.keyManagerFactoryPasswordProvider = keyManagerFactoryPasswordProvider;
+      return this;
+    }
+
+    public SSLContext build()
+    {
+      Preconditions.checkNotNull(trustStorePath, "must specify a trustStorePath");
+
+      return createSSLContext(
+        protocol,
+        trustStoreType,
+        trustStorePath,
+        trustStoreAlgorithm,
+        trustStorePasswordProvider,
+        keyStoreType,
+        keyStorePath,
+        keyStoreAlgorithm,
+        certAlias,
+        keyStorePasswordProvider,
+        keyManagerFactoryPasswordProvider
+      );
+    }
+  }
+
   public static SSLContext createSSLContext(
-      String protocol,
-      String trustStoreType,
+      @Nullable String protocol,
+      @Nullable String trustStoreType,
       String trustStorePath,
-      String trustStoreAlgorithm,
-      PasswordProvider trustStorePasswordProvider
+      @Nullable String trustStoreAlgorithm,
+      @Nullable PasswordProvider trustStorePasswordProvider,
+      @Nullable String keyStoreType,
+      @Nullable String keyStorePath,
+      @Nullable String keyStoreAlgorithm,
+      @Nullable String certAlias,
+      @Nullable PasswordProvider keyStorePasswordProvider,
+      @Nullable PasswordProvider keyManagerFactoryPasswordProvider
   )
   {
     SSLContext sslContext = null;
     try {
       sslContext = SSLContext.getInstance(protocol == null ? "TLSv1.2" : protocol);
-      KeyStore keyStore = KeyStore.getInstance(trustStoreType == null
+      KeyStore trustStore = KeyStore.getInstance(trustStoreType == null
                                                ? KeyStore.getDefaultType()
                                                : trustStoreType);
-      keyStore.load(
+      trustStore.load(
           new FileInputStream(trustStorePath),
-          trustStorePasswordProvider.getPassword().toCharArray()
+          trustStorePasswordProvider == null ? null : trustStorePasswordProvider.getPassword().toCharArray()
       );
       TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(trustStoreAlgorithm == null
                                                                                 ? TrustManagerFactory.getDefaultAlgorithm()
                                                                                 : trustStoreAlgorithm);
-      trustManagerFactory.init(keyStore);
-      sslContext.init(null, trustManagerFactory.getTrustManagers(), null);
+      trustManagerFactory.init(trustStore);
+
+
+      KeyManager[] keyManagers;
+      if (keyStorePath != null) {
+        KeyStore keyStore = KeyStore.getInstance(keyStoreType == null
+                                                 ? KeyStore.getDefaultType()
+                                                 : keyStoreType);
+        keyStore.load(
+            new FileInputStream(keyStorePath),
+            keyStorePasswordProvider == null ? null : keyStorePasswordProvider.getPassword().toCharArray()
+        );
+
+        KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(
+            keyStoreAlgorithm == null ?
+            KeyManagerFactory.getDefaultAlgorithm() : keyStoreAlgorithm
+        );
+        keyManagerFactory.init(
+            keyStore,
+            keyManagerFactoryPasswordProvider == null ? null : keyManagerFactoryPasswordProvider.getPassword().toCharArray()
+        );
+        keyManagers = createAliasedKeyManagers(keyManagerFactory.getKeyManagers(), certAlias);
+      } else {
+        keyManagers = null;
+      }
+
+      sslContext.init(
+          keyManagers,
+          trustManagerFactory.getTrustManagers(),
+          null
+      );
     }
-    catch (CertificateException | KeyManagementException | IOException | KeyStoreException | NoSuchAlgorithmException e) {
+    catch (CertificateException | KeyManagementException | IOException | KeyStoreException | NoSuchAlgorithmException | UnrecoverableKeyException e) {
       Throwables.propagate(e);
     }
     return sslContext;
   }
+
+  // Use Jetty's aliased KeyManager for consistency between server/client TLS configs
+  private static KeyManager[] createAliasedKeyManagers(KeyManager[] delegates, String certAlias)
+  {
+    KeyManager[] aliasedManagers = new KeyManager[delegates.length];
+    for (int i = 0; i < delegates.length; i++) {
+      if (delegates[i] instanceof X509ExtendedKeyManager) {
+        aliasedManagers[i] = new AliasedX509ExtendedKeyManager((X509ExtendedKeyManager) delegates[i], certAlias);
+      } else {
+        aliasedManagers[i] = delegates[i];
+      }
+    }
+    return aliasedManagers;
+  }
 }
diff --git a/server/src/test/java/org/apache/druid/server/http/OverlordProxyServletTest.java b/server/src/test/java/org/apache/druid/server/http/OverlordProxyServletTest.java
index 0a3dd93e260..dd364b74361 100644
--- a/server/src/test/java/org/apache/druid/server/http/OverlordProxyServletTest.java
+++ b/server/src/test/java/org/apache/druid/server/http/OverlordProxyServletTest.java
@@ -41,7 +41,7 @@ public class OverlordProxyServletTest
 
     EasyMock.replay(druidLeaderClient, request);
 
-    URI uri = URI.create(new OverlordProxyServlet(druidLeaderClient).rewriteTarget(request));
+    URI uri = URI.create(new OverlordProxyServlet(druidLeaderClient, null, null).rewriteTarget(request));
     Assert.assertEquals("https://overlord:port/druid/overlord/worker?param1=test&param2=test2", uri.toString());
   }
 
diff --git a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
index 5503c85b84c..750a9fa5a8c 100644
--- a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
+++ b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
@@ -45,6 +45,7 @@ import org.apache.druid.guice.LifecycleModule;
 import org.apache.druid.guice.ManageLifecycle;
 import org.apache.druid.guice.annotations.CoordinatorIndexingServiceHelper;
 import org.apache.druid.guice.annotations.EscalatedGlobal;
+import org.apache.druid.guice.http.JettyHttpClientModule;
 import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.http.client.HttpClient;
@@ -127,6 +128,8 @@ public class CliCoordinator extends ServerRunnable
   {
     List<Module> modules = new ArrayList<>();
 
+    modules.add(JettyHttpClientModule.global());
+
     modules.add(
         new Module()
         {