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 be118857fc60..4e15ec279305 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
@@ -104,6 +104,15 @@ public class ConnectionConfiguration {
   private final long pauseMs;
   private final long pauseMsForServerOverloaded;
 
+  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
@@ -117,11 +126,11 @@ 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,
+    this.operationTimeout = getNonNegativeInt(conf, HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
       HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
 
-    this.operationTimeout = conf.getInt(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);
@@ -129,14 +138,14 @@ public class ConnectionConfiguration {
     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,
+      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,
@@ -150,16 +159,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);
 
     this.scanTimeout = conf.getInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
       HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
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 66375000049e..c2a8a7ac565d 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,9 +18,13 @@
 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;
 import org.apache.hadoop.hbase.testclassification.ClientTests;
 import org.apache.hadoop.hbase.testclassification.SmallTests;
@@ -46,4 +50,60 @@ public void itHandlesDeprecatedPauseForCQTBE() {
     config = new ConnectionConfiguration(conf);
     assertEquals(timeoutMs, config.getPauseMillisForServerOverloaded());
   }
+
+  @Test
+  public void testDefaultMetaOperationTimeout() {
+    Configuration conf = HBaseConfiguration.create();
+    long clientOperationTimeoutMs = 1000;
+    conf.setLong(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, clientOperationTimeoutMs);
+    ConnectionConfiguration config = new ConnectionConfiguration(conf);
+    assertEquals(clientOperationTimeoutMs, config.getOperationTimeout());
+    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());
+  }
+
 }