Repository: redis/lettuce (java)

Issue:
`DefaultClientResources.dnsResolver(DnsResolver resolver)` should be marked as deprecated.
Just like definitions at its interface:

https://github.com/redis/lettuce/blob/cc9d2401e654810668d533a0637f32c6d3708b33/src/main/java/io/lettuce/core/resource/ClientResources.java#L164-L175

The implementation method should be marked as deprecated to inform users to use the new settings.

https://github.com/redis/lettuce/blob/cc9d2401e654810668d533a0637f32c6d3708b33/src/main/java/io/lettuce/core/resource/DefaultClientResources.java#L452

Pull request:
Mark dnsResolver(DnsResolver) as deprecated.
<!--
Thank you for proposing a pull request. This template will guide you through the essential steps necessary for a pull request.
-->
Make sure that:

- [x] You have read the [contribution guidelines](https://github.com/lettuce-io/lettuce-core/blob/main/.github/CONTRIBUTING.md).
- [ ] You have created a feature request first to discuss your contribution intent. Please reference the feature request ticket number in the pull request. (unrelated)
- [ ] You use the code formatters provided [here](https://github.com/lettuce-io/lettuce-core/blob/main/formatting.xml) and have them applied to your changes. Don’t submit any formatting related changes. (unrelated)
- [ ] You submit test cases (unit or integration tests) that back your changes. (unrelated)

 <!--
Great! Live long and prosper.
-->

Fixes #2854 .

Patch:
diff --git a/src/main/java/io/lettuce/core/cluster/ClusterPubSubConnectionProvider.java b/src/main/java/io/lettuce/core/cluster/ClusterPubSubConnectionProvider.java
index 747018ca6..2340ef934 100644
--- a/src/main/java/io/lettuce/core/cluster/ClusterPubSubConnectionProvider.java
+++ b/src/main/java/io/lettuce/core/cluster/ClusterPubSubConnectionProvider.java
@@ -177,6 +177,11 @@ class ClusterPubSubConnectionProvider<K, V> extends PooledClusterConnectionProvi
             notifications.ssubscribed(getNode(), channel, count);
         }
 
+        @Override
+        public void sunsubscribed(K channel, long count) {
+            notifications.sunsubscribed(getNode(), channel, count);
+        }
+
         private RedisClusterNode getNode() {
             return nodeId != null ? getPartitions().getPartitionByNodeId(nodeId) : getPartitions().getPartition(host, port);
         }
diff --git a/src/main/java/io/lettuce/core/cluster/PubSubClusterEndpoint.java b/src/main/java/io/lettuce/core/cluster/PubSubClusterEndpoint.java
index 9288dd8f8..9ec0a019c 100644
--- a/src/main/java/io/lettuce/core/cluster/PubSubClusterEndpoint.java
+++ b/src/main/java/io/lettuce/core/cluster/PubSubClusterEndpoint.java
@@ -94,6 +94,9 @@ public class PubSubClusterEndpoint<K, V> extends PubSubEndpoint<K, V> {
             case ssubscribe:
                 multicast.ssubscribed(clusterNode, output.channel(), output.count());
                 break;
+            case sunsubscribe:
+                multicast.sunsubscribed(clusterNode, output.channel(), output.count());
+                break;
             default:
                 throw new UnsupportedOperationException("Operation " + output.type() + " not supported");
         }
@@ -207,6 +210,12 @@ public class PubSubClusterEndpoint<K, V> extends PubSubEndpoint<K, V> {
             clusterListeners.forEach(listener -> listener.ssubscribed(node, channel, count));
         }
 
+        @Override
+        public void sunsubscribed(RedisClusterNode node, K channel, long count) {
+            getListeners().forEach(listener -> listener.sunsubscribed(channel, count));
+            clusterListeners.forEach(listener -> listener.sunsubscribed(node, channel, count));
+        }
+
     }
 
 }
diff --git a/src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubAdapter.java b/src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubAdapter.java
index 554efa300..3c7822af6 100644
--- a/src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubAdapter.java
+++ b/src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubAdapter.java
@@ -52,4 +52,9 @@ public class RedisClusterPubSubAdapter<K, V> implements RedisClusterPubSubListen
         // empty adapter method
     }
 
