Repository: redis/lettuce (java)

Issue:
Add support for CLUSTER LINKS
Add support for the command as documented [here](https://redis.io/commands/cluster-links)

Pull request:
Add support clusterlinks command
<!--
Thank you for proposing a pull request. This template will guide you through the essential steps necessary for a pull request.
-->
Closes #2755 
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.

 <!--
Great! Live long and prosper.
-->


Patch:
diff --git a/src/main/java/io/lettuce/core/AbstractRedisAsyncCommands.java b/src/main/java/io/lettuce/core/AbstractRedisAsyncCommands.java
index d72accd63..60a009664 100644
--- a/src/main/java/io/lettuce/core/AbstractRedisAsyncCommands.java
+++ b/src/main/java/io/lettuce/core/AbstractRedisAsyncCommands.java
@@ -3189,6 +3189,11 @@ public abstract class AbstractRedisAsyncCommands<K, V> implements RedisAclAsyncC
         return dispatch(commandBuilder.zunionstore(destination, zStoreArgs, keys));
     }
 
+    @Override
+    public RedisFuture<List<Map<String, Object>>> clusterLinks() {
+        return dispatch(commandBuilder.clusterLinks());
+    }
+
     private byte[] encodeFunction(String functionCode) {
         LettuceAssert.notNull(functionCode, "Function code must not be null");
         LettuceAssert.notEmpty(functionCode, "Function code script must not be empty");
diff --git a/src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java b/src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java
index 675bc31ba..8614a6eb5 100644
--- a/src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java
+++ b/src/main/java/io/lettuce/core/AbstractRedisReactiveCommands.java
@@ -3257,6 +3257,11 @@ public abstract class AbstractRedisReactiveCommands<K, V>
         return createMono(() -> commandBuilder.zunionstore(destination, zStoreArgs, keys));
     }
 
