diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionConfiguration.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionConfiguration.java
index 545a85b16c99..82eabbbc8de5 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionConfiguration.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionConfiguration.java
@@ -80,6 +80,15 @@ public class ConnectionConfiguration {
   // toggle for async/sync prefetch
   private final boolean clientScannerAsyncPrefetch;
 
+  private static int getNonNegativeInt(Configuration conf, String key, int defaultValue) {
+    int value = conf.getInt(key, defaultValue);
+    if (value < 0) {
+      throw new IllegalArgumentException(
+        "The " + key + " must be non-negative, current value is " + value);
+    }
+    return value;
+  }
+
   /**
    * Constructor
    * @param conf Configuration object
@@ -93,28 +102,27 @@ public class ConnectionConfiguration {
     this.writeBufferPeriodicFlushTimerTickMs = conf.getLong(
       WRITE_BUFFER_PERIODIC_FLUSH_TIMERTICK_MS, WRITE_BUFFER_PERIODIC_FLUSH_TIMERTICK_MS_DEFAULT);
 
-    this.metaOperationTimeout = conf.getInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT,
-      conf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
-        HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT));
-
-    this.operationTimeout = conf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
+    this.operationTimeout = getNonNegativeInt(conf, HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
       HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
 
+    this.metaOperationTimeout = getNonNegativeInt(conf,
+      HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, operationTimeout);
+
     this.scannerCaching = conf.getInt(HConstants.HBASE_CLIENT_SCANNER_CACHING,
       HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING);
 
     this.scannerMaxResultSize = conf.getLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
       HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
 
-    this.primaryCallTimeoutMicroSecond =
-      conf.getInt(PRIMARY_CALL_TIMEOUT_MICROSECOND, PRIMARY_CALL_TIMEOUT_MICROSECOND_DEFAULT);
+    this.primaryCallTimeoutMicroSecond = getNonNegativeInt(conf, PRIMARY_CALL_TIMEOUT_MICROSECOND,
+      PRIMARY_CALL_TIMEOUT_MICROSECOND_DEFAULT);
 
-    this.replicaCallTimeoutMicroSecondScan =
-      conf.getInt(PRIMARY_SCAN_TIMEOUT_MICROSECOND, PRIMARY_SCAN_TIMEOUT_MICROSECOND_DEFAULT);
+    this.replicaCallTimeoutMicroSecondScan = getNonNegativeInt(conf, PRIMARY_SCAN_TIMEOUT_MICROSECOND,
+      PRIMARY_SCAN_TIMEOUT_MICROSECOND_DEFAULT);
 
-    this.metaReplicaCallTimeoutMicroSecondScan =
-      conf.getInt(HConstants.HBASE_CLIENT_META_REPLICA_SCAN_TIMEOUT,
-        HConstants.HBASE_CLIENT_META_REPLICA_SCAN_TIMEOUT_DEFAULT);
+    this.metaReplicaCallTimeoutMicroSecondScan = getNonNegativeInt(conf,
+      HConstants.HBASE_CLIENT_META_REPLICA_SCAN_TIMEOUT,
+      HConstants.HBASE_CLIENT_META_REPLICA_SCAN_TIMEOUT_DEFAULT);
 
     this.retries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
       HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
@@ -127,16 +135,17 @@ public class ConnectionConfiguration {
     this.bufferedMutatorMaxMutations =
       conf.getInt(BUFFERED_MUTATOR_MAX_MUTATIONS_KEY, BUFFERED_MUTATOR_MAX_MUTATIONS_DEFAULT);
 
-    this.rpcTimeout =
-      conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
+    this.rpcTimeout = getNonNegativeInt(conf, HConstants.HBASE_RPC_TIMEOUT_KEY,
+      HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
 
-    this.readRpcTimeout = conf.getInt(HConstants.HBASE_RPC_READ_TIMEOUT_KEY,
-      conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
+    this.readRpcTimeout =
+      getNonNegativeInt(conf, HConstants.HBASE_RPC_READ_TIMEOUT_KEY, rpcTimeout);
 
-    this.metaReadRpcTimeout = conf.getInt(HBASE_CLIENT_META_READ_RPC_TIMEOUT_KEY, readRpcTimeout);
+    this.metaReadRpcTimeout =
+      getNonNegativeInt(conf, HBASE_CLIENT_META_READ_RPC_TIMEOUT_KEY, readRpcTimeout);
 
-    this.writeRpcTimeout = conf.getInt(HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY,
-      conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
+    this.writeRpcTimeout =
+      getNonNegativeInt(conf, HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY, rpcTimeout);
   }
 
   /**
diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestConnectionConfiguration.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestConnectionConfiguration.java
index a2b6994087af..a048874aca10 100644
--- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestConnectionConfiguration.java
+++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestConnectionConfiguration.java
@@ -18,7 +18,11 @@
 package org.apache.hadoop.hbase.client;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.util.Arrays;
+import java.util.List;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HConstants;
@@ -44,4 +48,47 @@ public void testDefaultMetaOperationTimeout() {
     assertEquals(clientOperationTimeoutMs, config.getMetaOperationTimeout());
   }
 
+  @Test
+  public void testNegativeTimeoutsThrow() {
+    List<String> timeoutKeys = Arrays.asList(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT,
+      HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, ConnectionConfiguration.PRIMARY_CALL_TIMEOUT_MICROSECOND,
+      ConnectionConfiguration.PRIMARY_SCAN_TIMEOUT_MICROSECOND,
+      HConstants.HBASE_CLIENT_META_REPLICA_SCAN_TIMEOUT, HConstants.HBASE_RPC_TIMEOUT_KEY,
+      HConstants.HBASE_RPC_READ_TIMEOUT_KEY, ConnectionConfiguration.HBASE_CLIENT_META_READ_RPC_TIMEOUT_KEY,
+      HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY);
+
+    for (String key : timeoutKeys) {
+      Configuration conf = HBaseConfiguration.create();
+      conf.setInt(key, -1);
+      IllegalArgumentException error =
+        assertThrows(IllegalArgumentException.class, () -> new ConnectionConfiguration(conf), key);
+      assertTrue(error.getMessage().contains(key), key);
+    }
+  }
+
+  @Test
+  public void testZeroTimeoutsAreAllowed() {
+    Configuration conf = HBaseConfiguration.create();
+    conf.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, 0);
+    conf.setInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, 0);
+    conf.setInt(ConnectionConfiguration.PRIMARY_CALL_TIMEOUT_MICROSECOND, 0);
+    conf.setInt(ConnectionConfiguration.PRIMARY_SCAN_TIMEOUT_MICROSECOND, 0);
+    conf.setInt(HConstants.HBASE_CLIENT_META_REPLICA_SCAN_TIMEOUT, 0);
+    conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, 0);
+    conf.setInt(HConstants.HBASE_RPC_READ_TIMEOUT_KEY, 0);
+    conf.setInt(ConnectionConfiguration.HBASE_CLIENT_META_READ_RPC_TIMEOUT_KEY, 0);
+    conf.setInt(HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY, 0);
+
+    ConnectionConfiguration config = new ConnectionConfiguration(conf);
+    assertEquals(0, config.getOperationTimeout());
+    assertEquals(0, config.getMetaOperationTimeout());
+    assertEquals(0, config.getPrimaryCallTimeoutMicroSecond());
+    assertEquals(0, config.getReplicaCallTimeoutMicroSecondScan());
+    assertEquals(0, config.getMetaReplicaCallTimeoutMicroSecondScan());
+    assertEquals(0, config.getRpcTimeout());
+    assertEquals(0, config.getReadRpcTimeout());
+    assertEquals(0, config.getMetaReadRpcTimeout());
+    assertEquals(0, config.getWriteRpcTimeout());
+  }
+
 }