Throw an exception when trying to read from a broken connection (#1923)

- When attempting to read from a broken connection, Jedis could potentially try to read some corrupt data and interpret it according to the Redis protocol, this makes it throw an exception instead of trying that.
- #1747 describes that possible misinterpretations of the protocol can lead to trying to create very large arrays
- Also refactor clientPause test, it was testing that clients fail, CLIENT PAUSE command pauses all clients, so they should be expected to be delayed rather than failed

Author: mina-asham

src/main/java/redis/clients/jedis/Connection.java

@@ -310,6 +310,10 @@ protected void flush() {
   }
 
   protected Object readProtocolWithCheckingBroken() {
+    if (broken) {
+      throw new JedisConnectionException("Attempting to read from a broken connection");
+    }
+
     try {
       return Protocol.read(inputStream);
     } catch (JedisConnectionException exc) {

src/test/java/redis/clients/jedis/tests/ConnectionTest.java

@@ -75,4 +75,30 @@ public void sendCommand(ProtocolCommand cmd, byte[]... args) {
       assertEquals("ERR Protocol error: invalid multibulk length", jce.getMessage());
     }
   }
+
+  @Test
+  public void readWithBrokenConnection() {
+    class BrokenConnection extends Connection {
+      private BrokenConnection() {
+        super("nonexistinghost", 0);
+        try {
+          connect();
+          fail("Client should fail connecting to nonexistinghost");
+        } catch (JedisConnectionException ignored) {
+        }
+      }
+
+      private Object read() {
+        return readProtocolWithCheckingBroken();
+      }
+    }
+
+    BrokenConnection conn = new BrokenConnection();
+    try {
+      conn.read();
+      fail("Read should fail as connection is broken");
+    } catch (JedisConnectionException jce) {
+      assertEquals("Attempting to read from a broken connection", jce.getMessage());
+    }
+  }
 }

src/test/java/redis/clients/jedis/tests/commands/ControlCommandsTest.java

@@ -5,6 +5,12 @@
 import static org.junit.Assert.assertTrue;
 
 import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
@@ -138,40 +144,48 @@ public void waitReplicas() {
   }
 
   @Test
-  public void clientPause() throws InterruptedException {
-    assertEquals("PONG", jedis.ping());
-    jedis.clientPause(600);
+  public void clientPause() throws InterruptedException, ExecutionException {
+    ExecutorService executorService = Executors.newFixedThreadPool(2);
     try {
-      jedis.ping();
-    } catch (Exception e) {
-      assertEquals("java.net.SocketTimeoutException: Read timed out", e.getMessage());
-    }
-    Thread.sleep(100);
-    assertEquals("PONG", jedis.ping());
-
-    Jedis jedis1 = createJedis();
-    Jedis jedis2 = createJedis();
-    assertEquals("PONG", jedis1.ping());
-    assertEquals("PONG", jedis2.ping());
-    jedis.clientPause(1200);
-    try {
-      jedis1.ping();
-    } catch (Exception e) {
-      assertEquals("java.net.SocketTimeoutException: Read timed out", e.getMessage());
-    }
-    try {
-      jedis2.ping();
-    } catch (Exception e) {
-      assertEquals("java.net.SocketTimeoutException: Read timed out", e.getMessage());
-    }
-    Thread.sleep(200);
-    assertEquals("PONG", jedis1.ping());
-    assertEquals("PONG", jedis2.ping());
+      final Jedis jedisToPause1 = createJedis();
+      final Jedis jedisToPause2 = createJedis();
+
+      int pauseMillis = 1250;
+      jedis.clientPause(pauseMillis);
+
+      Future<Long> latency1 = executorService.submit(new Callable<Long>() {
+        @Override
+        public Long call() throws Exception {
+          long startMillis = System.currentTimeMillis();
+          assertEquals("PONG", jedisToPause1.ping());
+          return System.currentTimeMillis() - startMillis;
+        }
+      });
+      Future<Long> latency2 = executorService.submit(new Callable<Long>() {
+        @Override
+        public Long call() throws Exception {
+          long startMillis = System.currentTimeMillis();
+          assertEquals("PONG", jedisToPause2.ping());
+          return System.currentTimeMillis() - startMillis;
+        }
+      });
 
-    jedis1.close();
-    jedis2.close();
-  }
+      long latencyMillis1 = latency1.get();
+      long latencyMillis2 = latency2.get();
 
+      int pauseMillisDelta = 100;
+      assertTrue(pauseMillis <= latencyMillis1 && latencyMillis1 <= pauseMillis + pauseMillisDelta);
+      assertTrue(pauseMillis <= latencyMillis2 && latencyMillis2 <= pauseMillis + pauseMillisDelta);
+
+      jedisToPause1.close();
+      jedisToPause2.close();
+    } finally {
+      executorService.shutdown();
+      if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
+        executorService.shutdownNow();
+      }
+    }
+  }
 
   @Test
   public void memoryDoctorString() {
@@ -184,6 +198,4 @@ public void memoryDoctorBinary() {
     byte[] memoryInfo = jedis.memoryDoctorBinary();
     assertNotNull(memoryInfo);
   }
-
-
-}
+}