Repository: redis/lettuce (java)

Issue:
event loop thread gets blocked during disconnection/reconnection
## Bug Report

<!-- First of all: Have you checked the docs https://github.com/lettuce-io/lettuce-core/wiki, GitHub issues, or Stack Overflow whether someone else has already reported your issue? -->

#### Current Behavior

event loop thread gets blocked during disconnection/reconnection

#### Input Code

<!-- Java/Kotlin/Scala/Groovy/… or Repo link if applicable: -->


```java
public class MultiThreadSyncGet {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(MultiThreadSyncGet.class);

    private static final int LOOP_NUM = 10_000_000;

    private static final int BATCH_SIZE = 1000;

    private static final int DIGIT_NUM = 9;

    private static final String KEY_FORMATTER = String.format("key-%%0%dd", DIGIT_NUM);

    static {
        // noinspection ConstantValue
        LettuceAssert.assertState(DIGIT_NUM >= String.valueOf(LOOP_NUM).length() + 1, "digit num is not large enough");
    }

    void test() {
        RedisURI uri = RedisURI.create("redis.dev-d-okex.svc.dev.local", 6379);
        uri.setCredentialsProvider(new StaticCredentialsProvider(null, "123qweasd!@#".toCharArray()));
        try (RedisClient redisClient = RedisClient.create(uri)) {
            final ClientOptions.Builder optsBuilder = ClientOptions.builder()
                    .timeoutOptions(TimeoutOptions.builder().fixedTimeout(Duration.ofSeconds(60)).build());
            redisClient.setOptions(optsBuilder.build());
            final StatefulRedisConnection<byte[], byte[]> connection = redisClient.connect(ByteArrayCodec.INSTANCE);
            connection.setAutoFlushCommands(false);

            new Thread(() -> {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    logger.error("interrupted", e);
                }
                connection.close();
            }).start();

            for (int j = 0; j < LOOP_NUM && connection.isOpen(); j++) {
                for (int i = 0; i < BATCH_SIZE; i++) {
                    connection.async().get(genKey(j));
                }
                connection.flushCommands();
            }
        }
    }

    private byte[] genKey(int j) {
        return String.format(KEY_FORMATTER, j).getBytes();
    }

    public static void main(String[] args) {
        new MultiThreadSyncGet().test();
        logger.info("=====================================");
    }

}
```

I manipulate network using MacOS's `Network Link Conditioner`, set `Downlink Delay`  to 1000ms, then set debug breakpoint at  `stack.remove(command)`
```java
 public void operationComplete(Future<Void> future) {

            try {
                if (!future.isSuccess()) {
                    stack.remove(command);
                }
            } finally {
                recycle();
            }
        }
```
Then evaluate
```java
new ArrayList<>(stack).indexOf(command)
```
got `3971`.

evaluate
```java
stack.size()
```
got `85000`.

we can conclude the compute complexity to remove all the failed-to-flush commands is O(3971*(85000-3971)), which is super big. If there are more successfully flushed commands before the first failed-to-flush command, the complexity could be even higher.
</details>

#### Expected behavior/code

Thread not blocked

#### Environment

- Lettuce version(s): git_sha_dee8020d92e6bff562cef723dcacdea714b89982
- Redis version: 7.0.0

#### Possible Solution

Use HashIndexedQueue instead which provides O(1) complexity to remove an element from the queue.
This is due to ArrayDeque#remove is O(n), if there are lots of tasks failing at AddToStask#operationComplete, then the event loop thread may get blocked for too long

#### Additional context
problematic code: 
```java
    static class AddToStack implements GenericFutureListener<Future<Void>> {
       ...

        AddToStack(Recycler.Handle<AddToStack> handle) {
            this.handle = handle;
        }

        @SuppressWarnings("unchecked")
        @Override
        public void operationComplete(Future<Void> future) {

            try {
                if (!future.isSuccess()) {
                    stack.remove(command);
                }
            } finally {
                recycle();
            }
        }
}
```




