Unsubscribe only once on Subscription.close().

We now unsubscribe from Redis only once when closing a Subscription object to avoid duplicate traffic and unexpected Redis responses.

Jedis does not have a mechanism to consume the additional unsubscribe response which leaves protocol frames on the InputStream leading to a corrupt state.

Closes #2355

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

@@ -42,6 +42,7 @@ class JedisSubscription extends AbstractSubscription {
 	 */
 	@Override
 	protected void doClose() {
+
 		if (!getChannels().isEmpty()) {
 			jedisPubSub.unsubscribe();
 		}

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceSubscription.java

@@ -80,10 +80,6 @@ public class LettuceSubscription extends AbstractSubscription {
 	@Override
 	protected void doClose() {
 
-		if (!isAlive()) {
-			return;
-		}
-
 		List<CompletableFuture<?>> futures = new ArrayList<>();
 
 		if (!getChannels().isEmpty()) {

src/main/java/org/springframework/data/redis/connection/util/AbstractSubscription.java

@@ -103,8 +103,19 @@ public abstract class AbstractSubscription implements Subscription {
 	 */
 	@Override
 	public void close() {
-		doClose();
-		alive.set(false);
+
+		if (alive.compareAndSet(true, false)) {
+
+			doClose();
+
+			synchronized (channels) {
+				channels.clear();
+			}
+
+			synchronized (patterns) {
+				patterns.clear();
+			}
+		}
 	}
 
 	/**

src/test/java/org/springframework/data/redis/connection/jedis/JedisSubscriptionUnitTests.java

@@ -35,6 +35,7 @@ import org.springframework.data.redis.connection.RedisInvalidSubscriptionExcepti
  * Unit test of {@link JedisSubscription}
  *
  * @author Jennifer Hickey
+ * @author Mark Paluch
  */
 @ExtendWith(MockitoExtension.class)
 class JedisSubscriptionUnitTests {
@@ -305,4 +306,15 @@ class JedisSubscriptionUnitTests {
 		verify(jedisPubSub, times(1)).punsubscribe();
 	}
 
+	@Test // GH-2355
+	void closeTwiceShouldUnsubscribeOnce() {
+
+		subscription.subscribe(new byte[][] { "a".getBytes() });
+
+		subscription.close();
+		subscription.close();
+
+		verify(jedisPubSub, times(1)).unsubscribe();
+	}
+
 }