private <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectStandaloneAsync(
        RedisCodec<K, V> codec, RedisURI redisURI, Duration timeout) {
    
    // Call #1: Endpoint creation
    DefaultEndpoint endpoint = new DefaultEndpoint(getOptions(), getResources());
    RedisChannelWriter writer = endpoint;
    
    // Call #2: Check if CommandExpiryWriter is supported
    if (CommandExpiryWriter.isSupported(getOptions())) {
        // Call #3: Create CommandExpiryWriter
        writer = CommandExpiryWriter.buildCommandExpiryWriter(writer, getOptions(), getResources());
    }
    
    // ...
    
    // Call #4: Create CommandHandler (via lambda, executed later on Netty thread)
    ConnectionFuture<StatefulRedisConnection<K, V>> future = connectStatefulAsync(
        connection, endpoint, redisURI,
        () -> new CommandHandler(getOptions(), getResources(), endpoint), false);
    
    // ...
}

package io.lettuce.core;

import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.codec.StringCodec;
import io.lettuce.test.settings.TestSettings;
import org.junit.jupiter.api.Test;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

/**
 * Demonstrates race condition when calling setOptions() during async connection establishment.
 * <p>
 * The test shows that different components of the same connection can be initialized with different ClientOptions if
 * setOptions() is called while connectAsync() is in progress.
 */
class ClientOptionsRaceConditionTest extends TestSupport {

    @Test
    void demonstrateRaceCondition() throws Exception {
        RedisURI redisURI = RedisURI.create(TestSettings.host(), TestSettings.port());
        // Create a test client that captures getOptions() calls
        TestRedisClient client = new TestRedisClient(redisURI);

        try {
            // Initial options with 10 second timeout
            ClientOptions initialOptions = ClientOptions.builder()
                    .socketOptions(SocketOptions.builder().connectTimeout(Duration.ofSeconds(10)).build()).build();

            client.setOptions(initialOptions);

            // Start async connection
            ConnectionFuture<StatefulRedisConnection<String, String>> connectionFuture = client.connectAsync(StringCodec.UTF8,
                    redisURI);

            // Immediately change options (race condition!)
            ClientOptions newOptions = ClientOptions.builder()
                    .socketOptions(SocketOptions.builder().connectTimeout(Duration.ofSeconds(60)).build()).build();

            client.setOptions(newOptions);

            // Wait for connection to complete
            StatefulRedisConnection<String, String> connection = connectionFuture.get(10, TimeUnit.SECONDS);
            connection.close();

            // Verify the race condition: different components saw different options
            List<ClientOptions> capturedOptions = client.getCapturedOptions();

            // Multiple getOptions() calls should have been made during connection
            assertThat(capturedOptions).hasSizeGreaterThan(1);

            // Check if different options were observed
            boolean hasInitialOptions = capturedOptions.stream()
                    .anyMatch(opts -> opts.getSocketOptions().getConnectTimeout().equals(Duration.ofSeconds(10)));

            boolean hasNewOptions = capturedOptions.stream()
                    .anyMatch(opts -> opts.getSocketOptions().getConnectTimeout().equals(Duration.ofSeconds(60)));

            // Race condition: some components saw initial options, others saw new options
            assertThat(hasInitialOptions && hasNewOptions)
                    .as("Race condition detected: different components saw different ClientOptions").isTrue();

        } finally {
            client.shutdown();
        }
    }

    /**
     * Test RedisClient that captures all getOptions() calls during connection establishment.
     */
    static class TestRedisClient extends RedisClient {

        private final List<ClientOptions> capturedOptions = new CopyOnWriteArrayList<>();

        private volatile boolean connectionStarted = false;

        TestRedisClient(RedisURI redisURI) {
            super(null, redisURI);
        }

        @Override
        public ClientOptions getOptions() {
            ClientOptions options = super.getOptions();

            // Only intercept getOptions() calls that happen during connection
            if (connectionStarted) {
                capturedOptions.add(options);
            }

            return options;
        }

        @Override
        public <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectAsync(io.lettuce.core.codec.RedisCodec<K, V> codec,
                RedisURI redisURI) {
            // Mark that connection has started
            connectionStarted = true;
            return super.connectAsync(codec, redisURI);
        }

        List<ClientOptions> getCapturedOptions() {
            return capturedOptions;
        }

    }

}