From e11d67404945d80db2eb8a99453606419dbdc938 Mon Sep 17 00:00:00 2001
From: Inigo Goiri <inigoiri@apache.org>
Date: Fri, 1 Jun 2018 16:59:04 -0700
Subject: [PATCH] HDFS-13637. RBF: Router fails when threadIndex (in
 ConnectionPool) wraps around Integer.MIN_VALUE. Contributed by CR Hota.

---
 .../server/federation/router/ConnectionPool.java    | 12 +++++++++++-
 .../federation/router/TestConnectionManager.java    | 13 +++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/ConnectionPool.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/ConnectionPool.java
index 6b416dd25e5..5fcde5b2e85 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/ConnectionPool.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/ConnectionPool.java
@@ -29,6 +29,7 @@
 
 import javax.net.SocketFactory;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
@@ -148,6 +149,14 @@ protected ConnectionPoolId getConnectionPoolId() {
     return this.connectionPoolId;
   }
 
+  /**
+   * Get the clientIndex used to calculate index for lookup.
+   */
+  @VisibleForTesting
+  public AtomicInteger getClientIndex() {
+    return this.clientIndex;
+  }
+
   /**
    * Return the next connection round-robin.
    *
@@ -161,7 +170,8 @@ protected ConnectionContext getConnection() {
     ConnectionContext conn = null;
     List<ConnectionContext> tmpConnections = this.connections;
     int size = tmpConnections.size();
-    int threadIndex = this.clientIndex.getAndIncrement();
+    // Inc and mask off sign bit, lookup index should be non-negative int
+    int threadIndex = this.clientIndex.getAndIncrement() & 0x7FFFFFFF;
     for (int i=0; i<size; i++) {
       int index = (threadIndex + i) % size;
       conn = tmpConnections.get(index);
diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestConnectionManager.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestConnectionManager.java
index a7316481d7e..0e1eb40783f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestConnectionManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestConnectionManager.java
@@ -32,6 +32,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.junit.Assert.assertNotNull;
 
 /**
  * Test functionalities of {@link ConnectionManager}, which manages a pool
@@ -149,6 +150,18 @@ public void testGetConnection() throws Exception {
     assertTrue(cc.isActive());
   }
 
+  @Test
+  public void testValidClientIndex() throws Exception {
+    ConnectionPool pool = new ConnectionPool(
+        conf, TEST_NN_ADDRESS, TEST_USER1, 2, 2, ClientProtocol.class);
+    for(int i = -3; i <= 3; i++) {
+      pool.getClientIndex().set(i);
+      ConnectionContext conn = pool.getConnection();
+      assertNotNull(conn);
+      assertTrue(conn.isUsable());
+    }
+  }
+
   @Test
   public void getGetConnectionNamenodeProtocol() throws Exception {
     Map<ConnectionPoolId, ConnectionPool> poolMap = connManager.getPools();