diff --git a/java/interface-annotations/pom.xml b/java/interface-annotations/pom.xml
index 90691ca856d0..e9dc30b84eb0 100644
--- a/java/interface-annotations/pom.xml
+++ b/java/interface-annotations/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
 
   <artifactId>interface-annotations</artifactId>
diff --git a/java/pom.xml b/java/pom.xml
index 73531ff5f1c6..e5bc0ac0da9c 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -40,7 +40,7 @@
 
   <groupId>org.yb</groupId>
   <artifactId>yb-parent</artifactId>
-  <version>0.8.2-SNAPSHOT</version>
+  <version>0.8.3-SNAPSHOT</version>
   <packaging>pom</packaging>
 
   <name>YugaByte</name>
@@ -112,6 +112,8 @@
 
     <postgresql.jdbc.driver.version>42.2.5</postgresql.jdbc.driver.version>
 
+    <bouncycastle.version>1.61</bouncycastle.version>
+
     <!-- Test configuration -->
     <testdata.dir>target/testdata</testdata.dir>
     <testArgLine>-enableassertions -Xmx3000m
@@ -128,7 +130,6 @@
     <module>yb-client</module>
     <module>yb-cli</module>
     <module>yb-cql</module>
-    <module>yb-cql-4x</module>
     <module>yb-pgsql</module>
     <module>yb-loadtester</module>
     <module>yb-jedis-tests</module>
@@ -342,7 +343,7 @@
       <dependency>
         <groupId>org.yb</groupId>
         <artifactId>yb-client</artifactId>
-        <version>0.8.2-SNAPSHOT</version>
+        <version>0.8.3-SNAPSHOT</version>
       </dependency>
       <dependency>
         <groupId>org.yb</groupId>
@@ -474,6 +475,11 @@
         <artifactId>jedis</artifactId>
         <version>${jedis.version}</version>
       </dependency>
+      <dependency>
+        <groupId>org.bouncycastle</groupId>
+        <artifactId>bcpkix-jdk15on</artifactId>
+        <version>${bouncycastle.version}</version>
+      </dependency>
 
       <!--
         Dependencies typically used in tests.
@@ -494,7 +500,7 @@
       <dependency>
         <groupId>org.yb</groupId>
         <artifactId>yb-client</artifactId>
-        <version>0.8.2-SNAPSHOT</version>
+        <version>0.8.3-SNAPSHOT</version>
         <type>test-jar</type>
       </dependency>
       <dependency>
diff --git a/java/yb-cdc/pom.xml b/java/yb-cdc/pom.xml
index d268126191e8..b489075f8e6b 100644
--- a/java/yb-cdc/pom.xml
+++ b/java/yb-cdc/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
   <groupId>org.yb</groupId>
   <artifactId>yb-cdc</artifactId>
@@ -43,12 +43,12 @@
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
     </dependency>
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
diff --git a/java/yb-cli/pom.xml b/java/yb-cli/pom.xml
index 23280aac7ce2..498b73de3d90 100644
--- a/java/yb-cli/pom.xml
+++ b/java/yb-cli/pom.xml
@@ -25,7 +25,7 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
 
   <artifactId>yb-cli</artifactId>
@@ -40,12 +40,12 @@
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
     </dependency>
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
@@ -70,6 +70,11 @@
       <groupId>org.springframework.shell</groupId>
       <artifactId>spring-shell</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.bouncycastle</groupId>
+      <artifactId>bcpkix-jdk15on</artifactId>
+      <version>1.61</version>
+    </dependency>
   </dependencies>
 
   <build>
diff --git a/java/yb-cli/src/main/java/org/yb/ybcli/commands/YBCliCommands.java b/java/yb-cli/src/main/java/org/yb/ybcli/commands/YBCliCommands.java
index f624a3ad550e..cedd52cdc86f 100644
--- a/java/yb-cli/src/main/java/org/yb/ybcli/commands/YBCliCommands.java
+++ b/java/yb-cli/src/main/java/org/yb/ybcli/commands/YBCliCommands.java
@@ -28,6 +28,7 @@
 import org.yb.Common.HostPortPB;
 import org.yb.Schema;
 import org.yb.client.AsyncYBClient;
+import org.yb.client.AsyncYBClient.AsyncYBClientBuilder;
 import org.yb.client.ChangeConfigResponse;
 import org.yb.client.GetLoadMovePercentResponse;
 import org.yb.client.GetMasterClusterConfigResponse;