+    @Override
+    public void sunsubscribed(RedisClusterNode node, K channel, long count) {
+        // empty adapter method
+    }
+
 }
diff --git a/src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubListener.java b/src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubListener.java
index 93da2f531..b1755f80b 100644
--- a/src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubListener.java
+++ b/src/main/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubListener.java
@@ -91,4 +91,16 @@ public interface RedisClusterPubSubListener<K, V> {
         subscribed(node, shardChannel, count);
     }
 
+    /**
+     * Unsubscribed from a shard channel.
+     *
+     * @param node the {@link RedisClusterNode} from which the {@code message} originates.
+     * @param shardChannel Shard channel
+     * @param count Subscription count.
+     * @since 7.0
+     */
+    default void sunsubscribed(RedisClusterNode node, K shardChannel, long count) {
+        unsubscribed(node, shardChannel, count);
+    }
+
 }
diff --git a/src/main/java/io/lettuce/core/cluster/pubsub/api/async/NodeSelectionPubSubAsyncCommands.java b/src/main/java/io/lettuce/core/cluster/pubsub/api/async/NodeSelectionPubSubAsyncCommands.java
index 512afbaab..19f1980e0 100644
--- a/src/main/java/io/lettuce/core/cluster/pubsub/api/async/NodeSelectionPubSubAsyncCommands.java
+++ b/src/main/java/io/lettuce/core/cluster/pubsub/api/async/NodeSelectionPubSubAsyncCommands.java
@@ -51,4 +51,13 @@ public interface NodeSelectionPubSubAsyncCommands<K, V> {
      */
     AsyncExecutions<Void> ssubscribe(K... shardChannels);
 
+    /**
+     * Stop listening for messages posted to the given shard channels.
+     *
+     * @param shardChannels the channels
+     * @return RedisFuture&lt;Void&gt; Future to synchronize {@code unsubscribe} completion.
+     * @since 7.0
+     */
+    AsyncExecutions<Void> sunsubscribe(K... shardChannels);
+
 }
diff --git a/src/main/java/io/lettuce/core/cluster/pubsub/api/reactive/NodeSelectionPubSubReactiveCommands.java b/src/main/java/io/lettuce/core/cluster/pubsub/api/reactive/NodeSelectionPubSubReactiveCommands.java
index b61395cf2..3da91dddd 100644
--- a/src/main/java/io/lettuce/core/cluster/pubsub/api/reactive/NodeSelectionPubSubReactiveCommands.java
+++ b/src/main/java/io/lettuce/core/cluster/pubsub/api/reactive/NodeSelectionPubSubReactiveCommands.java
@@ -51,4 +51,13 @@ public interface NodeSelectionPubSubReactiveCommands<K, V> {
      */
     ReactiveExecutions<Void> ssubscribe(K... shardCchannels);
 
+    /**
+     * Stop listening for messages posted to the given shard channels.
+     *
+     * @param shardCchannels the channels
+     * @return RedisFuture&lt;Void&gt; Future to synchronize {@code unsubscribe} completion.
+     * @since 7.0
+     */
+    ReactiveExecutions<Void> sunsubscribe(K... shardCchannels);
+
 }
diff --git a/src/main/java/io/lettuce/core/cluster/pubsub/api/sync/NodeSelectionPubSubCommands.java b/src/main/java/io/lettuce/core/cluster/pubsub/api/sync/NodeSelectionPubSubCommands.java
index ba7fe76af..6a3716c9f 100644
--- a/src/main/java/io/lettuce/core/cluster/pubsub/api/sync/NodeSelectionPubSubCommands.java
+++ b/src/main/java/io/lettuce/core/cluster/pubsub/api/sync/NodeSelectionPubSubCommands.java
@@ -51,4 +51,13 @@ public interface NodeSelectionPubSubCommands<K, V> {
      */
     Executions<Void> ssubscribe(K... shardChannels);
 
+    /**
+     * Stop listening for messages posted to the given channels.
+     *
+     * @param shardChannels the channels
+     * @return Executions Future to synchronize {@code unsubscribe} completion.
+     * @since 7.0
+     */
+    Executions<Void> sunsubscribe(K... shardChannels);
+
 }