Pull request:
fix: prevent blocking event loop thread by replacing ArrayDeque with HashIndexedQueue
Closes #2937

Make sure that:

- [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 applied code formatting rules using the `mvn formatter:format` target. 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/ClientOptions.java b/src/main/java/io/lettuce/core/ClientOptions.java
index 9f1f1c33d..52a5a950d 100644
--- a/src/main/java/io/lettuce/core/ClientOptions.java
+++ b/src/main/java/io/lettuce/core/ClientOptions.java
@@ -85,6 +85,8 @@ public class ClientOptions implements Serializable {
 
     public static final TimeoutOptions DEFAULT_TIMEOUT_OPTIONS = TimeoutOptions.enabled();
 
+    public static final boolean DEFAULT_USE_HASH_INDEX_QUEUE = true;
+
     private final boolean autoReconnect;
 
     private final boolean cancelCommandsOnReconnectFailure;
@@ -115,6 +117,8 @@ public class ClientOptions implements Serializable {
 
     private final TimeoutOptions timeoutOptions;
 
+    private final boolean useHashIndexedQueue;
+
     protected ClientOptions(Builder builder) {
         this.autoReconnect = builder.autoReconnect;
         this.cancelCommandsOnReconnectFailure = builder.cancelCommandsOnReconnectFailure;
@@ -131,6 +135,7 @@ public class ClientOptions implements Serializable {
         this.sslOptions = builder.sslOptions;
         this.suspendReconnectOnProtocolFailure = builder.suspendReconnectOnProtocolFailure;
         this.timeoutOptions = builder.timeoutOptions;
+        this.useHashIndexedQueue = builder.useHashIndexedQueue;
     }
 
     protected ClientOptions(ClientOptions original) {
@@ -149,6 +154,7 @@ public class ClientOptions implements Serializable {
         this.sslOptions = original.getSslOptions();
         this.suspendReconnectOnProtocolFailure = original.isSuspendReconnectOnProtocolFailure();
         this.timeoutOptions = original.getTimeoutOptions();
+        this.useHashIndexedQueue = original.isUseHashIndexedQueue();
     }
 
     /**
@@ -214,6 +220,8 @@ public class ClientOptions implements Serializable {
 
         private TimeoutOptions timeoutOptions = DEFAULT_TIMEOUT_OPTIONS;
 
+        private boolean useHashIndexedQueue = DEFAULT_USE_HASH_INDEX_QUEUE;
+
         protected Builder() {
         }
 
@@ -269,8 +277,8 @@ public class ClientOptions implements Serializable {
          *
          * @param policy the policy to use in {@link io.lettuce.core.protocol.CommandHandler}
          * @return {@code this}
-         * @since 6.0
          * @see DecodeBufferPolicies
+         * @since 6.0
          */
         public Builder decodeBufferPolicy(DecodeBufferPolicy policy) {
 
@@ -317,8 +325,8 @@ public class ClientOptions implements Serializable {
          *
          * @param protocolVersion version to use.
          * @return {@code this}
-         * @since 6.0
          * @see ProtocolVersion#newestSupported()
+         * @since 6.0
          */
         public Builder protocolVersion(ProtocolVersion protocolVersion) {
 
@@ -337,9 +345,9 @@ public class ClientOptions implements Serializable {
          *
          * @param publishOnScheduler true/false
          * @return {@code this}
-         * @since 5.2
          * @see org.reactivestreams.Subscriber#onNext(Object)
          * @see ClientResources#eventExecutorGroup()
+         * @since 5.2
          */
         public Builder publishOnScheduler(boolean publishOnScheduler) {
             this.publishOnScheduler = publishOnScheduler;
@@ -459,6 +467,20 @@ public class ClientOptions implements Serializable {
             return this;
         }
 
+        /**
+         * Use hash indexed queue, which provides O(1) remove(Object) thus won't cause blocking issues.
+         *
+         * @param useHashIndexedQueue true/false
+         * @return {@code this}
+         * @see io.lettuce.core.protocol.CommandHandler.AddToStack
+         * @since 6.6
+         */
+        @SuppressWarnings("JavadocReference")
+        public Builder useHashIndexQueue(boolean useHashIndexedQueue) {
+            this.useHashIndexedQueue = useHashIndexedQueue;
+            return this;
+        }
+
         /**
          * Create a new instance of {@link ClientOptions}.
          *
@@ -476,7 +498,6 @@ public class ClientOptions implements Serializable {
      *
      * @return a {@link ClientOptions.Builder} to create new {@link ClientOptions} whose settings are replicated from the
      *         current {@link ClientOptions}.
-     *
      * @since 5.1
      */
     public ClientOptions.Builder mutate() {
@@ -535,7 +556,6 @@ public class ClientOptions implements Serializable {
      *
      * @return zero.
      * @since 5.2
-     *
      * @deprecated since 6.0 in favor of {@link DecodeBufferPolicy}.
      */
     @Deprecated
@@ -684,6 +704,15 @@ public class ClientOptions implements Serializable {
         return timeoutOptions;
     }
 
+    /**
+     * Whether we should use hash indexed queue, which provides O(1) remove(Object)
+     *
+     * @return if hash indexed queue should be used
+     */
+    public boolean isUseHashIndexedQueue() {
+        return useHashIndexedQueue;
+    }
+
     /**
      * Behavior of connections in disconnected state.
      */
diff --git a/src/main/java/io/lettuce/core/datastructure/queue/HashIndexedQueue.java b/src/main/java/io/lettuce/core/datastructure/queue/HashIndexedQueue.java
new file mode 100644
index 000000000..b265ba456
--- /dev/null
+++ b/src/main/java/io/lettuce/core/datastructure/queue/HashIndexedQueue.java
@@ -0,0 +1,266 @@
+/*
+ * Copyright 2011-Present, Redis Ltd. and Contributors
+ * All rights reserved.
+ *
+ * Licensed under the MIT License.
+ */
+package io.lettuce.core.datastructure.queue;
+
+import java.util.AbstractQueue;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+import io.lettuce.core.internal.LettuceAssert;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * A queue implementation that supports O(1) removal of elements. The queue is backed by a hash map and a doubly linked list.
+ * 
+ * @author chenxiaofan
+ */
+@SuppressWarnings("unchecked")
+public class HashIndexedQueue<E> extends AbstractQueue<E> {
+
+    private final Map<E, Object> map; // Object can be Node<E> or List<Node<E>>
+
+    private Node<E> head;
+
+    private Node<E> tail;
+
+    private int size;
+
+    private static class Node<E> {
+
+        E value;
+
+        Node<E> next;
+
+        Node<E> prev;
+
+        Node(E value) {
+            this.value = value;
+        }
+
+    }
+
+    /**
+     * Create a new instance of the {@link HashIndexedQueue}.
+     */
+    public HashIndexedQueue() {
+        map = new HashMap<>();
+        size = 0;
+    }
+
+    @Override
+    public boolean add(E e) {
+        return offer(e);
+    }
+
+    @Override
+    public boolean offer(E e) {
+        final Node<E> newNode = new Node<>(e);
+        if (tail == null) {
+            head = tail = newNode;
+        } else {
+            tail.next = newNode;
+            newNode.prev = tail;
+            tail = newNode;
+        }
+
+        if (!map.containsKey(e)) {
+            map.put(e, newNode);
+        } else {
+            Object current = map.get(e);
+            if (current instanceof Node) {
+                List<Node<E>> nodes = new ArrayList<>();
+                nodes.add((Node<E>) current);
+                nodes.add(newNode);
+                map.put(e, nodes);
+            } else {
+                ((List<Node<E>>) current).add(newNode);
+            }
+        }
+        size++;
+        return true;
+    }
+
+    @Override
+    public E poll() {
+        if (head == null) {
+            return null;
+        }
+        E value = head.value;
+        removeNodeFromMap(head);
+        head = head.next;
+        if (head == null) {
+            tail = null;
+        } else {
+            head.prev = null;
+        }
+        size--;
+        return value;
+    }
+
+    @Override
+    public E peek() {
+        if (head == null) {
+            return null;
+        }
+        return head.value;
+    }
+
+    @Override
+    public boolean remove(Object o) {
+        return removeFirstOccurrence(o);
+    }
+
+    @Override
+    public int size() {
+        return size;
+    }
+
+    @Override
+    public boolean contains(Object o) {
+        return map.containsKey(o);
+    }
+
+    public class Iterator implements java.util.Iterator<E> {
+
+        private Node<E> current;
+
+        private Node<E> prev;
+
+        private Iterator() {
+            current = HashIndexedQueue.this.head;
+            prev = null;
+        }
+
+        @Override
+        public boolean hasNext() {
+            return current != null;
+        }
+
+        @Override
+        public E next() {
+            if (!hasNext()) {
+                throw new NoSuchElementException();
+            }
+            E value = current.value;
+            prev = current;
+            current = current.next;
+            return value;
+        }
+
+        @Override
+        public void remove() {
+            if (prev != null) {
+                removeNodeFromMap(prev);
+                removeNode(prev);
+                size--;
+                // remove once
+                prev = null;
+            }
+        }
+
+    }
+
+    @NotNull
+    @Override
+    public Iterator iterator() {
+        return new Iterator();
+    }
+
+    @Override
+    public boolean removeAll(Collection<?> c) {
+        boolean modified = false;
+        for (Object e : c) {
+            if (removeAllOccurrences(e)) {
+                modified = true;
+            }
+        }
+        return modified;
+    }
+
+    @Override
+    public void clear() {
+        head = null;
+        tail = null;
+        map.clear();
+        size = 0;
+    }
+
+    private boolean removeFirstOccurrence(Object element) {
+        Object current = map.get(element);
+        if (current == null) {
+            return false;
+        }
+        if (current instanceof Node) {
+            Node<E> node = (Node<E>) current;
+            removeNode(node);
+            map.remove(element);
+        } else {
+            List<Node<E>> nodes = (List<Node<E>>) current;
+            Node<E> node = nodes.remove(0);
+            if (nodes.isEmpty()) {
+                map.remove(element);
+            }
+            removeNode(node);
+        }
+        size--;
+        return true;
+    }
+
+    private boolean removeAllOccurrences(Object element) {
+        Object current = map.get(element);
+        if (current == null) {
+            return false;
+        }
+        if (current instanceof Node) {
+            final Node<E> node = (Node<E>) current;
+            removeNode(node);
+            size--;
+        } else {
+            final List<Node<E>> nodes = (List<Node<E>>) current;
+            for (Node<E> node : nodes) {
+                removeNode(node);
+                size--;
+            }
+        }
+        map.remove(element);
+        return true;
+    }
+
+    private void removeNode(Node<E> node) {
+        if (node.prev != null) {
+            node.prev.next = node.next;
+        } else {
+            head = node.next;
+        }
+        if (node.next != null) {
+            node.next.prev = node.prev;
+        } else {
+            tail = node.prev;
+        }
+    }
+
+    private void removeNodeFromMap(Node<E> node) {
+        E value = node.value;
+        Object current = map.get(value);
+        if (current instanceof Node) {
+            LettuceAssert.assertState(current == node, "current != node");
+            map.remove(value);
+        } else {
+            List<Node<E>> nodes = (List<Node<E>>) current;
+            final boolean removed = nodes.remove(node);
+            LettuceAssert.assertState(removed, "!nodes.remove(node)");
+            if (nodes.isEmpty()) {
+                map.remove(value);
+            }
+        }
+    }
+
+}
diff --git a/src/main/java/io/lettuce/core/protocol/CommandHandler.java b/src/main/java/io/lettuce/core/protocol/CommandHandler.java
index 063a0ab21..6791be04c 100644
--- a/src/main/java/io/lettuce/core/protocol/CommandHandler.java
+++ b/src/main/java/io/lettuce/core/protocol/CommandHandler.java
@@ -43,6 +43,7 @@ import io.lettuce.core.RedisException;
 import io.lettuce.core.RedisURI;
 import io.lettuce.core.api.push.PushListener;
 import io.lettuce.core.api.push.PushMessage;
+import io.lettuce.core.datastructure.queue.HashIndexedQueue;
 import io.lettuce.core.internal.LettuceAssert;
 import io.lettuce.core.internal.LettuceSets;
 import io.lettuce.core.metrics.CommandLatencyRecorder;
@@ -98,7 +99,7 @@ public class CommandHandler extends ChannelDuplexHandler implements HasQueuedCom
 
     private final Endpoint endpoint;
 
-    private final ArrayDeque<RedisCommand<?, ?, ?>> stack = new ArrayDeque<>();
+    private final Queue<RedisCommand<?, ?, ?>> stack;
 
     private final long commandHandlerId = COMMAND_HANDLER_COUNTER.incrementAndGet();
 
@@ -157,6 +158,7 @@ public class CommandHandler extends ChannelDuplexHandler implements HasQueuedCom
         this.commandLatencyRecorder = clientResources.commandLatencyRecorder();
         this.latencyMetricsEnabled = commandLatencyRecorder.isEnabled();
         this.boundedQueues = clientOptions.getRequestQueueSize() != Integer.MAX_VALUE;
+        this.stack = clientOptions.isUseHashIndexedQueue() ? new HashIndexedQueue<>() : new ArrayDeque<>();
 
         Tracing tracing = clientResources.tracing();
 
@@ -1041,7 +1043,7 @@ public class CommandHandler extends ChannelDuplexHandler implements HasQueuedCom
 
         private final Recycler.Handle<AddToStack> handle;
 
-        private ArrayDeque<Object> stack;
+        private Queue<RedisCommand<?, ?, ?>> stack;
 
         private RedisCommand<?, ?, ?> command;
 
@@ -1057,11 +1059,11 @@ public class CommandHandler extends ChannelDuplexHandler implements HasQueuedCom
          * @return
          */
         @SuppressWarnings("unchecked")
-        static AddToStack newInstance(ArrayDeque<?> stack, RedisCommand<?, ?, ?> command) {
+        static AddToStack newInstance(Queue<RedisCommand<?, ?, ?>> stack, RedisCommand<?, ?, ?> command) {
 
             AddToStack entry = RECYCLER.get();
 
-            entry.stack = (ArrayDeque<Object>) stack;
+            entry.stack = stack;
             entry.command = command;
 
             return entry;


Test patch:
diff --git a/src/test/java/io/lettuce/core/datastructure/queue/HashIndexedQueueTests.java b/src/test/java/io/lettuce/core/datastructure/queue/HashIndexedQueueTests.java
new file mode 100644
index 000000000..8d837c048
--- /dev/null
+++ b/src/test/java/io/lettuce/core/datastructure/queue/HashIndexedQueueTests.java
@@ -0,0 +1,185 @@
+package io.lettuce.core.datastructure.queue;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.NoSuchElementException;
+
+import io.lettuce.test.LettuceExtension;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@SuppressWarnings("ConstantValue")
+@ExtendWith(LettuceExtension.class)
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class HashIndexedQueueTests {
+
+    private HashIndexedQueue<Integer> queue;
+
+    @BeforeEach
+    void setUp() {
+        queue = new HashIndexedQueue<>();
+    }
+
+    @Test
+    void testInitialization() {
+        assertTrue(queue.isEmpty());
+        assertEquals(0, queue.size());
+    }
+
+    @Test
+    void testAddAndOffer() {
+        assertTrue(queue.add(1));
+        assertTrue(queue.offer(2));
+        assertEquals(2, queue.size());
+        assertTrue(queue.contains(1));
+        assertTrue(queue.contains(2));
+    }
+
+    @Test
+    void testRemoveAndPoll() {
+        queue.add(1);
+        queue.add(2);
+        assertEquals(1, queue.remove());
+        assertEquals(1, queue.size());
+        assertEquals(2, queue.poll());
+        assertTrue(queue.isEmpty());
+    }
+
+    @Test
+    void testPeekAndElement() {
+        queue.add(1);
+        queue.add(2);
+        assertEquals(1, queue.peek());
+        assertEquals(1, queue.element());
+        queue.remove();
+        assertEquals(2, queue.peek());
+        assertEquals(2, queue.element());
+    }
+
+    @Test
+    void testContains() {
+        queue.add(1);
+        assertTrue(queue.contains(1));
+        assertFalse(queue.contains(2));
+    }
+
+    @Test
+    void testSizeAndIsEmpty() {
+        assertTrue(queue.isEmpty());
+        queue.add(1);
+        assertFalse(queue.isEmpty());
+        assertEquals(1, queue.size());
+    }
+
+    @Test
+    void testClear() {
+        queue.add(1);
+        queue.add(2);
+        queue.clear();
+        assertTrue(queue.isEmpty());
+        assertEquals(0, queue.size());
+    }
+
+    @Test
+    void testRemove() {
+        queue.add(1);
+        queue.add(1);
+        queue.add(2);
+        queue.add(1);
+        assertTrue(queue.remove(1));
+        assertEquals(3, queue.size());
+        assertTrue(queue.contains(1));
+        assertTrue(queue.contains(2));
+        assertTrue(queue.remove(1));
+        assertEquals(2, queue.size());
+        assertTrue(queue.remove(1));
+        assertEquals(1, queue.size());
+        assertTrue(queue.contains(2));
+    }
+
+    @Test
+    void testRemoveAll() {
+        queue.add(1);
+        queue.add(1);
+        queue.add(2);
+        queue.add(1);
+        assertTrue(queue.removeAll(Collections.singletonList(1)));
+        assertEquals(1, queue.size());
+        assertTrue(queue.contains(2));
+    }
+
+    @Test
+    void testRemoveAll2() {
+        queue.add(1);
+        queue.add(1);
+        queue.add(1);
+        queue.add(1);
+        assertTrue(queue.removeAll(Collections.singletonList(1)));
+        assertEquals(0, queue.size());
+    }
+
+    @Test
+    void testRemoveAllAndRetainAll() {
+        queue.addAll(Arrays.asList(1, 2, 3, 4));
+        queue.removeAll(Arrays.asList(1, 2));
+        assertEquals(2, queue.size());
+        assertFalse(queue.contains(1));
+        assertFalse(queue.contains(2));
+        queue.retainAll(Collections.singletonList(3));
+        assertEquals(1, queue.size());
+        assertTrue(queue.contains(3));
+        assertFalse(queue.contains(4));
+    }
+
+    @Test
+    void testRetainAll() {
+        queue.addAll(Arrays.asList(1, 2, 3, 4, 1, 2, 3, 4));
+        queue.retainAll(Arrays.asList(1, 2));
+        assertEquals(4, queue.size());
+        HashIndexedQueue<Integer>.Iterator iterator = queue.iterator();
+        iterator.remove();
+        assertEquals(1, iterator.next());
+        assertEquals(2, iterator.next());
+        assertEquals(1, iterator.next());
+        assertEquals(2, iterator.next());
+        iterator.remove();
+        // no effect
+        iterator.remove();
+        iterator.remove();
+
+        iterator = queue.iterator();
+        assertEquals(1, iterator.next());
+        assertEquals(2, iterator.next());
+        assertEquals(1, iterator.next());
+        assertFalse(iterator.hasNext());
+    }
+
+    @Test
+    void testRemoveNonExistentElement() {
+        assertThrows(NoSuchElementException.class, () -> queue.remove());
+    }
+
+    @Test
+    void testPollEmptyQueue() {
+        assertNull(queue.poll());
+    }
+
+    @Test
+    void testElementEmptyQueue() {
+        assertThrows(NoSuchElementException.class, () -> queue.element());
+    }
+
+    @Test
+    void testPeekEmptyQueue() {
+        assertNull(queue.peek());
+    }
+
+}
diff --git a/src/test/java/io/redis/examples/async/HashExample.java b/src/test/java/io/redis/examples/async/HashExample.java
new file mode 100644
index 000000000..eba2f4e0c
--- /dev/null
+++ b/src/test/java/io/redis/examples/async/HashExample.java
@@ -0,0 +1,166 @@
+// EXAMPLE: hash_tutorial
+package io.redis.examples.async;
+
+import io.lettuce.core.*;
+import io.lettuce.core.api.async.RedisAsyncCommands;
+import io.lettuce.core.api.StatefulRedisConnection;
+
+// REMOVE_START
+import org.junit.jupiter.api.Test;
+// REMOVE_END
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+// REMOVE_START
+import static org.assertj.core.api.Assertions.assertThat;
+// REMOVE_END
+
+public class HashExample {
+
+    @Test
+    public void run() {
+        RedisClient redisClient = RedisClient.create("redis://localhost:6379");
+
+        try (StatefulRedisConnection<String, String> connection = redisClient.connect()) {
+            RedisAsyncCommands<String, String> asyncCommands = connection.async();
+            // REMOVE_START
+            CompletableFuture<Long> delResult = asyncCommands.del("bike:1", "bike:1:stats").toCompletableFuture();
+
+            // REMOVE_END
+
+            // STEP_START set_get_all
+            Map<String, String> bike1 = new HashMap<>();
+            bike1.put("model", "Deimos");
+            bike1.put("brand", "Ergonom");
+            bike1.put("type", "Enduro bikes");
+            bike1.put("price", "4972");
+
+            CompletableFuture<Void> setGetAll = asyncCommands.hset("bike:1", bike1).thenCompose(res1 -> {
+                System.out.println(res1); // >>> 4
+                // REMOVE_START
+                assertThat(res1).isEqualTo(4);
+                // REMOVE_END
+                return asyncCommands.hget("bike:1", "model");
+            }).thenCompose(res2 -> {
+                System.out.println(res2); // >>> Deimos
+                // REMOVE_START
+                assertThat(res2).isEqualTo("Deimos");
+                // REMOVE_END
+                return asyncCommands.hget("bike:1", "price");
+            }).thenCompose(res3 -> {
+                System.out.println(res3); // >>> 4972
+                // REMOVE_START
+                assertThat(res3).isEqualTo("4972");
+                // REMOVE_END
+                return asyncCommands.hgetall("bike:1");
+            })
+                    // REMOVE_START
+                    .thenApply(res -> {
+                        assertThat(res.get("type")).isEqualTo("Enduro bikes");
+                        assertThat(res.get("brand")).isEqualTo("Ergonom");
+                        assertThat(res.get("price")).isEqualTo("4972");
+                        assertThat(res.get("model")).isEqualTo("Deimos");
+
+                        return res;
+                    })
+                    // REMOVE_END
+                    .thenAccept(System.out::println)
+                    // >>> {type=Enduro bikes, brand=Ergonom, price=4972, model=Deimos}
+                    .toCompletableFuture();
+            // STEP_END
+
+            // STEP_START hmget
+            CompletableFuture<Void> hmGet = setGetAll.thenCompose(res4 -> {
+                return asyncCommands.hmget("bike:1", "model", "price");
+            })
+                    // REMOVE_START
+                    .thenApply(res -> {
+                        assertThat(res.toString()).isEqualTo("[KeyValue[model, Deimos], KeyValue[price, 4972]]");
+                        return res;
+                    })
+                    // REMOVE_END
+                    .thenAccept(System.out::println)
+                    // [KeyValue[model, Deimos], KeyValue[price, 4972]]
+                    .toCompletableFuture();
+            // STEP_END
+
+            // STEP_START hincrby
+            CompletableFuture<Void> hIncrBy = hmGet.thenCompose(r -> {
+                return asyncCommands.hincrby("bike:1", "price", 100);
+            }).thenCompose(res6 -> {
+                System.out.println(res6); // >>> 5072
+                // REMOVE_START
+                assertThat(res6).isEqualTo(5072L);
+                // REMOVE_END
+                return asyncCommands.hincrby("bike:1", "price", -100);
+            })
+                    // REMOVE_START
+                    .thenApply(res -> {
+                        assertThat(res).isEqualTo(4972L);
+                        return res;
+                    })
+                    // REMOVE_END
+                    .thenAccept(System.out::println)
+                    // >>> 4972
+                    .toCompletableFuture();
+            // STEP_END
+
+            // STEP_START incrby_get_mget
+            CompletableFuture<Void> incrByGetMget = asyncCommands.hincrby("bike:1:stats", "rides", 1).thenCompose(res7 -> {
+                System.out.println(res7); // >>> 1
+                // REMOVE_START
+                assertThat(res7).isEqualTo(1L);
+                // REMOVE_END
+                return asyncCommands.hincrby("bike:1:stats", "rides", 1);
+            }).thenCompose(res8 -> {
+                System.out.println(res8); // >>> 2
+                // REMOVE_START
+                assertThat(res8).isEqualTo(2L);
+                // REMOVE_END
+                return asyncCommands.hincrby("bike:1:stats", "rides", 1);
+            }).thenCompose(res9 -> {
+                System.out.println(res9); // >>> 3
+                // REMOVE_START
+                assertThat(res9).isEqualTo(3L);
+                // REMOVE_END
+                return asyncCommands.hincrby("bike:1:stats", "crashes", 1);
+            }).thenCompose(res10 -> {
+                System.out.println(res10); // >>> 1
+                // REMOVE_START
+                assertThat(res10).isEqualTo(1L);
+                // REMOVE_END
+                return asyncCommands.hincrby("bike:1:stats", "owners", 1);
+            }).thenCompose(res11 -> {
+                System.out.println(res11); // >>> 1
+                // REMOVE_START
+                assertThat(res11).isEqualTo(1L);
+                // REMOVE_END
+                return asyncCommands.hget("bike:1:stats", "rides");
+            }).thenCompose(res12 -> {
+                System.out.println(res12); // >>> 3
+                // REMOVE_START
+                assertThat(res12).isEqualTo("3");
+                // REMOVE_END
+                return asyncCommands.hmget("bike:1:stats", "crashes", "owners");
+            })
+                    // REMOVE_START
+                    .thenApply(res -> {
+                        assertThat(res.toString()).isEqualTo("[KeyValue[crashes, 1], KeyValue[owners, 1]]");
+                        return res;
+                    })
+                    // REMOVE_END
+                    .thenAccept(System.out::println)
+                    // >>> [KeyValue[crashes, 1], KeyValue[owners, 1]]
+                    .toCompletableFuture();
+            // STEP_END
+
+            CompletableFuture.allOf(
+                    // REMOVE_START
+                    delResult,
+                    // REMOVE_END,
+                    hIncrBy, incrByGetMget).join();
+        } finally {
+            redisClient.shutdown();
+        }
+    }
+
+}