@@ -84,9 +85,28 @@ public String connect(
       @CliOption(key = { "masters", "m" },
                  mandatory = true,
                  help = "Comma separated list of masters as '<host>:<port>'")
-      final String masterAddresses) {
+      final String masterAddresses,
+      @CliOption(key = { "certFile", "cert"},
+                 help = "CA Certificate for SSL connections.")
+      final String certFile,
+      @CliOption(key = { "clientCertFile", "clientCert"},
+                 help = "Client Certificate for mTLS connections.")
+      final String clientCertFile,
+      @CliOption(key = { "clientKeyFile", "clientKey"},
+                 help = "Client Private Key for mTLS connections.")
+      final String clientKeyFile) {
     try {
-      AsyncYBClient asyncClient = new AsyncYBClient.AsyncYBClientBuilder(masterAddresses).build();
+      AsyncYBClientBuilder builder = new AsyncYBClientBuilder(masterAddresses);
+      if (certFile != null) {
+        builder.sslCertFile(certFile);
+      }
+      if (clientCertFile != null) {
+        if (clientKeyFile == null) {
+          return "ClientKey cannot be null when ClientCert is provided.";
+        }
+        builder.sslClientCertFiles(clientCertFile, clientKeyFile);
+      }
+      AsyncYBClient asyncClient = builder.build();
       ybClient = new YBClient(asyncClient);
       this.masterAddresses = masterAddresses;
       connectedToDatabase = true;
diff --git a/java/yb-client/pom.xml b/java/yb-client/pom.xml
index 1f920b905a86..7376f89b8030 100644
--- a/java/yb-client/pom.xml
+++ b/java/yb-client/pom.xml
@@ -25,17 +25,22 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
 
   <artifactId>yb-client</artifactId>
   <name>YB Java Client</name>
-  <version>0.8.2-SNAPSHOT</version>
+  <version>0.8.3-SNAPSHOT</version>
   <properties>
     <yb.thirdparty.dir>${project.basedir}/../../thirdparty</yb.thirdparty.dir>
   </properties>
 
   <dependencies>
+    <dependency>
+      <groupId>org.bouncycastle</groupId>
+      <artifactId>bcpkix-jdk15on</artifactId>
+      <version>1.61</version>
+    </dependency>
     <dependency>
       <groupId>com.google.code.gson</groupId>
       <artifactId>gson</artifactId>
diff --git a/java/yb-client/src/main/java/org/yb/client/AsyncYBClient.java b/java/yb-client/src/main/java/org/yb/client/AsyncYBClient.java
index bf0ca77e7c0c..9d62853704a6 100644
--- a/java/yb-client/src/main/java/org/yb/client/AsyncYBClient.java
+++ b/java/yb-client/src/main/java/org/yb/client/AsyncYBClient.java
@@ -52,7 +52,9 @@
 import com.stumbleupon.async.Callback;
 import com.stumbleupon.async.Deferred;
 
-import org.jboss.netty.buffer.ChannelBuffer;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+import org.bouncycastle.util.io.pem.PemObject;
+import org.bouncycastle.util.io.pem.PemReader;
 import org.yb.Common;
 import org.yb.Common.YQLDatabase;
 import org.yb.Schema;
@@ -66,6 +68,7 @@
 import org.yb.util.NetUtil;
 import org.yb.util.Pair;
 import org.yb.util.Slice;
+import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.channel.ChannelEvent;
 import org.jboss.netty.channel.ChannelStateEvent;
 import org.jboss.netty.channel.DefaultChannelPipeline;
@@ -83,14 +86,21 @@
 
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
-import javax.net.ssl.TrustManager;
+import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.TrustManagerFactory;
 
 import java.security.cert.CertificateFactory;
+import java.security.cert.Certificate;
 import java.security.cert.X509Certificate;
 import java.security.KeyStore;
+import java.security.KeyFactory;
+import java.security.PrivateKey;
+import java.security.Security;
+import java.security.spec.InvalidKeySpecException;
+import java.security.spec.PKCS8EncodedKeySpec;
 
 import java.io.FileInputStream;
+import java.io.FileReader;
 
 import javax.annotation.concurrent.GuardedBy;
 import java.net.InetAddress;
@@ -262,6 +272,12 @@ public class AsyncYBClient implements AutoCloseable {
 
   private final String certFile;
 
+  private final String clientCertFile;
+  private final String clientKeyFile;
+
+  private final String clientHost;
+  private final int clientPort;
+
   private volatile boolean closed;
 
   private AsyncYBClient(AsyncYBClientBuilder b) {
@@ -272,6 +288,10 @@ private AsyncYBClient(AsyncYBClientBuilder b) {
     this.defaultOperationTimeoutMs = b.defaultOperationTimeoutMs;
     this.defaultAdminOperationTimeoutMs = b.defaultAdminOperationTimeoutMs;
     this.certFile = b.certFile;
+    this.clientCertFile = b.clientCertFile;
+    this.clientKeyFile = b.clientKeyFile;
+    this.clientHost = b.clientHost;
+    this.clientPort = b.clientPort;
     this.defaultSocketReadTimeoutMs = b.defaultSocketReadTimeoutMs;
   }
 
@@ -1797,6 +1817,9 @@ TabletClient newClient(String uuid, final String host, final int port) {
     // Java since the JRE doesn't expose any way to call setsockopt() with
     // TCP_KEEPIDLE.  And of course the default timeout is >2h. Sigh.
     config.setKeepAlive(true);
+    if (clientHost != null) {
+      chan.bind(new InetSocketAddress(clientHost, clientPort));
+    }
     chan.connect(new InetSocketAddress(host, port));  // Won't block.
     return client;
   }
@@ -2055,7 +2078,7 @@ private final class TabletClientPipeline extends DefaultChannelPipeline {
     TabletClient init(String uuid) {
       final TabletClient client = new TabletClient(AsyncYBClient.this, uuid);
       if (certFile != null) {
-        SslHandler sslHandler = this.createSslHandler(certFile);
+        SslHandler sslHandler = this.createSslHandler(certFile, clientCertFile, clientKeyFile);
         if (sslHandler != null) {
           sslHandler.setIssueHandshake(true);
           super.addFirst("ssl", sslHandler);
@@ -2088,16 +2111,36 @@ public void sendUpstream(final ChannelEvent event) {
       super.sendUpstream(event);
     }
 
-    private SslHandler createSslHandler(String certfile) {
+    private PrivateKey getPrivateKey(String keyFile) {
       try {
+        PemReader pemReader = new PemReader(new FileReader(keyFile));
+        PemObject pemObject = pemReader.readPemObject();
+        byte[] bytes = pemObject.getContent();
+        PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(bytes);
+        KeyFactory kf = KeyFactory.getInstance("RSA");
+        PrivateKey pk = kf.generatePrivate(spec);
+        return pk;
+      } catch (InvalidKeySpecException e) {
+        log.error("Could not read the private key file.", e);
+        throw new RuntimeException("InvalidKeySpecException while reading key: " + keyFile);
+      } catch (Exception e) {
+        log.error("Issue reading pem file.", e);
+        throw new RuntimeException("IOException reading key: " + keyFile);
+      }
+    }
+
+    private SslHandler createSslHandler(String certfile, String clientCertFile,
+                                        String clientKeyFile) {
+      try {
+        Security.addProvider(new BouncyCastleProvider());
         CertificateFactory cf = CertificateFactory.getInstance("X.509");
         FileInputStream fis = new FileInputStream(certFile);
         X509Certificate ca;
         try {
           ca = (X509Certificate) cf.generateCertificate(fis);
         } catch (Exception e) {
-          log.error("Exception generating certificate from input file: ", e);
-          return null;
+          log.error("Exception generating CA certificate from input file: ", e);
+          throw e;
         } finally {
           fis.close();
         }
@@ -2113,14 +2156,52 @@ private SslHandler createSslHandler(String certfile) {
         TrustManagerFactory tmf = TrustManagerFactory.getInstance(tmfAlgorithm);
         tmf.init(keyStore);
 
+        X509Certificate clientCert = null;
+        KeyStore clientKeyStore = null;
+        KeyManagerFactory kmf = null;
+        if (clientCertFile != null) {
+          if (clientKeyFile == null) {
+            log.error("Both client cert and key needed for mutual auth.");
+            return null;
+          }
+          fis = new FileInputStream(clientCertFile);
+          try {
+            clientCert = (X509Certificate) cf.generateCertificate(fis);
+          } catch (Exception e) {
+            log.error("Exception generating CA certificate from input file: ", e);
+            throw e;
+          } finally {
+            fis.close();
+          }
+          PrivateKey pk = getPrivateKey(clientKeyFile);
+          Certificate[] chain = new Certificate[2];
+          chain[0] = clientCert;
+          chain[1] = ca;
+
+          clientKeyStore = KeyStore.getInstance(keyStoreType);
+          clientKeyStore.load(null, null);
+          clientKeyStore.setCertificateEntry("node_crt", clientCert);
+          String password = "password";
+          char[] ksPass = password.toCharArray();
+          clientKeyStore.setKeyEntry("node_key", pk, ksPass, chain);
+
+          kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+          kmf.init(clientKeyStore, ksPass);
+        }
+
         SSLContext sslContext = SSLContext.getInstance("TLS");
-        sslContext.init(null, tmf.getTrustManagers(), null);
+        // mTLS is enabled.
+        if (kmf != null) {
+          sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
+        } else {
+          sslContext.init(null, tmf.getTrustManagers(), null);
+        }
         SSLEngine sslEngine = sslContext.createSSLEngine();
         sslEngine.setUseClientMode(true);
         return new SslHandler(sslEngine);
       } catch (Exception e) {
         log.error("Exception creating sslContext: ", e);
-        return null;
+        throw new RuntimeException("SSLContext creation failed: " + e.toString());
       }
     }
 
@@ -2470,6 +2551,10 @@ public final static class AsyncYBClientBuilder {
     private long defaultSocketReadTimeoutMs = DEFAULT_SOCKET_READ_TIMEOUT_MS;
 
     private String certFile = null;
+    private String clientCertFile = null;
+    private String clientKeyFile = null;
+    private String clientHost = null;
+    private int clientPort = 0;
 
     private Executor bossExecutor;
     private Executor workerExecutor;
@@ -2555,7 +2640,7 @@ public AsyncYBClientBuilder defaultSocketReadTimeoutMs(long timeoutMs) {
      * Optional.
      * If not provided, defaults to null.
      * A value of null disables an SSL connection.
-     * @param certFile the path to the certificate.
+     * @param certFile the path to the certificate (PEM Encoded).
      * @return this builder
      */
     public AsyncYBClientBuilder sslCertFile(String certFile) {
@@ -2563,6 +2648,35 @@ public AsyncYBClientBuilder sslCertFile(String certFile) {
       return this;
     }
 
+    /**
+     * Sets the client cert and key files if mutual auth is enabled.
+     * Optional.
+     * If not provided, defaults to null.
+     * A value of null disables an mTLS connection.
+     * @param certFile the path to the client certificate (PEM Encoded).
+     * @param keyFile the path to the client private key file.
+     * @return this builder
+     */
+    public AsyncYBClientBuilder sslClientCertFiles(String certFile, String keyFile) {
+      this.clientCertFile = certFile;
+      this.clientKeyFile = keyFile;
+      return this;
+    }
+
+    /**
+     * Sets the outbond client host:port on which the socket binds.
+     * Optional.
+     * If not provided, defaults to null.
+     * @param clientHost the address to bind to.
+     * @param clientPort the port to bind to (0 means any free port).
+     * @return this builder
+     */
+    public AsyncYBClientBuilder bindHostAddress(String clientHost, int clientPort) {
+      this.clientHost = clientHost;
+      this.clientPort = clientPort;
+      return this;
+    }
+
     /**
      * Set the executors which will be used for the embedded Netty boss and workers.
      * Optional.
diff --git a/java/yb-client/src/main/java/org/yb/client/YBClient.java b/java/yb-client/src/main/java/org/yb/client/YBClient.java
index f2119daee338..bf563b93ce27 100644
--- a/java/yb-client/src/main/java/org/yb/client/YBClient.java
+++ b/java/yb-client/src/main/java/org/yb/client/YBClient.java
@@ -1326,6 +1326,33 @@ public YBClientBuilder sslCertFile(String certFile) {
       return this;
     }
 
+    /**
+     * Sets the client cert and key files if mutual auth is enabled.
+     * Optional.
+     * If not provided, defaults to null.
+     * A value of null disables an mTLS connection.
+     * @param certFile the path to the client certificate.
+     * @param keyFile the path to the client private key file.
+     * @return this builder
+     */
+    public YBClientBuilder sslClientCertFiles(String certFile, String keyFile) {
+      clientBuilder.sslClientCertFiles(certFile, keyFile);
+      return this;
+    }
+
+    /**
+     * Sets the outbound client host:port on which the socket binds.
+     * Optional.
+     * If not provided, defaults to null.
+     * @param host the address to bind to.
+     * @param port the port to bind to (0 means any free port).
+     * @return this builder
+     */
+    public YBClientBuilder bindHostAddress(String host, int port) {
+      clientBuilder.bindHostAddress(host, port);
+      return this;
+    }
+
     /**
      * Set the executors which will be used for the embedded Netty boss and workers.
      * Optional.
diff --git a/java/yb-client/src/test/java/org/yb/client/BaseYBClientTest.java b/java/yb-client/src/test/java/org/yb/client/BaseYBClientTest.java
index b97c5b9a8479..bb9c6599d9ac 100644
--- a/java/yb-client/src/test/java/org/yb/client/BaseYBClientTest.java
+++ b/java/yb-client/src/test/java/org/yb/client/BaseYBClientTest.java
@@ -76,6 +76,8 @@ protected void afterStartingMiniCluster() throws Exception {
         .defaultOperationTimeoutMs(DEFAULT_SLEEP)
         .defaultSocketReadTimeoutMs(DEFAULT_SLEEP)
         .sslCertFile(certFile)
+        .sslClientCertFiles(clientCertFile, clientKeyFile)
+        .bindHostAddress(clientHost, clientPort)
         .build();
 
     syncClient = new YBClient(client);
diff --git a/java/yb-client/src/test/java/org/yb/clientent/TestSSLClient.java b/java/yb-client/src/test/java/org/yb/clientent/TestSSLClient.java
index 5b3d68561f97..60991e2cd8a8 100644
--- a/java/yb-client/src/test/java/org/yb/clientent/TestSSLClient.java
+++ b/java/yb-client/src/test/java/org/yb/clientent/TestSSLClient.java
@@ -18,6 +18,7 @@
 
 import com.google.protobuf.ByteString;
 
+import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.BeforeClass;
 import org.junit.runner.RunWith;
@@ -45,7 +46,11 @@ public class TestSSLClient extends TestYBClient {
   private static final String READ_ONLY_TS = "readOnly";
   private static final String READ_ONLY_NEW_TS = "readOnlyNew";
 
-  private void setup() throws Exception {
+  public enum TestMode {
+    TLS, MUTUAL_TLS_NO_CLIENT_VERIFY, MUTUAL_TLS_CLIENT_VERIFY
+  }
+
+  private void setup(TestMode mode) throws Exception {
     destroyMiniCluster();
 
     certFile = String.format("%s/%s", certsDir(), "ca.crt");
@@ -53,9 +58,21 @@ private void setup() throws Exception {
 
     String certDirs = String.format("--certs_dir=%s", certsDir());
 
-    List<String> flagsToAdd = Arrays.asList(
+    List<String> flagsToAdd = new LinkedList<String>(Arrays.asList(
         "--use_node_to_node_encryption=true", "--use_client_to_server_encryption=true",
-        "--allow_insecure_connections=false", certDirs);
+        "--allow_insecure_connections=false", certDirs));
+
+    // If not basic TLS, set-up the client certs.
+    if (mode != TestMode.TLS) {
+      clientCertFile = String.format("%s/%s", certsDir(), "node.127.0.0.100.crt");
+      clientKeyFile = String.format("%s/%s", certsDir(), "node.127.0.0.100.key");
+      flagsToAdd.add("--node_to_node_encryption_use_client_certificates=true");
+    }
+    // If verify client hostname mode, set the correct host for the client.
+    if (mode == TestMode.MUTUAL_TLS_CLIENT_VERIFY) {
+      clientHost = "127.0.0.100";
+      flagsToAdd.add("--verify_client_endpoint=true");
+    }
 
     List<List<String>> tserverArgs = new ArrayList<List<String>>();
     tserverArgs.add(flagsToAdd);
@@ -73,7 +90,7 @@ private void setup() throws Exception {
   public void testClientCorrectCertificate() throws Exception {
     LOG.info("Starting testClientCorrectCertificate");
 
-    setup();
+    setup(TestMode.TLS);
 
     YBClient myClient = null;
 
@@ -84,7 +101,6 @@ public void testClientCorrectCertificate() throws Exception {
     myClient.waitForMasterLeader(Timeouts.adjustTimeoutSecForBuildType(10000));
     myClient.close();
     myClient = null;
-
   }
 
   /**
@@ -96,7 +112,7 @@ public void testClientNoCertificate() throws Exception {
     LOG.info("Starting testClientNoCertificate");
 
     boolean connectSuccess = true;
-    setup();
+    setup(TestMode.TLS);
 
     YBClient myClient = null;
     AsyncYBClient aClient = new AsyncYBClient.AsyncYBClientBuilder(masterAddresses)
@@ -111,7 +127,6 @@ public void testClientNoCertificate() throws Exception {
     myClient.close();
     myClient = null;
     assertFalse(connectSuccess);
-
   }
 
   /**
@@ -123,7 +138,7 @@ public void testClientIncorrectCertificate() throws Exception {
     LOG.info("Starting testClientIncorrectCertificate");
 
     boolean connectSuccess = true;
-    setup();
+    setup(TestMode.TLS);
     String incorrectCert = String.format("%s/%s", certsDir(), "pseudo.crt");
 
     YBClient myClient = null;
@@ -140,7 +155,111 @@ public void testClientIncorrectCertificate() throws Exception {
     myClient.close();
     myClient = null;
     assertFalse(connectSuccess);
+  }
+
+  /**
+   * Test to check that client connection succeeds with mTLS enabled.
+   * @throws Exception
+   */
+  @Test(timeout = 100000)
+  public void testMutualTLS() throws Exception {
+    LOG.info("Starting testMutualTLS");
+
+    setup(TestMode.MUTUAL_TLS_NO_CLIENT_VERIFY);
+
+    YBClient myClient = null;
+
+    AsyncYBClient aClient = new AsyncYBClient.AsyncYBClientBuilder(masterAddresses)
+                            .sslCertFile(certFile)
+                            .sslClientCertFiles(clientCertFile, clientKeyFile)
+                            .build();
+    myClient = new YBClient(aClient);
+    myClient.waitForMasterLeader(Timeouts.adjustTimeoutSecForBuildType(10000));
+    myClient.close();
+    myClient = null;
+  }
 
+  /**
+   * Test to check that client connection fails with mTLS enabled when
+   * no client certs are provided..
+   * @throws Exception
+   */
+  @Test(timeout = 100000)
+  public void testMutualTLSNoClientCert() throws Exception {
+    LOG.info("Starting testMutualTLSNoClientCert");
+
+    boolean connectSuccess = true;
+    setup(TestMode.MUTUAL_TLS_NO_CLIENT_VERIFY);
+
+    YBClient myClient = null;
+    AsyncYBClient aClient = new AsyncYBClient.AsyncYBClientBuilder(masterAddresses)
+                            .sslCertFile(certFile)
+                            .build();
+    myClient = new YBClient(aClient);
+    try {
+      LOG.info("Trying to send an RPC...");
+      myClient.waitForMasterLeader(Timeouts.adjustTimeoutSecForBuildType(10000));
+    } catch (Exception e) {
+      connectSuccess = false;
+    }
+    myClient.close();
+    myClient = null;
+    assertFalse(connectSuccess);
+  }
+
+  /**
+   * Test to check that client connection succeeds with mTLS and client hostname verification.
+   * @throws Exception
+   */
+  @Test(timeout = 100000)
+  public void testMutualTLSClientVerify() throws Exception {
+    LOG.info("Starting testMutualTLSClientVerify");
+
+    setup(TestMode.MUTUAL_TLS_CLIENT_VERIFY);
+
+    YBClient myClient = null;
+
+    AsyncYBClient aClient = new AsyncYBClient.AsyncYBClientBuilder(masterAddresses)
+                            .sslCertFile(certFile)
+                            .sslClientCertFiles(clientCertFile, clientKeyFile)
+                            .bindHostAddress(clientHost, clientPort)
+                            .build();
+    myClient = new YBClient(aClient);
+    myClient.waitForMasterLeader(Timeouts.adjustTimeoutSecForBuildType(10000));
+    myClient.close();
+    myClient = null;
+  }
+
+  /**
+   * Test to check that client connection fails with mTLS and client hostname verification
+   * when using different host that the certificate.
+   * @throws Exception
+   */
+  @Test(timeout = 100000)
+  public void testMutualTLSClientVerifyWrongHost() throws Exception {
+    LOG.info("Starting testMutualTLSClientVerifyWrongHost");
+
+    setup(TestMode.MUTUAL_TLS_CLIENT_VERIFY);
+
+    boolean connectSuccess = true;
+    YBClient myClient = null;
+
+    AsyncYBClient aClient = new AsyncYBClient.AsyncYBClientBuilder(masterAddresses)
+                            .sslCertFile(certFile)
+                            .sslClientCertFiles(clientCertFile, clientKeyFile)
+                            // Cert file being used is of host 127.0.0.100
+                            .bindHostAddress("127.0.0.1", clientPort)
+                            .build();
+    myClient = new YBClient(aClient);
+    try {
+      LOG.info("Trying to send an RPC...");
+      myClient.waitForMasterLeader(Timeouts.adjustTimeoutSecForBuildType(10000));
+    } catch (Exception e) {
+      connectSuccess = false;
+    }
+    myClient.close();
+    myClient = null;
+    assertFalse(connectSuccess);
   }
 
   private static String certsDir() {
diff --git a/java/yb-client/src/test/java/org/yb/minicluster/BaseMiniClusterTest.java b/java/yb-client/src/test/java/org/yb/minicluster/BaseMiniClusterTest.java
index 0e797438438c..e499f059dae0 100644
--- a/java/yb-client/src/test/java/org/yb/minicluster/BaseMiniClusterTest.java
+++ b/java/yb-client/src/test/java/org/yb/minicluster/BaseMiniClusterTest.java
@@ -78,6 +78,15 @@ public class BaseMiniClusterTest extends BaseYBTest {
 
   protected String certFile = null;
 
+  // The client cert files for mTLS.
+  protected String clientCertFile = null;
+  protected String clientKeyFile = null;
+
+  // This is used as the default bind address (Used only for mTLS verification).
+  protected String clientHost = null;
+  protected int clientPort = 0;
+
+
   // Comma separate describing the master addresses and ports.
   protected static String masterAddresses;
   protected static List<HostAndPort> masterHostPorts;
@@ -195,7 +204,9 @@ public void createMiniCluster(int numMasters, List<List<String>> tserverArgs,
                       .numShardsPerTServer(overridableNumShardsPerTServer())
                       .useIpWithCertificate(useIpWithCertificate)
                       .replicationFactor(getReplicationFactor())
-                      .sslCertFile(certFile);
+                      .sslCertFile(certFile)
+                      .sslClientCertFiles(clientCertFile, clientKeyFile)
+                      .bindHostAddress(clientHost, clientPort);
 
     if (tserverEnvVars != null) {
       clusterBuilder.addEnvironmentVariables(tserverEnvVars);
@@ -242,6 +253,8 @@ public void createMiniCluster(int numMasters, List<String> masterArgs,
                       .useIpWithCertificate(useIpWithCertificate)
                       .perTServerArgs(tserverArgs)
                       .sslCertFile(certFile)
+                      .sslClientCertFiles(clientCertFile, clientKeyFile)
+                      .bindHostAddress(clientHost, clientPort)
                       .enablePgTransactions(enablePgTransactions)
                       .build();
     masterAddresses = miniCluster.getMasterAddresses();
diff --git a/java/yb-client/src/test/java/org/yb/minicluster/MiniYBCluster.java b/java/yb-client/src/test/java/org/yb/minicluster/MiniYBCluster.java
index bf59f721ffa6..f359a29dd6c8 100644
--- a/java/yb-client/src/test/java/org/yb/minicluster/MiniYBCluster.java
+++ b/java/yb-client/src/test/java/org/yb/minicluster/MiniYBCluster.java
@@ -144,6 +144,14 @@ public class MiniYBCluster implements AutoCloseable {
 
   private String certFile = null;
 
+  // The client cert files for mTLS.
+  private String clientCertFile = null;
+  private String clientKeyFile = null;
+
+  // This is used as the default bind address (Used only for mTLS verification).
+  private String clientHost = null;
+  private int clientPort = 0;
+
   /**
    * Not to be invoked directly, but through a {@link MiniYBClusterBuilder}.
    */
@@ -153,14 +161,22 @@ public class MiniYBCluster implements AutoCloseable {
                 List<String> commonTServerArgs,
                 Map<String, String> tserverEnvVars,
                 String testClassName,
-                String certFile) throws Exception {
+                String certFile,
+                String clientCertFile,
+                String clientKeyFile,
+                String clientHost,
+                int clientPort) throws Exception {
     this.clusterParameters = clusterParameters;
     this.testClassName = testClassName;
     this.certFile = certFile;
+    this.clientCertFile = clientCertFile;
+    this.clientKeyFile = clientKeyFile;
     if (clusterParameters.pgTransactionsEnabled && !clusterParameters.startPgSqlProxy) {
       throw new AssertionError(
           "Attempting to enable PostgreSQL transactions without enabling PostgreSQL API");
     }
+    this.clientHost = clientHost;
+    this.clientPort = clientPort;
 
     startCluster(
         clusterParameters.numMasters, clusterParameters.numTservers, masterArgs, tserverArgs,
@@ -177,6 +193,8 @@ public void startSyncClient(boolean waitForMasterLeader) throws Exception {
         .defaultAdminOperationTimeoutMs(clusterParameters.defaultTimeoutMs)
         .defaultOperationTimeoutMs(clusterParameters.defaultTimeoutMs)
         .sslCertFile(certFile)
+        .sslClientCertFiles(clientCertFile, clientKeyFile)
+        .bindHostAddress(clientHost, clientPort)
         .build();
 
     if (waitForMasterLeader) {
diff --git a/java/yb-client/src/test/java/org/yb/minicluster/MiniYBClusterBuilder.java b/java/yb-client/src/test/java/org/yb/minicluster/MiniYBClusterBuilder.java
index 3bc7e62bee6d..1b00acf58f88 100644
--- a/java/yb-client/src/test/java/org/yb/minicluster/MiniYBClusterBuilder.java
+++ b/java/yb-client/src/test/java/org/yb/minicluster/MiniYBClusterBuilder.java
@@ -36,6 +36,14 @@ public class MiniYBClusterBuilder {
 
   private String certFile = null;
 
+  // The client cert files for mTLS.
+  private String clientCertFile = null;
+  private String clientKeyFile = null;
+
+  // This is used as the default bind address (Used only for mTLS verification).
+  private String clientHost = null;
+  private int clientPort = 0;
+
   private Map<String, String> tserverEnvVars = new TreeMap<String, String>();
 
   public MiniYBClusterBuilder numMasters(int numMasters) {
@@ -63,6 +71,18 @@ public MiniYBClusterBuilder sslCertFile(String certFile) {
     return this;
   }
 
+  public MiniYBClusterBuilder sslClientCertFiles(String certFile, String keyFile) {
+    this.clientCertFile = certFile;
+    this.clientKeyFile = keyFile;
+    return this;
+  }
+
+  public MiniYBClusterBuilder bindHostAddress(String clientHost, int clientPort) {
+    this.clientHost = clientHost;
+    this.clientPort = clientPort;
+    return this;
+  }
+
   /**
    * Configures the internal client to use the given timeout for all operations. Also uses the
    * timeout for tasks like waiting for tablet servers to check in with the master.
@@ -196,6 +216,10 @@ public MiniYBCluster build() throws Exception {
         commonTServerArgs,
         tserverEnvVars,
         testClassName,
-        certFile);
+        certFile,
+        clientCertFile,
+        clientKeyFile,
+        clientHost,
+        clientPort);
   }
 }
diff --git a/java/yb-cql-4x/pom.xml b/java/yb-cql-4x/pom.xml
index b968020cacec..5af431f29c14 100644
--- a/java/yb-cql-4x/pom.xml
+++ b/java/yb-cql-4x/pom.xml
@@ -7,7 +7,7 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
   <artifactId>yb-cql-4x</artifactId>
   <name>YB CQL Support for 4.x Driver</name>
diff --git a/java/yb-cql/pom.xml b/java/yb-cql/pom.xml
index 7d6026d4f8da..3f5f20846f03 100644
--- a/java/yb-cql/pom.xml
+++ b/java/yb-cql/pom.xml
@@ -7,7 +7,7 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
   <artifactId>yb-cql</artifactId>
   <name>YB CQL Support</name>
@@ -16,13 +16,13 @@
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
diff --git a/java/yb-jedis-tests/pom.xml b/java/yb-jedis-tests/pom.xml
index 7a6f213dc3a4..95d422be5553 100644
--- a/java/yb-jedis-tests/pom.xml
+++ b/java/yb-jedis-tests/pom.xml
@@ -7,7 +7,7 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
   <artifactId>yb-jedis-tests</artifactId>
   <name>YB Jedis Tests</name>
@@ -36,13 +36,13 @@
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
diff --git a/java/yb-loadtester/pom.xml b/java/yb-loadtester/pom.xml
index c70355964373..81abe3bfb6b7 100644
--- a/java/yb-loadtester/pom.xml
+++ b/java/yb-loadtester/pom.xml
@@ -6,7 +6,7 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
 
   <artifactId>yb-loadtester</artifactId>
@@ -16,14 +16,14 @@
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <scope>test</scope>
     </dependency>
     <dependency>
diff --git a/java/yb-multiapi/pom.xml b/java/yb-multiapi/pom.xml
index b7a69afc2fb9..3cf585f14666 100644
--- a/java/yb-multiapi/pom.xml
+++ b/java/yb-multiapi/pom.xml
@@ -9,7 +9,7 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
 
   <artifactId>yb-multiapi</artifactId>
@@ -19,14 +19,14 @@
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <scope>test</scope>
     </dependency>
     <dependency>
diff --git a/java/yb-pgsql/pom.xml b/java/yb-pgsql/pom.xml
index adc44551934f..d7306dcfb32a 100644
--- a/java/yb-pgsql/pom.xml
+++ b/java/yb-pgsql/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>org.yb</groupId>
     <artifactId>yb-parent</artifactId>
-    <version>0.8.2-SNAPSHOT</version>
+    <version>0.8.3-SNAPSHOT</version>
   </parent>
   <artifactId>yb-pgsql</artifactId>
   <name>YB PostgreSQL Support</name>
@@ -17,13 +17,13 @@
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.yb</groupId>
       <artifactId>yb-client</artifactId>
-      <version>0.8.2-SNAPSHOT</version>
+      <version>0.8.3-SNAPSHOT</version>
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
diff --git a/managed/build.sbt b/managed/build.sbt
index a2ac2da2df5d..676822ba5842 100644
--- a/managed/build.sbt
+++ b/managed/build.sbt
@@ -183,7 +183,7 @@ externalResolvers := {
   validateResolver(ybClientSnapshotResolver, ybClientSnapshotResolverDescription)
 }
 
-libraryDependencies += "org.yb" % "yb-client" % "0.8.2-SNAPSHOT"
+libraryDependencies += "org.yb" % "yb-client" % "0.8.3-SNAPSHOT"
 
 dependencyOverrides += "io.netty" % "netty-handler" % "4.0.36.Final"
 dependencyOverrides += "com.google.protobuf" % "protobuf-java" % "latest.integration"