Handle exceptions on close in JedisSentinelPool (#2102)

When returning a Jedis object to the pool, handle any exceptions and still return the connection to prevent deadlock.
This is similar to issue and fix for #791, but for the JedisSentinelPool.

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

@@ -237,8 +237,13 @@ protected void returnBrokenResource(final Jedis resource) {
   @Override
   protected void returnResource(final Jedis resource) {
     if (resource != null) {
-      resource.resetState();
-      returnResourceObject(resource);
+      try {
+        resource.resetState();
+        returnResourceObject(resource);
+      } catch (Exception e) {
+        returnBrokenResource(resource);
+        throw new JedisException("Resource is returned to the pool as broken", e);
+      }
     }
   }
 

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

@@ -5,7 +5,11 @@
 
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.PooledObjectFactory;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.junit.Before;
 import org.junit.Test;
@@ -175,6 +179,61 @@ public void ensureSafeTwiceFailover() throws InterruptedException {
     // you can test failover as much as possible
   }
 
+  @Test
+  public void returnResourceDestroysResourceOnException() {
+    class CrashingJedis extends Jedis {
+      public CrashingJedis(final HostAndPort hp) {
+        super(hp);
+      }
+      @Override
+      public void resetState() {
+        throw new RuntimeException();
+      }
+    }
+
+    final AtomicInteger destroyed = new AtomicInteger(0);
+
+    class CrashingJedisPooledObjectFactory implements PooledObjectFactory<Jedis> {
+
+      @Override
+      public PooledObject<Jedis> makeObject() throws Exception {
+        return new DefaultPooledObject<Jedis>(new CrashingJedis(master));
+      }
+
+      @Override
+      public void destroyObject(PooledObject<Jedis> p) throws Exception {
+        destroyed.incrementAndGet();
+      }
+
+      @Override
+      public boolean validateObject(PooledObject<Jedis> p) {
+        return true;
+      }
+
+      @Override
+      public void activateObject(PooledObject<Jedis> p) throws Exception {
+      }
+
+      @Override
+      public void passivateObject(PooledObject<Jedis> p) throws Exception {
+      }
+    }
+
+    GenericObjectPoolConfig config = new GenericObjectPoolConfig();
+    config.setMaxTotal(1);
+    JedisSentinelPool pool = new JedisSentinelPool(MASTER_NAME, sentinels, config, 1000,
+        "foobared", 2);
+    pool.initPool(config, new CrashingJedisPooledObjectFactory());
+    Jedis crashingJedis = pool.getResource();
+
+    try {
+      crashingJedis.close();
+    } catch (Exception ignored) {
+    }
+
+    assertEquals(1, destroyed.get());
+  }
+
   private void forceFailover(JedisSentinelPool pool) throws InterruptedException {
     HostAndPort oldMaster = pool.getCurrentHostMaster();