diff --git a/src/main/java/io/lettuce/core/protocol/CommandType.java b/src/main/java/io/lettuce/core/protocol/CommandType.java
index da1f6499c..37ae008fb 100644
--- a/src/main/java/io/lettuce/core/protocol/CommandType.java
+++ b/src/main/java/io/lettuce/core/protocol/CommandType.java
@@ -74,7 +74,7 @@ public enum CommandType implements ProtocolKeyword {
 
     // Pub/Sub
 
-    PSUBSCRIBE, PUBLISH, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, PUBSUB, SSUBSCRIBE, SPUBLISH,
+    PSUBSCRIBE, PUBLISH, PUNSUBSCRIBE, SUBSCRIBE, UNSUBSCRIBE, PUBSUB, SSUBSCRIBE, SPUBLISH, SUNSUBSCRIBE,
 
     // Sets
 
diff --git a/src/main/java/io/lettuce/core/pubsub/PubSubCommandBuilder.java b/src/main/java/io/lettuce/core/pubsub/PubSubCommandBuilder.java
index 3da52d877..65c226593 100644
--- a/src/main/java/io/lettuce/core/pubsub/PubSubCommandBuilder.java
+++ b/src/main/java/io/lettuce/core/pubsub/PubSubCommandBuilder.java
@@ -118,6 +118,14 @@ class PubSubCommandBuilder<K, V> extends BaseRedisCommandBuilder<K, V> {
         return pubSubCommand(SUBSCRIBE, new PubSubOutput<>(codec), channels);
     }
 
+    @SafeVarargs
+    final Command<K, V, V> sunsubscribe(K... shardChannels) {
+        LettuceAssert.notEmpty(shardChannels, "Shard channels " + MUST_NOT_BE_EMPTY);
+
+        CommandArgs<K, V> args = new CommandArgs<>(codec).addKeys(shardChannels);
+        return createCommand(SUNSUBSCRIBE, new PubSubOutput<>(codec), args);
+    }
+
     @SafeVarargs
     final Command<K, V, V> unsubscribe(K... channels) {
         return pubSubCommand(UNSUBSCRIBE, new PubSubOutput<>(codec), channels);
diff --git a/src/main/java/io/lettuce/core/pubsub/PubSubCommandHandler.java b/src/main/java/io/lettuce/core/pubsub/PubSubCommandHandler.java
index 0cb9bd1a5..30d811c71 100644
--- a/src/main/java/io/lettuce/core/pubsub/PubSubCommandHandler.java
+++ b/src/main/java/io/lettuce/core/pubsub/PubSubCommandHandler.java
@@ -227,6 +227,9 @@ public class PubSubCommandHandler<K, V> extends CommandHandler {
             case unsubscribe:
                 return commandType.equalsIgnoreCase("UNSUBSCRIBE");
 
+            case sunsubscribe:
+                return commandType.equalsIgnoreCase("SUNSUBSCRIBE");
+
             case punsubscribe:
                 return commandType.equalsIgnoreCase("PUNSUBSCRIBE");
         }
diff --git a/src/main/java/io/lettuce/core/pubsub/PubSubEndpoint.java b/src/main/java/io/lettuce/core/pubsub/PubSubEndpoint.java
index 6b6b267fe..6da336c4d 100644
--- a/src/main/java/io/lettuce/core/pubsub/PubSubEndpoint.java
+++ b/src/main/java/io/lettuce/core/pubsub/PubSubEndpoint.java
@@ -269,6 +269,9 @@ public class PubSubEndpoint<K, V> extends DefaultEndpoint {
                 case ssubscribe:
                     listener.ssubscribed(message.channel(), message.count());
                     break;
+                case sunsubscribe:
+                    listener.sunsubscribed(message.channel(), message.count());
+                    break;
                 default:
                     throw new UnsupportedOperationException("Operation " + message.type() + " not supported");
             }
@@ -293,6 +296,9 @@ public class PubSubEndpoint<K, V> extends DefaultEndpoint {
             case ssubscribe:
                 shardChannels.add(new Wrapper<>(message.channel()));
                 break;
+            case sunsubscribe:
+                shardChannels.remove(new Wrapper<>(message.channel()));
+                break;
             default:
                 break;
         }
diff --git a/src/main/java/io/lettuce/core/pubsub/PubSubOutput.java b/src/main/java/io/lettuce/core/pubsub/PubSubOutput.java
index 18c19ccdd..0ec41ff64 100644
--- a/src/main/java/io/lettuce/core/pubsub/PubSubOutput.java
+++ b/src/main/java/io/lettuce/core/pubsub/PubSubOutput.java
@@ -38,7 +38,7 @@ public class PubSubOutput<K, V> extends CommandOutput<K, V, V> implements PubSub
 
     public enum Type {
 
-        message, pmessage, psubscribe, punsubscribe, subscribe, unsubscribe, ssubscribe, smessage;
+        message, pmessage, psubscribe, punsubscribe, subscribe, unsubscribe, ssubscribe, smessage, sunsubscribe;
 
         private final static Set<String> names = new HashSet<>();
 
@@ -124,6 +124,7 @@ public class PubSubOutput<K, V> extends CommandOutput<K, V, V> implements PubSub
             case subscribe:
             case unsubscribe:
             case ssubscribe:
+            case sunsubscribe:
                 channel = codec.decodeKey(bytes);
                 break;
             default:
diff --git a/src/main/java/io/lettuce/core/pubsub/RedisPubSubAdapter.java b/src/main/java/io/lettuce/core/pubsub/RedisPubSubAdapter.java
index 8b061bacf..20f03c1bb 100644
--- a/src/main/java/io/lettuce/core/pubsub/RedisPubSubAdapter.java
+++ b/src/main/java/io/lettuce/core/pubsub/RedisPubSubAdapter.java
@@ -69,4 +69,9 @@ public class RedisPubSubAdapter<K, V> implements RedisPubSubListener<K, V> {
         // empty adapter method
     }
 
+    @Override
+    public void sunsubscribed(K shardChannel, long count) {
+        // empty adapter method
+    }
+
 }
diff --git a/src/main/java/io/lettuce/core/pubsub/RedisPubSubAsyncCommandsImpl.java b/src/main/java/io/lettuce/core/pubsub/RedisPubSubAsyncCommandsImpl.java
index 20515e60f..3e282b541 100644
--- a/src/main/java/io/lettuce/core/pubsub/RedisPubSubAsyncCommandsImpl.java
+++ b/src/main/java/io/lettuce/core/pubsub/RedisPubSubAsyncCommandsImpl.java
@@ -111,6 +111,12 @@ public class RedisPubSubAsyncCommandsImpl<K, V> extends RedisAsyncCommandsImpl<K
         return (RedisFuture<Void>) dispatch(commandBuilder.ssubscribe(channels));
     }
 
+    @Override
+    @SuppressWarnings("unchecked")
+    public RedisFuture<Void> sunsubscribe(K... channels) {
+        return (RedisFuture<Void>) dispatch(commandBuilder.sunsubscribe(channels));
+    }
+
     @Override
     @SuppressWarnings("unchecked")
     public StatefulRedisPubSubConnection<K, V> getStatefulConnection() {
diff --git a/src/main/java/io/lettuce/core/pubsub/RedisPubSubListener.java b/src/main/java/io/lettuce/core/pubsub/RedisPubSubListener.java
index ccc1f0e1e..ae0aff2e4 100644
--- a/src/main/java/io/lettuce/core/pubsub/RedisPubSubListener.java
+++ b/src/main/java/io/lettuce/core/pubsub/RedisPubSubListener.java
@@ -88,6 +88,17 @@ public interface RedisPubSubListener<K, V> {
         subscribed(shardChannel, count);
     }
 
+    /**
+     * Unsubscribed from a shard channel.
+     *
+     * @param shardChannel Channel
+     * @param count Subscription count.
+     * @since 7.0
+     */
+    default void sunsubscribed(K shardChannel, long count) {
+        unsubscribed(shardChannel, count);
+    }
+
     /**
      * Message received from a shard channel subscription.
      *
diff --git a/src/main/java/io/lettuce/core/pubsub/RedisPubSubReactiveCommandsImpl.java b/src/main/java/io/lettuce/core/pubsub/RedisPubSubReactiveCommandsImpl.java
index adff1b58f..19e878f3f 100644
--- a/src/main/java/io/lettuce/core/pubsub/RedisPubSubReactiveCommandsImpl.java
+++ b/src/main/java/io/lettuce/core/pubsub/RedisPubSubReactiveCommandsImpl.java
@@ -169,6 +169,11 @@ public class RedisPubSubReactiveCommandsImpl<K, V> extends RedisReactiveCommands
         return createFlux(() -> commandBuilder.ssubscribe(shardChannels)).then();
     }
 
+    @Override
+    public Mono<Void> sunsubscribe(K... shardChannels) {
+        return createFlux(() -> commandBuilder.sunsubscribe(shardChannels)).then();
+    }
+
     @Override
     @SuppressWarnings("unchecked")
     public StatefulRedisPubSubConnection<K, V> getStatefulConnection() {
diff --git a/src/main/java/io/lettuce/core/pubsub/api/async/RedisPubSubAsyncCommands.java b/src/main/java/io/lettuce/core/pubsub/api/async/RedisPubSubAsyncCommands.java
index 90c301a58..875058855 100644
--- a/src/main/java/io/lettuce/core/pubsub/api/async/RedisPubSubAsyncCommands.java
+++ b/src/main/java/io/lettuce/core/pubsub/api/async/RedisPubSubAsyncCommands.java
@@ -60,4 +60,13 @@ public interface RedisPubSubAsyncCommands<K, V> extends RedisAsyncCommands<K, V>
      */
     RedisFuture<Void> ssubscribe(K... shardChannels);
 
+    /**
+     * Stop listening for messages posted to the given channels.
+     *
+     * @param shardChannels the shard channels
+     * @return RedisFuture&lt;Void&gt; Future to synchronize {@code unsubscribe} completion.
+     * @since 7.0
+     */
+    RedisFuture<Void> sunsubscribe(K... shardChannels);
+
 }
diff --git a/src/main/java/io/lettuce/core/pubsub/api/reactive/RedisPubSubReactiveCommands.java b/src/main/java/io/lettuce/core/pubsub/api/reactive/RedisPubSubReactiveCommands.java
index d6f0476d6..6f1f0e081 100644
--- a/src/main/java/io/lettuce/core/pubsub/api/reactive/RedisPubSubReactiveCommands.java
+++ b/src/main/java/io/lettuce/core/pubsub/api/reactive/RedisPubSubReactiveCommands.java
@@ -107,6 +107,16 @@ public interface RedisPubSubReactiveCommands<K, V> extends RedisReactiveCommands
      */
     Mono<Void> ssubscribe(K... shardChannels);
 
+    /**
+     * Stop listening for messages posted to the given channels. The {@link Mono} completes without a result as soon as the
+     * subscription is unregistered.
+     *
+     * @param shardChannels the channels.
+     * @return Mono&lt;Void&gt; Mono for {@code unsubscribe} command.
+     * @since 7.0
+     */
+    Mono<Void> sunsubscribe(K... shardChannels);
+
     /**
      * @return the underlying connection.
      * @since 6.2, will be removed with Lettuce 7 to avoid exposing the underlying connection.
diff --git a/src/main/java/io/lettuce/core/pubsub/api/sync/RedisPubSubCommands.java b/src/main/java/io/lettuce/core/pubsub/api/sync/RedisPubSubCommands.java
index 44a59b6c6..b3ea69fb6 100644
--- a/src/main/java/io/lettuce/core/pubsub/api/sync/RedisPubSubCommands.java
+++ b/src/main/java/io/lettuce/core/pubsub/api/sync/RedisPubSubCommands.java
@@ -49,6 +49,14 @@ public interface RedisPubSubCommands<K, V> extends RedisCommands<K, V> {
      */
     void ssubscribe(K... shardChannels);
 
+    /**
+     * Stop listening for messages posted to the given channels.
+     *
+     * @param shardChannels the channels
+     * @since 7.0
+     */
+    void sunsubscribe(K... shardChannels);
+
     /**
      * @return the underlying connection.
      */
diff --git a/src/main/java/io/lettuce/core/resource/DefaultClientResources.java b/src/main/java/io/lettuce/core/resource/DefaultClientResources.java
index 1922bfef6..592ffff4a 100644
--- a/src/main/java/io/lettuce/core/resource/DefaultClientResources.java
+++ b/src/main/java/io/lettuce/core/resource/DefaultClientResources.java
@@ -447,7 +447,9 @@ public class DefaultClientResources implements ClientResources {
          * @param dnsResolver the DNS resolver, must not be {@code null}.
          * @return {@code this} {@link Builder}.
          * @since 4.3
+         * @deprecated since 6.1. Configure {@link AddressResolverGroup} instead.
          */
+        @Deprecated
         @Override
         public Builder dnsResolver(DnsResolver dnsResolver) {
 


Test patch:
diff --git a/src/test/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubConnectionIntegrationTests.java b/src/test/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubConnectionIntegrationTests.java
index 772af41da..cd08f520e 100644
--- a/src/test/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubConnectionIntegrationTests.java
+++ b/src/test/java/io/lettuce/core/cluster/pubsub/RedisClusterPubSubConnectionIntegrationTests.java
@@ -211,6 +211,22 @@ class RedisClusterPubSubConnectionIntegrationTests extends TestSupport {
         cmd.getStatefulConnection().close();
     }
 
+    @Test
+    @EnabledOnCommand("SSUBSCRIBE")
+    void unubscribeFromShardChannel() {
+        pubSubConnection.sync().ssubscribe(shardChannel);
+        pubSubConnection.sync().spublish(shardChannel, "msg1");
+
+        pubSubConnection.sync().sunsubscribe(shardChannel);
+        pubSubConnection.sync().spublish(shardChannel, "msg2");
+
+        pubSubConnection.sync().ssubscribe(shardChannel);
+        pubSubConnection.sync().spublish(shardChannel, "msg3");
+
+        Wait.untilEquals("msg1", connectionListener.getMessages()::poll).waitOrTimeout();
+        Wait.untilEquals("msg3", connectionListener.getMessages()::poll).waitOrTimeout();
+    }
+
     @Test
     void myIdWorksAfterDisconnect() throws InterruptedException {
 
diff --git a/src/test/java/io/lettuce/core/pubsub/PubSubCommandBuilderUnitTests.java b/src/test/java/io/lettuce/core/pubsub/PubSubCommandBuilderUnitTests.java
index 4d4e79818..89e93b234 100644
--- a/src/test/java/io/lettuce/core/pubsub/PubSubCommandBuilderUnitTests.java
+++ b/src/test/java/io/lettuce/core/pubsub/PubSubCommandBuilderUnitTests.java
@@ -139,4 +139,15 @@ class PubSubCommandBuilderUnitTests {
         assertThat(command.getOutput()).isInstanceOf(PubSubOutput.class);
     }
 
+    @Test
+    void sunsubscribe() {
+        String channel = "channelPattern";
+        Command<String, String, String> command = this.commandBuilder.sunsubscribe(channel);
+
+        assertThat(command.getType()).isEqualTo(SUNSUBSCRIBE);
+        assertThat(command.getArgs()).isInstanceOf(CommandArgs.class);
+        assertThat(command.getArgs().toCommandString()).isEqualTo("key<channelPattern>");
+        assertThat(command.getOutput()).isInstanceOf(PubSubOutput.class);
+    }
+
 }
diff --git a/src/test/java/io/lettuce/core/pubsub/PubSubReactiveTest.java b/src/test/java/io/lettuce/core/pubsub/PubSubReactiveTest.java
index e5d91a156..7dd7d1100 100644
--- a/src/test/java/io/lettuce/core/pubsub/PubSubReactiveTest.java
+++ b/src/test/java/io/lettuce/core/pubsub/PubSubReactiveTest.java
@@ -357,6 +357,13 @@ class PubSubReactiveTest extends AbstractRedisClientTest implements RedisPubSubL
         assertThat((long) counts.take()).isGreaterThan(0);
     }
 
+    @Test
+    void sunsubscribe() throws Exception {
+        StepVerifier.create(pubsub.sunsubscribe(channel)).verifyComplete();
+        assertThat(shardChannels.take()).isEqualTo(channel);
+        assertThat((long) counts.take()).isEqualTo(0);
+    }
+
     @Test
     void pubsubCloseOnClientShutdown() {
 
@@ -515,6 +522,12 @@ class PubSubReactiveTest extends AbstractRedisClientTest implements RedisPubSubL
         counts.add(count);
     }
 
+    @Override
+    public void sunsubscribed(String shardChannel, long count) {
+        shardChannels.add(shardChannel);
+        counts.add(count);
+    }
+
     <T> T block(Mono<T> mono) {
         return mono.block();
     }
diff --git a/src/test/java/io/lettuce/core/pubsub/RedisPubSubAsyncCommandsImplUnitTests.java b/src/test/java/io/lettuce/core/pubsub/RedisPubSubAsyncCommandsImplUnitTests.java
index 65b0a960a..49dee5c2d 100644
--- a/src/test/java/io/lettuce/core/pubsub/RedisPubSubAsyncCommandsImplUnitTests.java
+++ b/src/test/java/io/lettuce/core/pubsub/RedisPubSubAsyncCommandsImplUnitTests.java
@@ -15,7 +15,6 @@ import java.util.concurrent.ExecutionException;
 
 import static io.lettuce.core.protocol.CommandType.*;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.Assert.*;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.*;
@@ -50,7 +49,7 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(PubSubOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("key<channelPattern>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
     @Test
@@ -69,7 +68,7 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(PubSubOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("key<channelPattern>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
     @Test
@@ -88,7 +87,7 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(PubSubOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("key<channelPattern>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
     @Test
@@ -107,7 +106,7 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(PubSubOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("key<channelPattern>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
     @Test
@@ -128,7 +127,7 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(IntegerOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("key<acmeChannel> value<acmeMessage>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
     @Test
@@ -148,7 +147,7 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(KeyListOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("CHANNELS key<channelPattern>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
     @Test
@@ -168,7 +167,7 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(MapOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("NUMSUB key<channelPattern>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
     @Test
@@ -188,7 +187,7 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(KeyListOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("SHARDCHANNELS key<channelPattern>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
     @Test
@@ -208,7 +207,7 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(MapOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("SHARDNUMSUB key<channelPattern>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
     @Test
@@ -227,7 +226,26 @@ class RedisPubSubAsyncCommandsImplUnitTests {
         assertInstanceOf(PubSubOutput.class, capturedCommand.getValue().getOutput());
         assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("key<channelPattern>");
 
-        assertNotEquals(capturedCommand.getValue(), dispachedMock);
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
+    }
+
+    @Test
+    void sunsubscribe() throws ExecutionException, InterruptedException {
+        String pattern = "channelPattern";
+        AsyncCommand dispachedMock = mock(AsyncCommand.class);
+        when(mockedConnection.dispatch((RedisCommand<String, String, Object>) any())).thenReturn(dispachedMock);
+
+        commands.sunsubscribe(pattern).get();
+
+        ArgumentCaptor<AsyncCommand> capturedCommand = ArgumentCaptor.forClass(AsyncCommand.class);
+
+        verify(mockedConnection).dispatch(capturedCommand.capture());
+
+        assertThat(capturedCommand.getValue().getType()).isEqualTo(SUNSUBSCRIBE);
+        assertInstanceOf(PubSubOutput.class, capturedCommand.getValue().getOutput());
+        assertThat(capturedCommand.getValue().getArgs().toCommandString()).isEqualTo("key<channelPattern>");
+
+        assertThat(capturedCommand.getValue()).isNotEqualTo(dispachedMock);
     }
 
 }
diff --git a/src/test/java/io/lettuce/core/support/PubSubTestListener.java b/src/test/java/io/lettuce/core/support/PubSubTestListener.java
index 1afcb4276..fa7d0f6fc 100644
--- a/src/test/java/io/lettuce/core/support/PubSubTestListener.java
+++ b/src/test/java/io/lettuce/core/support/PubSubTestListener.java
@@ -74,6 +74,12 @@ public class PubSubTestListener implements RedisPubSubListener<String, String> {
         shardCounts.add(count);
     }
 
+    @Override
+    public void sunsubscribed(String shardChannel, long count) {
+        shardChannels.add(shardChannel);
+        counts.add(count);
+    }
+
     @Override
     public void psubscribed(String pattern, long count) {
         patterns.add(pattern);