+    @Override
+    public Mono<List<Map<String, Object>>> clusterLinks() {
+        return createMono(commandBuilder::clusterLinks);
+    }
+
     private byte[] encodeFunction(String functionCode) {
         LettuceAssert.notNull(functionCode, "Function code must not be null");
         LettuceAssert.notEmpty(functionCode, "Function code script must not be empty");
diff --git a/src/main/java/io/lettuce/core/ClientOptions.java b/src/main/java/io/lettuce/core/ClientOptions.java
index aa3d2ba18..3e32967bf 100644
--- a/src/main/java/io/lettuce/core/ClientOptions.java
+++ b/src/main/java/io/lettuce/core/ClientOptions.java
@@ -67,7 +67,7 @@ public class ClientOptions implements Serializable {
 
     public static final boolean DEFAULT_SUSPEND_RECONNECT_PROTO_FAIL = false;
 
-    public static final TimeoutOptions DEFAULT_TIMEOUT_OPTIONS = TimeoutOptions.create();
+    public static final TimeoutOptions DEFAULT_TIMEOUT_OPTIONS = TimeoutOptions.enabled();
 
     private final boolean autoReconnect;
 
diff --git a/src/main/java/io/lettuce/core/RedisCommandBuilder.java b/src/main/java/io/lettuce/core/RedisCommandBuilder.java
index 46e5cb809..a192690d9 100644
--- a/src/main/java/io/lettuce/core/RedisCommandBuilder.java
+++ b/src/main/java/io/lettuce/core/RedisCommandBuilder.java
@@ -4433,6 +4433,11 @@ class RedisCommandBuilder<K, V> extends BaseRedisCommandBuilder<K, V> {
         return createCommand(ZUNIONSTORE, new IntegerOutput<>(codec), args);
     }
 
+    Command<K, V, List<Map<String, Object>>> clusterLinks() {
+        CommandArgs<K, V> args = new CommandArgs<>(codec).add(LINKS);
+        return createCommand(CLUSTER, (CommandOutput) new ObjectOutput<>(StringCodec.UTF8), args);
+    }
+
     private boolean allElementsInstanceOf(Object[] objects, Class<?> expectedAssignableType) {
 
         for (Object object : objects) {
diff --git a/src/main/java/io/lettuce/core/cluster/api/async/RedisClusterAsyncCommands.java b/src/main/java/io/lettuce/core/cluster/api/async/RedisClusterAsyncCommands.java
index 2835968fd..6f9e1e902 100644
--- a/src/main/java/io/lettuce/core/cluster/api/async/RedisClusterAsyncCommands.java
+++ b/src/main/java/io/lettuce/core/cluster/api/async/RedisClusterAsyncCommands.java
@@ -380,4 +380,11 @@ public interface RedisClusterAsyncCommands<K, V> extends BaseRedisAsyncCommands<
      */
     RedisFuture<String> readWrite();
 
+    /**
+     * Retrieves information about the TCP links between nodes in a Redis Cluster.
+     * 
+     * @return List of maps containing attributes and values for each peer link.
+     */
+    RedisFuture<List<Map<String, Object>>> clusterLinks();
+
 }
diff --git a/src/main/java/io/lettuce/core/cluster/api/reactive/RedisClusterReactiveCommands.java b/src/main/java/io/lettuce/core/cluster/api/reactive/RedisClusterReactiveCommands.java
index 5f1d611f9..2416b380e 100644
--- a/src/main/java/io/lettuce/core/cluster/api/reactive/RedisClusterReactiveCommands.java
+++ b/src/main/java/io/lettuce/core/cluster/api/reactive/RedisClusterReactiveCommands.java
@@ -21,6 +21,7 @@ package io.lettuce.core.cluster.api.reactive;
 
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
 
 import io.lettuce.core.Range;
 import io.lettuce.core.api.reactive.*;
@@ -368,4 +369,11 @@ public interface RedisClusterReactiveCommands<K, V> extends BaseRedisReactiveCom
      */
     Mono<String> readWrite();
 
+    /**
+     * Retrieves information about the TCP links between nodes in a Redis Cluster.
+     * 
+     * @return List of maps containing attributes and values for each peer link.
+     */
+    Mono<List<Map<String, Object>>> clusterLinks();
+
 }
diff --git a/src/main/java/io/lettuce/core/cluster/api/sync/RedisClusterCommands.java b/src/main/java/io/lettuce/core/cluster/api/sync/RedisClusterCommands.java
index 2e18bc68d..ec02e83a9 100644
--- a/src/main/java/io/lettuce/core/cluster/api/sync/RedisClusterCommands.java
+++ b/src/main/java/io/lettuce/core/cluster/api/sync/RedisClusterCommands.java
@@ -21,6 +21,7 @@ package io.lettuce.core.cluster.api.sync;
 
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
 
 import io.lettuce.core.Range;
 import io.lettuce.core.api.sync.*;
@@ -368,4 +369,11 @@ public interface RedisClusterCommands<K, V> extends BaseRedisCommands<K, V>, Red
     @Override
     String readWrite();
 
+    /**
+     * Retrieves information about the TCP links between nodes in a Redis Cluster.
+     * 
+     * @return List of maps containing attributes and values for each peer link.
+     */
+    List<Map<String, Object>> clusterLinks();
+
 }
diff --git a/src/main/java/io/lettuce/core/protocol/CommandKeyword.java b/src/main/java/io/lettuce/core/protocol/CommandKeyword.java
index a422ad1f1..bd56b713d 100644
--- a/src/main/java/io/lettuce/core/protocol/CommandKeyword.java
+++ b/src/main/java/io/lettuce/core/protocol/CommandKeyword.java
@@ -39,7 +39,7 @@ public enum CommandKeyword implements ProtocolKeyword {
 
     FAILOVER, FORGET, FIELDS, FLAGS, FLUSH, FORCE, FREQ, FLUSHSLOTS, GENPASS, GETNAME, GETUSER, GETKEYSINSLOT, GETREDIR, GROUP, GROUPS, HTSTATS, ID, IDLE, INFO,
 
-    IDLETIME, JUSTID, KILL, KEYSLOT, LEFT, LEN, LIMIT, LIST, LOAD, LOG, MATCH,
+    IDLETIME, JUSTID, KILL, KEYSLOT, LEFT, LEN, LIMIT, LINKS, LIST, LOAD, LOG, MATCH,
 
     MAX, MAXLEN, MEET, MIN, MINID, MOVED, NO, NOACK, NOCOMMANDS, NODE, NODES, NOMKSTREAM, NOPASS, NOSAVE, NOT, NOVALUES, NUMSUB, SHARDCHANNELS, SHARDNUMSUB, NUMPAT, NX, OFF, ON, ONE, OR, PAUSE, PREFIXES,
 
diff --git a/src/main/java/io/lettuce/core/protocol/SharedLock.java b/src/main/java/io/lettuce/core/protocol/SharedLock.java
index 13a9cb8cf..c3ad425c1 100644
--- a/src/main/java/io/lettuce/core/protocol/SharedLock.java
+++ b/src/main/java/io/lettuce/core/protocol/SharedLock.java
@@ -26,6 +26,8 @@ class SharedLock {
 
     private final Lock lock = new ReentrantLock();
 
+    private final ThreadLocal<Integer> threadWriters = ThreadLocal.withInitial(() -> 0);
+
     private volatile long writers = 0;
 
     private volatile Thread exclusiveLockOwner;
@@ -45,6 +47,7 @@ class SharedLock {
 
                 if (WRITERS.get(this) >= 0) {
                     WRITERS.incrementAndGet(this);
+                    threadWriters.set(threadWriters.get() + 1);
                     return;
                 }
             }
@@ -63,6 +66,7 @@ class SharedLock {
         }
 
         WRITERS.decrementAndGet(this);
+        threadWriters.set(threadWriters.get() - 1);
     }
 
     /**
@@ -121,7 +125,8 @@ class SharedLock {
         try {
             for (;;) {
 
-                if (WRITERS.compareAndSet(this, 0, -1)) {
+                // allow reentrant exclusive lock by comparing writers count and threadWriters count
+                if (WRITERS.compareAndSet(this, threadWriters.get(), -1)) {
                     exclusiveLockOwner = Thread.currentThread();
                     return;
                 }
@@ -137,9 +142,13 @@ class SharedLock {
     private void unlockWritersExclusive() {
 
         if (exclusiveLockOwner == Thread.currentThread()) {
-            if (WRITERS.incrementAndGet(this) == 0) {
+            // check exclusive look not reentrant first
+            if (WRITERS.compareAndSet(this, -1, threadWriters.get())) {
                 exclusiveLockOwner = null;
+                return;
             }
+            // otherwise unlock until no more reentrant left
+            WRITERS.incrementAndGet(this);
         }
     }
 


Test patch:
diff --git a/src/test/java/io/lettuce/core/ClientOptionsIntegrationTests.java b/src/test/java/io/lettuce/core/ClientOptionsIntegrationTests.java
index 347ec18e9..8ca4c896e 100644
--- a/src/test/java/io/lettuce/core/ClientOptionsIntegrationTests.java
+++ b/src/test/java/io/lettuce/core/ClientOptionsIntegrationTests.java
@@ -92,7 +92,8 @@ class ClientOptionsIntegrationTests extends TestSupport {
     @Test
     void requestQueueSize() {
 
-        client.setOptions(ClientOptions.builder().requestQueueSize(10).build());
+        client.setOptions(ClientOptions.builder().requestQueueSize(10)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         StatefulRedisConnection<String, String> connection = client.connect();
         getConnectionWatchdog(connection).setListenOnChannelInactive(false);
@@ -115,7 +116,8 @@ class ClientOptionsIntegrationTests extends TestSupport {
     @Test
     void requestQueueSizeAppliedForReconnect() {
 
-        client.setOptions(ClientOptions.builder().requestQueueSize(10).build());
+        client.setOptions(ClientOptions.builder().requestQueueSize(10)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         RedisAsyncCommands<String, String> connection = client.connect().async();
         testHitRequestQueueLimit(connection);
@@ -127,7 +129,7 @@ class ClientOptionsIntegrationTests extends TestSupport {
         WithPassword.run(client, () -> {
 
             client.setOptions(ClientOptions.builder().protocolVersion(ProtocolVersion.RESP2).pingBeforeActivateConnection(false)
-                    .requestQueueSize(10).build());
+                    .requestQueueSize(10).timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
             RedisAsyncCommands<String, String> connection = client.connect().async();
             connection.auth(passwd);
@@ -142,7 +144,7 @@ class ClientOptionsIntegrationTests extends TestSupport {
         WithPassword.run(client, () -> {
 
             client.setOptions(ClientOptions.builder().protocolVersion(ProtocolVersion.RESP2).pingBeforeActivateConnection(false)
-                    .requestQueueSize(10).build());
+                    .requestQueueSize(10).timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
             RedisAsyncCommands<String, String> connection = client.connect().async();
             connection.auth(username, passwd);
@@ -154,7 +156,9 @@ class ClientOptionsIntegrationTests extends TestSupport {
     void testHitRequestQueueLimitReconnectWithUriAuth() {
 
         WithPassword.run(client, () -> {
-            client.setOptions(ClientOptions.builder().requestQueueSize(10).build());
+            client.setOptions(ClientOptions.builder().requestQueueSize(10)
+                    .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
+            ;
 
             RedisURI redisURI = RedisURI.create(host, port);
             redisURI.setPassword(passwd);
@@ -169,7 +173,8 @@ class ClientOptionsIntegrationTests extends TestSupport {
 
         WithPassword.run(client, () -> {
 
-            client.setOptions(ClientOptions.builder().requestQueueSize(10).build());
+            client.setOptions(ClientOptions.builder().requestQueueSize(10)
+                    .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
             RedisURI redisURI = RedisURI.create(host, port);
             redisURI.setPassword(passwd);
@@ -207,7 +212,8 @@ class ClientOptionsIntegrationTests extends TestSupport {
     @Test
     void requestQueueSizeOvercommittedReconnect() {
 
-        client.setOptions(ClientOptions.builder().requestQueueSize(10).build());
+        client.setOptions(ClientOptions.builder().requestQueueSize(10)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         StatefulRedisConnection<String, String> connection = client.connect();
         ConnectionWatchdog watchdog = getConnectionWatchdog(connection);
@@ -262,8 +268,8 @@ class ClientOptionsIntegrationTests extends TestSupport {
     @Test
     void disconnectedRejectCommands() {
 
-        client.setOptions(
-                ClientOptions.builder().disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS).build());
+        client.setOptions(ClientOptions.builder().disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         RedisAsyncCommands<String, String> connection = client.connect().async();
 
@@ -471,7 +477,8 @@ class ClientOptionsIntegrationTests extends TestSupport {
 
     @Test
     void pingBeforeConnectWithQueuedCommandsAndReconnect() throws Exception {
-
+        client.setOptions(
+                ClientOptions.builder().timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
         StatefulRedisConnection<String, String> controlConnection = client.connect();
 
         StatefulRedisConnection<String, String> redisConnection = client.connect(RedisURI.create("redis://localhost:6479/5"));
@@ -513,6 +520,8 @@ class ClientOptionsIntegrationTests extends TestSupport {
         WithPassword.run(client, () -> {
 
             RedisURI redisURI = RedisURI.Builder.redis(host, port).withPassword(passwd).withDatabase(5).build();
+            client.setOptions(
+                    ClientOptions.builder().timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
             StatefulRedisConnection<String, String> controlConnection = client.connect(redisURI);
 
             StatefulRedisConnection<String, String> redisConnection = client.connect(redisURI);
diff --git a/src/test/java/io/lettuce/core/RedisCommandBuilderUnitTests.java b/src/test/java/io/lettuce/core/RedisCommandBuilderUnitTests.java
index f99393cc1..16276b170 100644
--- a/src/test/java/io/lettuce/core/RedisCommandBuilderUnitTests.java
+++ b/src/test/java/io/lettuce/core/RedisCommandBuilderUnitTests.java
@@ -7,6 +7,8 @@ import io.netty.buffer.Unpooled;
 import org.junit.jupiter.api.Test;
 
 import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -168,4 +170,14 @@ class RedisCommandBuilderUnitTests {
                 .isEqualTo("*2\r\n" + "$7\r\n" + "CLUSTER\r\n" + "$9\r\n" + "MYSHARDID\r\n");
     }
 
+    @Test
+    void shouldCorrectlyConstructClusterLinks() {
+
+        Command<String, String, List<Map<String, Object>>> command = sut.clusterLinks();
+        ByteBuf buf = Unpooled.directBuffer();
+        command.encode(buf);
+
+        assertThat(buf.toString(StandardCharsets.UTF_8)).isEqualTo("*2\r\n$7\r\nCLUSTER\r\n$5\r\nLINKS\r\n");
+    }
+
 }
diff --git a/src/test/java/io/lettuce/core/cluster/ClusterCommandIntegrationTests.java b/src/test/java/io/lettuce/core/cluster/ClusterCommandIntegrationTests.java
index 2cdff1699..ae92d0f23 100644
--- a/src/test/java/io/lettuce/core/cluster/ClusterCommandIntegrationTests.java
+++ b/src/test/java/io/lettuce/core/cluster/ClusterCommandIntegrationTests.java
@@ -5,6 +5,7 @@ import static org.assertj.core.api.Assertions.*;
 
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
 
 import javax.inject.Inject;
 
@@ -271,6 +272,26 @@ class ClusterCommandIntegrationTests extends TestSupport {
         assertThat(result.size()).isGreaterThan(0);
     }
 
+    @Test
+    void testClusterLinks() {
+        List<Map<String, Object>> values = sync.clusterLinks();
+        assertThat(values).isNotEmpty();
+        for (Map<String, Object> value : values) {
+            assertThat(value).containsKeys("direction", "node", "create-time", "events", "send-buffer-allocated",
+                    "send-buffer-used");
+        }
+    }
+
+    @Test
+    void testClusterLinksAsync() throws Exception {
+        RedisFuture<List<Map<String, Object>>> futureLinks = async.clusterLinks();
+        List<Map<String, Object>> values = futureLinks.get();
+        for (Map<String, Object> value : values) {
+            assertThat(value).containsKeys("direction", "node", "create-time", "events", "send-buffer-allocated",
+                    "send-buffer-used");
+        }
+    }
+
     private void prepareReadonlyTest(String key) {
 
         async.set(key, value);
diff --git a/src/test/java/io/lettuce/core/cluster/ClusterReactiveCommandIntegrationTests.java b/src/test/java/io/lettuce/core/cluster/ClusterReactiveCommandIntegrationTests.java
index 74ddc18d0..ec9a31582 100644
--- a/src/test/java/io/lettuce/core/cluster/ClusterReactiveCommandIntegrationTests.java
+++ b/src/test/java/io/lettuce/core/cluster/ClusterReactiveCommandIntegrationTests.java
@@ -3,6 +3,7 @@ package io.lettuce.core.cluster;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.List;
+import java.util.Map;
 
 import javax.inject.Inject;
 
@@ -97,4 +98,13 @@ class ClusterReactiveCommandIntegrationTests {
         assertThat(result.size()).isGreaterThan(0);
     }
 
+    @Test
+    void testClusterLinks() {
+        List<Map<String, Object>> values = reactive.clusterLinks().block();
+        for (Map<String, Object> value : values) {
+            assertThat(value).containsKeys("direction", "node", "create-time", "events", "send-buffer-allocated",
+                    "send-buffer-used");
+        }
+    }
+
 }
diff --git a/src/test/java/io/lettuce/core/cluster/RedisClusterSetupTest.java b/src/test/java/io/lettuce/core/cluster/RedisClusterSetupTest.java
index ce956e0de..14e7a4bed 100644
--- a/src/test/java/io/lettuce/core/cluster/RedisClusterSetupTest.java
+++ b/src/test/java/io/lettuce/core/cluster/RedisClusterSetupTest.java
@@ -31,6 +31,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import io.lettuce.core.TimeoutOptions;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
@@ -303,10 +304,12 @@ public class RedisClusterSetupTest extends TestSupport {
     public void disconnectedConnectionRejectTest() throws Exception {
 
         clusterClient.setOptions(ClusterClientOptions.builder().topologyRefreshOptions(PERIODIC_REFRESH_ENABLED)
-                .disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS).build());
+                .disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
         RedisAdvancedClusterAsyncCommands<String, String> clusterConnection = clusterClient.connect().async();
-        clusterClient.setOptions(ClusterClientOptions.builder()
-                .disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS).build());
+        clusterClient.setOptions(
+                ClusterClientOptions.builder().disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
+                        .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
         ClusterSetup.setup2Masters(clusterHelper);
 
         assertRoutedExecution(clusterConnection);
@@ -330,9 +333,11 @@ public class RedisClusterSetupTest extends TestSupport {
     @Test
     public void atLeastOnceForgetNodeFailover() throws Exception {
 
-        clusterClient.setOptions(ClusterClientOptions.builder().topologyRefreshOptions(PERIODIC_REFRESH_ENABLED).build());
+        clusterClient.setOptions(ClusterClientOptions.builder().topologyRefreshOptions(PERIODIC_REFRESH_ENABLED)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
         RedisAdvancedClusterAsyncCommands<String, String> clusterConnection = clusterClient.connect().async();
-        clusterClient.setOptions(ClusterClientOptions.create());
+        clusterClient.setOptions(
+                ClusterClientOptions.builder().timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
         ClusterSetup.setup2Masters(clusterHelper);
 
         assertRoutedExecution(clusterConnection);
diff --git a/src/test/java/io/lettuce/core/protocol/ConnectionFailureIntegrationTests.java b/src/test/java/io/lettuce/core/protocol/ConnectionFailureIntegrationTests.java
index 38f759c37..d2f720d3f 100644
--- a/src/test/java/io/lettuce/core/protocol/ConnectionFailureIntegrationTests.java
+++ b/src/test/java/io/lettuce/core/protocol/ConnectionFailureIntegrationTests.java
@@ -89,7 +89,8 @@ class ConnectionFailureIntegrationTests extends TestSupport {
     @Test
     void failOnReconnect() throws Exception {
 
-        ClientOptions clientOptions = ClientOptions.builder().suspendReconnectOnProtocolFailure(true).build();
+        ClientOptions clientOptions = ClientOptions.builder().suspendReconnectOnProtocolFailure(true)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build();
         client.setOptions(clientOptions);
 
         RandomResponseServer ts = getRandomResponseServer();
@@ -133,7 +134,8 @@ class ConnectionFailureIntegrationTests extends TestSupport {
     @Test
     void failOnReconnectShouldSendEvents() throws Exception {
 
-        client.setOptions(ClientOptions.builder().suspendReconnectOnProtocolFailure(false).build());
+        client.setOptions(ClientOptions.builder().suspendReconnectOnProtocolFailure(false)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         RandomResponseServer ts = getRandomResponseServer();
 
@@ -181,7 +183,8 @@ class ConnectionFailureIntegrationTests extends TestSupport {
     @Test
     void cancelCommandsOnReconnectFailure() throws Exception {
 
-        client.setOptions(ClientOptions.builder().cancelCommandsOnReconnectFailure(true).build());
+        client.setOptions(ClientOptions.builder().cancelCommandsOnReconnectFailure(true)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         RandomResponseServer ts = getRandomResponseServer();
 
@@ -235,7 +238,8 @@ class ConnectionFailureIntegrationTests extends TestSupport {
         RedisURI redisUri = RedisURI.create(defaultRedisUri.toURI());
         RedisClient client = RedisClient.create(clientResources);
 
-        client.setOptions(ClientOptions.builder().build());
+        client.setOptions(
+                ClientOptions.builder().timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         try {
             RedisAsyncCommandsImpl<String, String> connection = (RedisAsyncCommandsImpl<String, String>) client
@@ -329,7 +333,8 @@ class ConnectionFailureIntegrationTests extends TestSupport {
 
         RedisURI redisUri = RedisURI.create(TestSettings.host(), TestSettings.port());
         RedisClient client = RedisClient.create(clientResources, redisUri);
-        client.setOptions(ClientOptions.builder().pingBeforeActivateConnection(true).build());
+        client.setOptions(ClientOptions.builder().pingBeforeActivateConnection(true)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         StatefulRedisConnection<String, String> connection = client.connect();
 
@@ -365,6 +370,8 @@ class ConnectionFailureIntegrationTests extends TestSupport {
     void closingDisconnectedConnectionShouldDisableConnectionWatchdog() {
 
         client.setOptions(ClientOptions.create());
+        client.setOptions(
+                ClientOptions.builder().timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         RedisURI redisUri = RedisURI.Builder.redis(TestSettings.host(), TestSettings.port()).withTimeout(Duration.ofMinutes(10))
                 .build();
diff --git a/src/test/java/io/lettuce/core/protocol/SharedLockTest.java b/src/test/java/io/lettuce/core/protocol/SharedLockTest.java
new file mode 100644
index 000000000..a10e712ae
--- /dev/null
+++ b/src/test/java/io/lettuce/core/protocol/SharedLockTest.java
@@ -0,0 +1,57 @@
+package io.lettuce.core.protocol;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class SharedLockTest {
+
+    @Test
+    public void safety_on_reentrant_lock_exclusive_on_writers() throws InterruptedException {
+        final SharedLock sharedLock = new SharedLock();
+        CountDownLatch cnt = new CountDownLatch(1);
+        try {
+            sharedLock.incrementWriters();
+
+            String result = sharedLock.doExclusive(() -> {
+                return sharedLock.doExclusive(() -> {
+                    return "ok";
+                });
+            });
+            if ("ok".equals(result)) {
+                cnt.countDown();
+            }
+        } finally {
+            sharedLock.decrementWriters();
+        }
+
+        boolean await = cnt.await(1, TimeUnit.SECONDS);
+        Assertions.assertTrue(await);
+
+        // verify writers won't be negative after finally decrementWriters
+        String result = sharedLock.doExclusive(() -> {
+            return sharedLock.doExclusive(() -> {
+                return "ok";
+            });
+        });
+
+        Assertions.assertEquals("ok", result);
+
+        // and other writers should be passed after exclusive lock released
+        CountDownLatch cntOtherThread = new CountDownLatch(1);
+        new Thread(() -> {
+            try {
+                sharedLock.incrementWriters();
+                cntOtherThread.countDown();
+            } finally {
+                sharedLock.decrementWriters();
+            }
+        }).start();
+
+        await = cntOtherThread.await(1, TimeUnit.SECONDS);
+        Assertions.assertTrue(await);
+    }
+
+}
diff --git a/src/test/java/io/lettuce/core/reliability/AtLeastOnceTest.java b/src/test/java/io/lettuce/core/reliability/AtLeastOnceTest.java
index b29007505..9243d301d 100644
--- a/src/test/java/io/lettuce/core/reliability/AtLeastOnceTest.java
+++ b/src/test/java/io/lettuce/core/reliability/AtLeastOnceTest.java
@@ -8,6 +8,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
+import io.lettuce.core.TimeoutOptions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -44,7 +45,8 @@ class AtLeastOnceTest extends AbstractRedisClientTest {
 
     @BeforeEach
     void before() {
-        client.setOptions(ClientOptions.builder().autoReconnect(true).build());
+        client.setOptions(ClientOptions.builder().autoReconnect(true)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         // needs to be increased on slow systems...perhaps...
         client.setDefaultTimeout(3, TimeUnit.SECONDS);
diff --git a/src/test/java/io/lettuce/core/reliability/AtMostOnceTest.java b/src/test/java/io/lettuce/core/reliability/AtMostOnceTest.java
index 7963d5a98..4bc3406c0 100644
--- a/src/test/java/io/lettuce/core/reliability/AtMostOnceTest.java
+++ b/src/test/java/io/lettuce/core/reliability/AtMostOnceTest.java
@@ -9,6 +9,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
+import io.lettuce.core.TimeoutOptions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -46,7 +47,8 @@ class AtMostOnceTest extends AbstractRedisClientTest {
 
     @BeforeEach
     void before() {
-        client.setOptions(ClientOptions.builder().autoReconnect(false).build());
+        client.setOptions(ClientOptions.builder().autoReconnect(false)
+                .timeoutOptions(TimeoutOptions.builder().timeoutCommands(false).build()).build());
 
         // needs to be increased on slow systems...perhaps...
         client.setDefaultTimeout(3, TimeUnit.SECONDS);