DATAREDIS-684 - Release Jedis cluster node connections with close().

We now release Jedis cluster node connections with Jedis.close() to the pool instead of Pool.returnResource(…). The close() method itself checks whether the connection was broken and if so, the connection gets destroyed. Destroying broken connections prevents the pool from supplying broken connections on borrow when testOnBorrow is disabled.

The only case where we return broken resources ourselves to the Pool is when we discover a broken connection ourselves: If we run into a NullPointerException or RedisConnectionFailureException, then we consider a connection is broken.

Original Pull Request: #271

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterConnection.java

@@ -840,7 +840,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 		 */
 		@Override
 		public void returnResourceForSpecificNode(RedisClusterNode node, Object client) {
-			getResourcePoolForSpecificNode(node).returnResource((Jedis) client);
+			((Jedis) client).close();
 		}
 	}
 
@@ -897,7 +897,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 					errors.put(entry.getKey(), ex);
 				} finally {
 					if (jedis != null) {
-						entry.getValue().returnResource(jedis);
+						jedis.close();
 					}
 				}
 			}

src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java

@@ -343,29 +343,27 @@ public class JedisConnection extends AbstractRedisConnection {
 	 */
 	@Override
 	public void close() throws DataAccessException {
+
 		super.close();
+
 		// return the connection to the pool
 		if (pool != null) {
-			if (!broken) {
+
+			if (broken) {
+				pool.returnBrokenResource(jedis);
+			} else {
+
 				// reset the connection
 				try {
 					if (dbIndex > 0) {
 						jedis.select(0);
 					}
-					pool.returnResource(jedis);
 					return;
 				} catch (Exception ex) {
-					DataAccessException dae = convertJedisAccessException(ex);
-					if (broken) {
-						pool.returnBrokenResource(jedis);
-					} else {
-						pool.returnResource(jedis);
-					}
-					throw dae;
+					throw convertJedisAccessException(ex);
+				} finally {
+					jedis.close();
 				}
-			} else {
-				pool.returnBrokenResource(jedis);
-				return;
 			}
 		}
 		// else close the connection normally (doing the try/catch dance)
@@ -373,13 +371,13 @@ public class JedisConnection extends AbstractRedisConnection {
 		if (isQueueing()) {
 			try {
 				client.quit();
-			} catch (Exception ex) {
-				exc = ex;
+			} catch (Exception o_O) {
+				// ignore exception
 			}
 			try {
 				client.disconnect();
-			} catch (Exception ex) {
-				exc = ex;
+			} catch (Exception o_O) {
+				// ignore exception
 			}
 			return;
 		}