Repository: redis/lettuce (java)

Issue:
Add support for `SUNSUBSCRIBE`
Add support for the command as documented [here](https://redis.io/commands/sunsubscribe)

Pull request:
Add support for `SUNSUBSCRIBE` #2759
Closes/Fixes #2759 

- [X] You have read the [contribution guidelines](https://github.com/lettuce-io/lettuce-core/blob/main/.github/CONTRIBUTING.md).
- [X] You have created a feature request first to discuss your contribution intent. Please reference the feature request ticket number in the pull request.
- [X] 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.
- [X] You submit test cases (unit or integration tests) that back your changes


Patch:
diff --git a/src/main/java/io/lettuce/core/KillArgs.java b/src/main/java/io/lettuce/core/KillArgs.java
index 2a71b051d..d24325c50 100644
--- a/src/main/java/io/lettuce/core/KillArgs.java
+++ b/src/main/java/io/lettuce/core/KillArgs.java
@@ -54,6 +54,8 @@ public class KillArgs implements CompositeArgument {
 
     private String username;
 
+    private Long maxAge;
+
     /**
      * Builder entry points for {@link KillArgs}.
      */
@@ -160,6 +162,17 @@ public class KillArgs implements CompositeArgument {
             return new KillArgs().user(username);
         }
 
+        /**
+         * Creates new {@link KillArgs} setting {@literal MAXAGE}.
+         *
+         * @return new {@link KillArgs} with {@literal MAXAGE} set.
+         * @see KillArgs#maxAge(Long)
+         * @since 7.0
+         */
+        public static KillArgs maxAge(Long maxAge) {
+            return new KillArgs().maxAge(maxAge);
+        }
+
     }
 
     /**
@@ -241,6 +254,21 @@ public class KillArgs implements CompositeArgument {
         return this;
     }
 
+    /**
+     * Closes all the connections that are older than the specified age, in seconds.
+     *
+     * @param maxAge must not be {@code null}.
+     * @return {@code this} {@link KillArgs}.
+     * @since 7.0
+     */
+    public KillArgs maxAge(Long maxAge) {
+
+        LettuceAssert.notNull(maxAge, "MaxAge must not be null");
+
+        this.maxAge = maxAge;
+        return this;
+    }
+
     /**
      * Closes all the connections that are authenticated with the specified ACL {@code username}.
      *
@@ -282,6 +310,10 @@ public class KillArgs implements CompositeArgument {
         if (username != null) {
             args.add("USER").add(username);
         }
+
+        if (maxAge != null) {
+            args.add("MAXAGE").add(maxAge);
+        }
     }
 
 }
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.
      */


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/commands/ServerCommandIntegrationTests.java b/src/test/java/io/lettuce/core/commands/ServerCommandIntegrationTests.java
index 610079abf..b9be862e7 100644
--- a/src/test/java/io/lettuce/core/commands/ServerCommandIntegrationTests.java
+++ b/src/test/java/io/lettuce/core/commands/ServerCommandIntegrationTests.java
@@ -33,6 +33,7 @@ import java.util.regex.Pattern;
 
 import javax.inject.Inject;
 
+import io.lettuce.test.condition.RedisConditions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
@@ -186,6 +187,24 @@ public class ServerCommandIntegrationTests extends TestSupport {
         redis.aclDeluser("test_kill");
     }
 
+    @Test
+    void clientKillMaxAge() throws InterruptedException {
+        // can not find other new command to use `@EnabledOnCommand` for now, so check the version
+        assumeTrue(RedisConditions.of(redis).hasVersionGreaterOrEqualsTo("8.0"));
+
+        RedisCommands<String, String> connection2 = client.connect().sync();
+        long inactiveId = connection2.clientId();
+        long maxAge = 2L;
+        // sleep for maxAge * 2 seconds, to be sure
+        TimeUnit.SECONDS.sleep(maxAge * 2);
+        RedisCommands<String, String> connection3 = client.connect().sync();
+        long activeId = connection3.clientId();
+        assertThat(redis.clientKill(KillArgs.Builder.maxAge(maxAge))).isGreaterThan(0);
+
+        assertThat(redis.clientList(ClientListArgs.Builder.ids(inactiveId))).isBlank();
+        assertThat(redis.clientList(ClientListArgs.Builder.ids(activeId))).isNotBlank();
+    }
+
     @Test
     void clientId() {
         assertThat(redis.clientId()).isNotNull();
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);