[automatic failover] Add example for automatic failover (#3568)

* add DatabaseConfig.Builder

* healthCheckStrategySupplier now defaults to PingStrategy.DEFAULT in the builder

 - When using the builder without setting healthCheckStrategySupplier: Health checks will use PingStrategy.DEFAULT
 - When explicitly setting to null: Health checks will be disabled (as documented)
 - When setting to a custom supplier: Uses the custom health check strategy

 Example Usage:
 // Uses PingStrategy.DEFAULT for health checks
 DatabaseConfig config1 = DatabaseConfig.builder(uri)
     .weight(1.0f)
     .build();

 // Explicitly disables health checks
 DatabaseConfig config2 = DatabaseConfig.builder(uri)
     .healthCheckStrategySupplier(null)
     .build();

 // Uses custom health check strategy
 DatabaseConfig config3 = DatabaseConfig.builder(uri)
     .healthCheckStrategySupplier(customSupplier)
     .build();

* HealthCheckStrategySupplier.NO_HEALTH_CHECK instead null

* Remove DatabaseConfig constructors

// To create DatabaseConfig use provided builder
DatabaseConfig config = DatabaseConfig.builder(redisURI)
    .weight(1.5f)
    .clientOptions(options)
    .circuitBreakerConfig(cbConfig)
    .healthCheckStrategySupplier(supplier)
    .build();

* remove redundant public modifiers

* Builder for CircuitBreakerConfig

// Minimal configuration with defaults
CircuitBreakerConfig config = CircuitBreakerConfig.builder().build();

// Custom configuration
CircuitBreakerConfig config = CircuitBreakerConfig.builder()
    .failureRateThreshold(25.0f)
    .minimumNumberOfFailures(500)
    .metricsWindowSize(5)
    .build();

// With custom tracked exceptions
Set<Class<? extends Throwable>> customExceptions = new HashSet<>();
customExceptions.add(RuntimeException.class);

CircuitBreakerConfig config = CircuitBreakerConfig.builder()
    .failureRateThreshold(15.5f)
    .minimumNumberOfFailures(200)
    .trackedExceptions(customExceptions)
    .metricsWindowSize(3)
    .build();

* enforce min window size of 2s

* tracked exceptions should not be null

* add convenience methods for Tracked Exceptions

//Combine add and remove
CircuitBreakerConfig config = CircuitBreakerConfig.builder()
    .addTrackedExceptions(MyCustomException.class)
    .removeTrackedExceptions(TimeoutException.class)
    .build();

// Replace all tracked exceptions
Set<Class<? extends Throwable>> customExceptions = new HashSet<>();
customExceptions.add(RuntimeException.class);
customExceptions.add(IOException.class);
CircuitBreakerConfig config = CircuitBreakerConfig.builder()
    .trackedExceptions(customExceptions)
    .build();

* remove option to configure per database clientOptions till #3572 is resolved

* Disable health checks in test configs to isolate circuit breaker testing

Configure DB1, DB2, and DB3 with NO_HEALTH_CHECK to prevent health check
interference when testing circuit breaker failure detection.

* forma

* clean up

* address review comments (Copilot)

* Add example for automatic failover

* Use builders

* shutdown primary instance

* remove unused imports

* Update src/test/java/io/lettuce/examples/AutomaticFailover.java

Co-authored-by: atakavci <a_takavci@yahoo.com>

* revert accidentally disabled user timeout config

---------

Co-authored-by: ggivo <ivo.gaydazhiev@redis.com>
Co-authored-by: atakavci <a_takavci@yahoo.com>

Author: 3 people

src/test/java/io/lettuce/examples/AutomaticFailover.java

@@ -0,0 +1,166 @@
+package io.lettuce.examples;
+
+import io.lettuce.core.ClientOptions;
+import io.lettuce.core.RedisClient;
+import io.lettuce.core.RedisConnectionException;
+import io.lettuce.core.RedisURI;
+import io.lettuce.core.SocketOptions;
+import io.lettuce.core.api.StatefulRedisConnection;
+import io.lettuce.core.api.reactive.RedisReactiveCommands;
+import io.lettuce.core.failover.CircuitBreaker;
+import io.lettuce.core.failover.DatabaseConfig;
+import io.lettuce.core.failover.MultiDbClient;
+import io.lettuce.core.failover.api.StatefulRedisMultiDbConnection;
+import io.lettuce.core.failover.health.PingStrategy;
+import io.lettuce.test.Wait;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.time.Duration;
+import java.util.*;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Example of automatic failover using MultiDbClient. Automatic Failover API is subject to change since we are still in Beta and
+ * actively improving the API.
+ */
+public class AutomaticFailover {
+
+    private static final Logger log = LoggerFactory.getLogger(AutomaticFailover.class);
+
+    public static void main(String[] args) {
+        // Setup: Configure MultiDbClient
+        // Both endpoints should be available in the test environment.
+        // At this point multiDbClient lib does not check if the endpoints are actually available.
+        // This limitation will be addressed in the future.
+        // Local Redis instances can be started with:
+        // docker run -p 6380:6379 -it redis:8.2.1
+        // docker run -p 6381:6379 -it redis:8.2.1
+        List<String> endpoints = Arrays.asList("redis://localhost:6380", "redis://localhost:6381");
+
+        SocketOptions.TcpUserTimeoutOptions tcpUserTimeout = SocketOptions.TcpUserTimeoutOptions.builder()
+                .tcpUserTimeout(Duration.ofSeconds(4)).enable().build();
+
+        SocketOptions.KeepAliveOptions keepAliveOptions = SocketOptions.KeepAliveOptions.builder()
+                .interval(Duration.ofSeconds(1)).idle(Duration.ofSeconds(1)).count(3).enable().build();
+
+        ClientOptions clientOptions = ClientOptions.builder()
+                .socketOptions(SocketOptions.builder().tcpUserTimeout(tcpUserTimeout).keepAlive(keepAliveOptions).build())
+                .build();
+
+        List<DatabaseConfig> databaseConfigs = createDatabaseConfigs(endpoints);
+        MultiDbClient multiDbClient = MultiDbClient.create(databaseConfigs);
+
+        // Automatic failback are not supported in the current Beta release.
+        multiDbClient.setOptions(clientOptions);
+
+        // Connect to the MultiDbClient
+        StatefulRedisMultiDbConnection<String, String> connection = multiDbClient.connect();
+        log.info("Connected to {}", connection.getCurrentEndpoint());
+        log.info("Available Endpoints: {}", connection.getEndpoints());
+
+        // Get reactive commands interface
+        RedisReactiveCommands<String, String> reactive = connection.reactive();
+
+        String keyName = "multidb-counter";
+
+        // Setup: Set initial counter value
+        StepVerifier.create(reactive.set(keyName, "0")).expectNext("OK").verifyComplete();
+
+        AtomicLong commandsSubmitted = new AtomicLong();
+        List<Throwable> capturedExceptions = new CopyOnWriteArrayList<>();
+
+        // Start a flux that imitates an application using the multiDbClient
+        // This follows the exact pattern from ConnectionInterruptionReactiveTest#testWithReactiveCommands
+        Disposable subscription = Flux.interval(Duration.ofMillis(100)).flatMap(i -> reactive.incr(keyName)
+                // We should count all attempts, because Lettuce retransmits failed commands
+                .doFinally(value -> {
+                    commandsSubmitted.incrementAndGet();
+                    log.info("Commands submitted {}", commandsSubmitted.get());
+                }).onErrorResume(e -> {
+                    capturedExceptions.add(e);
+                    return Mono.empty();
+                })).subscribe();
+
+        // Wait for some commands to be executed before triggering the fault
+        Wait.untilTrue(() -> commandsSubmitted.get() > 10).waitOrTimeout();
+
+        // Direct connection to the Redis instances used to trigger Primary Redis instance shutdown
+        // and verify that commands are executed on the secondary Redis instance
+        RedisClient directClient = RedisClient.create();
+
+        // Shutdown the current Redis instance to trigger failover
+        RedisURI currentEndpoint = connection.getCurrentEndpoint();
+        log.info("Stoping Redis server [{}] to trigger failover", currentEndpoint);
+        shutdownRedisInstance(directClient, currentEndpoint);
+
+        // Wait for Commands to start being executed on the secondary Redis instances
+        RedisURI secondaryUri = RedisURI.create(endpoints.get(1));
+        StatefulRedisConnection<String, String> secondary = directClient.connect(secondaryUri);
+        log.info("Waiting for commands to be executed on [{}]", secondaryUri);
+        Wait.untilTrue(() -> (getMulitDbCounter(secondary)) > 20).during(Duration.ofSeconds(10)).waitOrTimeout();
+
+        // Stop the command execution
+        subscription.dispose();
+
+        log.info("Commands submitted total: {}", commandsSubmitted.get());
+        log.info("Commands executed on secondary [{}]: {}", secondaryUri, getMulitDbCounter(secondary));
+        log.info("Captured exceptions: {}", capturedExceptions);
+
+        // Cleanup
+        directClient.shutdown();
+        multiDbClient.shutdown();
+    }
+
+    private static int getMulitDbCounter(StatefulRedisConnection<String, String> connection) {
+        String count = connection.sync().get("multidb-counter");
+        return count == null ? 0 : Integer.parseInt(count);
+    }
+
+    private static void shutdownRedisInstance(RedisClient redis, RedisURI redisUri) {
+        log.info("Shutting down Redis instance [{}]", redisUri);
+        try (StatefulRedisConnection<String, String> redis1 = redis.connect(redisUri)) {
+            redis1.sync().shutdown(true);
+
+            Wait.untilTrue(() -> {
+                try {
+                    redis.connect(redisUri);
+                } catch (Exception e) {
+                    return e instanceof RedisConnectionException;
+                }
+                return false;
+            }).waitOrTimeout();
+
+            log.info("Redis instance [{}] is down", redisUri);
+        }
+    }
+
+    /**
+     * @return list of DatabaseConfig instances
+     */
+    private static List<DatabaseConfig> createDatabaseConfigs(List<String> endpoints) {
+        List<DatabaseConfig> configs = new ArrayList<>();
+
+        CircuitBreaker.CircuitBreakerConfig circuitBreakerConfig = CircuitBreaker.CircuitBreakerConfig.builder()
+                .failureRateThreshold(10.0f).minimumNumberOfFailures(5).metricsWindowSize(5).build();
+
+        // Create a DatabaseConfig for each endpoint
+        float weight = 1.0f;
+        for (String endpointUri : endpoints) {
+            configs.add(DatabaseConfig.builder(RedisURI.create(endpointUri)).weight(weight)
+                    .circuitBreakerConfig(circuitBreakerConfig).healthCheckStrategySupplier(PingStrategy.DEFAULT).build());
+
+            weight /= 2; // Decrease weight for subsequent endpoints
+        }
+
+        log.info("Created {} DatabaseConfig(s) from endpoint", configs.size());
+        return configs;
+    }
+
+}