diff --git a/src/main/java/dev/openfeature/sdk/multiprovider/ComparisonStrategy.java b/src/main/java/dev/openfeature/sdk/multiprovider/ComparisonStrategy.java new file mode 100644 index 000000000..69b030727 --- /dev/null +++ b/src/main/java/dev/openfeature/sdk/multiprovider/ComparisonStrategy.java @@ -0,0 +1,254 @@ +package dev.openfeature.sdk.multiprovider; + +import dev.openfeature.sdk.ErrorCode; +import dev.openfeature.sdk.EvaluationContext; +import dev.openfeature.sdk.FeatureProvider; +import dev.openfeature.sdk.ProviderEvaluation; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; +import java.util.function.Function; +import lombok.Getter; + +/** + * Comparison strategy. + * + *

Evaluates all providers in parallel and compares successful results. + * If all providers agree on the value, the fallback provider's result is returned. + * If providers disagree, the optional {@code onMismatch} callback is invoked + * and the fallback provider's result is returned. + * If any provider returns an error, all errors are collected and a {@link MultiProviderEvaluation} + * with {@link ErrorCode#GENERAL} and per-provider {@link ProviderError} details is returned. + */ +public class ComparisonStrategy implements Strategy { + + private static final long DEFAULT_TIMEOUT_MS = 30_000; + + @Getter + private final String fallbackProvider; + + private final BiConsumer>> onMismatch; + private final ExecutorService executorService; + private final long timeoutMs; + + /** + * Constructs a comparison strategy with a fallback provider. + * + *

Uses a shared {@link ForkJoinPool#commonPool()} for parallel evaluation. + * + * @param fallbackProvider provider name to use as fallback when successful + * providers disagree + */ + public ComparisonStrategy(String fallbackProvider) { + this(fallbackProvider, null); + } + + /** + * Constructs a comparison strategy with fallback provider and mismatch callback. + * + *

Uses a shared {@link ForkJoinPool#commonPool()} for parallel evaluation. + * + * @param fallbackProvider provider name to use as fallback when successful + * providers disagree + * @param onMismatch callback invoked with all successful evaluations + * when they disagree + */ + public ComparisonStrategy( + String fallbackProvider, BiConsumer>> onMismatch) { + this(fallbackProvider, onMismatch, ForkJoinPool.commonPool(), DEFAULT_TIMEOUT_MS); + } + + /** + * Constructs a comparison strategy with a caller-supplied executor. + * + * @param fallbackProvider provider name to use as fallback when successful + * providers disagree + * @param onMismatch callback invoked with all successful evaluations + * when they disagree (may be {@code null}) + * @param executorService executor to use for parallel evaluation + * @param timeoutMs maximum time in milliseconds to wait for all + * providers to complete + */ + public ComparisonStrategy( + String fallbackProvider, + BiConsumer>> onMismatch, + ExecutorService executorService, + long timeoutMs) { + this.fallbackProvider = Objects.requireNonNull(fallbackProvider, "fallbackProvider must not be null"); + this.onMismatch = onMismatch; + this.executorService = Objects.requireNonNull(executorService, "executorService must not be null"); + this.timeoutMs = timeoutMs; + } + + @Override + public ProviderEvaluation evaluate( + Map providers, + String key, + T defaultValue, + EvaluationContext ctx, + Function> providerFunction) { + if (providers.isEmpty()) { + return ProviderEvaluation.builder() + .errorCode(ErrorCode.GENERAL) + .errorMessage("No providers configured") + .build(); + } + if (!providers.containsKey(fallbackProvider)) { + throw new IllegalArgumentException("fallbackProvider not found in providers: " + fallbackProvider); + } + + int capacity = providers.size() * 4 / 3 + 1; + Map> successfulResults = new ConcurrentHashMap<>(capacity); + Map providerErrors = new ConcurrentHashMap<>(capacity); + + Optional> runFailure = + runEvaluations(providers, providerFunction, successfulResults, providerErrors); + if (runFailure.isPresent()) { + return runFailure.get(); + } + + if (!providerErrors.isEmpty()) { + return errorResult("Provider errors during comparison", providers, providerErrors); + } + + ProviderEvaluation fallbackResult = successfulResults.get(fallbackProvider); + if (fallbackResult == null) { + return errorResult( + "Fallback provider did not return a successful evaluation: " + fallbackProvider, + providers, + providerErrors); + } + + if (allEvaluationsMatch(successfulResults)) { + return fallbackResult; + } + + if (onMismatch != null) { + onMismatch.accept(key, orderedResults(providers, successfulResults)); + } + return fallbackResult; + } + + /** + * Evaluates every provider in parallel, recording each outcome into {@code successfulResults} or + * {@code providerErrors}. + * + * @return an error evaluation if the parallel run itself could not complete (timeout, + * interruption, or executor failure), otherwise {@link Optional#empty()} + */ + private Optional> runEvaluations( + Map providers, + Function> providerFunction, + Map> successfulResults, + Map providerErrors) { + try { + List> tasks = new ArrayList<>(providers.size()); + for (Map.Entry entry : providers.entrySet()) { + String providerName = entry.getKey(); + FeatureProvider provider = entry.getValue(); + tasks.add(() -> { + recordEvaluation(providerName, provider, providerFunction, successfulResults, providerErrors); + return null; + }); + } + List> futures = executorService.invokeAll(tasks, timeoutMs, TimeUnit.MILLISECONDS); + for (Future future : futures) { + if (future.isCancelled()) { + return Optional.of(errorResult( + "Comparison strategy timed out after " + timeoutMs + "ms", providers, providerErrors)); + } + future.get(); + } + return Optional.empty(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return Optional.of( + errorResult("Comparison strategy interrupted: " + e.getMessage(), providers, providerErrors)); + } catch (Exception e) { + return Optional.of(errorResult("Comparison strategy failed: " + e.getMessage(), providers, providerErrors)); + } + } + + /** Evaluates a single provider, recording either its result or its error. */ + private void recordEvaluation( + String providerName, + FeatureProvider provider, + Function> providerFunction, + Map> successfulResults, + Map providerErrors) { + try { + ProviderEvaluation evaluation = providerFunction.apply(provider); + if (evaluation == null) { + providerErrors.put( + providerName, ProviderError.fromResult(providerName, ErrorCode.GENERAL, "null evaluation")); + } else if (evaluation.getErrorCode() == null) { + successfulResults.put(providerName, evaluation); + } else { + providerErrors.put( + providerName, + ProviderError.fromResult( + providerName, evaluation.getErrorCode(), evaluation.getErrorMessage())); + } + } catch (Exception e) { + providerErrors.put(providerName, ProviderError.fromException(providerName, e)); + } + } + + /** + * Builds a {@link MultiProviderEvaluation} carrying per-provider error details, ordered by the + * provider registration order so the aggregate message is stable across runs. + */ + private ProviderEvaluation errorResult( + String baseMessage, Map providers, Map providerErrors) { + List orderedErrors = new ArrayList<>(providerErrors.size()); + for (String providerName : providers.keySet()) { + ProviderError error = providerErrors.get(providerName); + if (error != null) { + orderedErrors.add(error); + } + } + return MultiProviderEvaluation.builder() + .errorCode(ErrorCode.GENERAL) + .errorMessage(ProviderError.buildAggregateMessage(baseMessage, orderedErrors)) + .providerErrors(orderedErrors) + .build(); + } + + /** Returns the successful evaluations in provider registration order. */ + private Map> orderedResults( + Map providers, Map> successfulResults) { + Map> ordered = new LinkedHashMap<>(); + for (String providerName : providers.keySet()) { + ProviderEvaluation evaluation = successfulResults.get(providerName); + if (evaluation != null) { + ordered.put(providerName, evaluation); + } + } + return Collections.unmodifiableMap(ordered); + } + + private boolean allEvaluationsMatch(Map> results) { + ProviderEvaluation baseline = null; + for (ProviderEvaluation evaluation : results.values()) { + if (baseline == null) { + baseline = evaluation; + continue; + } + if (!Objects.equals(baseline.getValue(), evaluation.getValue())) { + return false; + } + } + return true; + } +} diff --git a/src/test/java/dev/openfeature/sdk/multiprovider/ComparisonStrategyTest.java b/src/test/java/dev/openfeature/sdk/multiprovider/ComparisonStrategyTest.java new file mode 100644 index 000000000..5589da657 --- /dev/null +++ b/src/test/java/dev/openfeature/sdk/multiprovider/ComparisonStrategyTest.java @@ -0,0 +1,283 @@ +package dev.openfeature.sdk.multiprovider; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.when; + +import dev.openfeature.sdk.ErrorCode; +import dev.openfeature.sdk.FeatureProvider; +import dev.openfeature.sdk.ProviderEvaluation; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; + +class ComparisonStrategyTest extends BaseStrategyTest { + + @Test + void shouldReturnFallbackResultWhenAllProvidersAgree() { + setupProviderSuccess(mockProvider1, "same"); + setupProviderSuccess(mockProvider2, "same"); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + ComparisonStrategy strategy = new ComparisonStrategy("provider2"); + ProviderEvaluation result = strategy.evaluate( + providers, FLAG_KEY, DEFAULT_STRING, null, p -> p.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null)); + + assertNotNull(result); + assertEquals("same", result.getValue()); + assertNull(result.getErrorCode()); + } + + @Test + void shouldCallMismatchCallbackAndReturnFallbackResult() { + setupProviderSuccess(mockProvider1, "first"); + setupProviderSuccess(mockProvider2, "second"); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + AtomicInteger callbackCount = new AtomicInteger(); + ComparisonStrategy strategy = + new ComparisonStrategy("provider2", (key, evaluations) -> callbackCount.incrementAndGet()); + + ProviderEvaluation result = strategy.evaluate( + providers, FLAG_KEY, DEFAULT_STRING, null, p -> p.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null)); + + assertEquals("second", result.getValue()); + assertNull(result.getErrorCode()); + assertEquals(1, callbackCount.get()); + } + + @Test + void shouldReturnGeneralErrorWhenAnyProviderFails() { + setupProviderSuccess(mockProvider1, "ok"); + setupProviderError(mockProvider2, ErrorCode.PARSE_ERROR); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + ComparisonStrategy strategy = new ComparisonStrategy("provider1"); + ProviderEvaluation result = strategy.evaluate( + providers, FLAG_KEY, DEFAULT_STRING, null, p -> p.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null)); + + assertEquals(ErrorCode.GENERAL, result.getErrorCode()); + assertTrue(result.getErrorMessage().contains("provider2")); + + List providerErrors = ((MultiProviderEvaluation) result).getProviderErrors(); + assertEquals(1, providerErrors.size()); + assertEquals("provider2", providerErrors.get(0).getProviderName()); + assertEquals(ErrorCode.PARSE_ERROR, providerErrors.get(0).getErrorCode()); + } + + @Test + void shouldThrowWhenFallbackProviderIsMissing() { + setupProviderSuccess(mockProvider1, "ok"); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + + ComparisonStrategy strategy = new ComparisonStrategy("provider2"); + assertThrows( + IllegalArgumentException.class, + () -> strategy.evaluate( + providers, + FLAG_KEY, + DEFAULT_STRING, + null, + p -> p.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null))); + } + + @Test + void shouldEvaluateProvidersConcurrently() { + // Use a latch to prove that providers run in parallel: + // both providers block on the latch, so they must be on + // separate threads for the test to complete. + CountDownLatch bothStarted = new CountDownLatch(2); + Set threadNames = ConcurrentHashMap.newKeySet(); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + setupProviderSuccess(mockProvider1, "val"); + setupProviderSuccess(mockProvider2, "val"); + + // A dedicated pool of two threads: the default ForkJoinPool.commonPool() can have a + // parallelism of 1 on single-core runners, which would make this assertion flaky. + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + ComparisonStrategy strategy = new ComparisonStrategy("provider1", null, executor, 5_000); + ProviderEvaluation result = + strategy.evaluate(providers, FLAG_KEY, DEFAULT_STRING, null, provider -> { + threadNames.add(Thread.currentThread().getName()); + bothStarted.countDown(); + try { + // Wait for both providers to signal they've started + bothStarted.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return provider.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null); + }); + + assertNotNull(result); + assertEquals("val", result.getValue()); + assertNull(result.getErrorCode()); + // Verify that at least 2 different threads were used + assertTrue( + threadNames.size() >= 2, + "Expected concurrent execution on multiple threads, but only saw: " + threadNames); + } finally { + executor.shutdownNow(); + } + } + + @Test + void shouldCollectAllProviderErrorsWhenMultipleFail() { + setupProviderError(mockProvider1, ErrorCode.PARSE_ERROR); + setupProviderError(mockProvider2, ErrorCode.FLAG_NOT_FOUND); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + ComparisonStrategy strategy = new ComparisonStrategy("provider1"); + ProviderEvaluation result = strategy.evaluate( + providers, FLAG_KEY, DEFAULT_STRING, null, p -> p.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null)); + + assertEquals(ErrorCode.GENERAL, result.getErrorCode()); + assertTrue(result.getErrorMessage().contains("provider1"), "Error should mention provider1"); + assertTrue(result.getErrorMessage().contains("provider2"), "Error should mention provider2"); + + // Errors follow the provider registration order, not the internal concurrent map order. + List providerErrors = ((MultiProviderEvaluation) result).getProviderErrors(); + assertEquals(2, providerErrors.size()); + assertEquals("provider1", providerErrors.get(0).getProviderName()); + assertEquals(ErrorCode.PARSE_ERROR, providerErrors.get(0).getErrorCode()); + assertEquals("provider2", providerErrors.get(1).getProviderName()); + assertEquals(ErrorCode.FLAG_NOT_FOUND, providerErrors.get(1).getErrorCode()); + } + + @Test + void shouldPassSuccessfulEvaluationsInRegistrationOrderToMismatchCallback() { + setupProviderSuccess(mockProvider1, "first"); + setupProviderSuccess(mockProvider2, "second"); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + AtomicReference>> captured = new AtomicReference<>(); + ComparisonStrategy strategy = + new ComparisonStrategy("provider2", (key, evaluations) -> captured.set(evaluations)); + + strategy.evaluate( + providers, FLAG_KEY, DEFAULT_STRING, null, p -> p.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null)); + + assertNotNull(captured.get()); + assertEquals( + List.of("provider1", "provider2"), List.copyOf(captured.get().keySet())); + } + + @Test + void shouldReturnErrorWhenNoProvidersConfigured() { + ComparisonStrategy strategy = new ComparisonStrategy("provider1"); + ProviderEvaluation result = strategy.evaluate( + new LinkedHashMap<>(), + FLAG_KEY, + DEFAULT_STRING, + null, + p -> p.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null)); + + assertEquals(ErrorCode.GENERAL, result.getErrorCode()); + assertEquals("No providers configured", result.getErrorMessage()); + } + + @Test + void shouldRecordThrownProviderExceptionAsProviderError() { + setupProviderSuccess(mockProvider1, "ok"); + setupProviderException(mockProvider2, new IllegalStateException("provider blew up")); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + ComparisonStrategy strategy = new ComparisonStrategy("provider1"); + ProviderEvaluation result = strategy.evaluate( + providers, FLAG_KEY, DEFAULT_STRING, null, p -> p.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null)); + + assertEquals(ErrorCode.GENERAL, result.getErrorCode()); + List providerErrors = ((MultiProviderEvaluation) result).getProviderErrors(); + assertEquals(1, providerErrors.size()); + assertEquals("provider2", providerErrors.get(0).getProviderName()); + assertEquals("provider blew up", providerErrors.get(0).getErrorMessage()); + assertNotNull(providerErrors.get(0).getException()); + } + + @Test + void shouldTreatNullEvaluationAsProviderError() { + setupProviderSuccess(mockProvider1, "ok"); + when(mockProvider2.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null)).thenReturn(null); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + ComparisonStrategy strategy = new ComparisonStrategy("provider1"); + ProviderEvaluation result = strategy.evaluate( + providers, FLAG_KEY, DEFAULT_STRING, null, p -> p.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null)); + + assertEquals(ErrorCode.GENERAL, result.getErrorCode()); + List providerErrors = ((MultiProviderEvaluation) result).getProviderErrors(); + assertEquals(1, providerErrors.size()); + assertEquals("null evaluation", providerErrors.get(0).getErrorMessage()); + } + + @Test + void shouldReturnTimeoutErrorWhenProvidersExceedTheTimeout() { + setupProviderSuccess(mockProvider1, "ok"); + setupProviderSuccess(mockProvider2, "ok"); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + ComparisonStrategy strategy = new ComparisonStrategy("provider1", null, executor, 50); + ProviderEvaluation result = + strategy.evaluate(providers, FLAG_KEY, DEFAULT_STRING, null, provider -> { + try { + Thread.sleep(5_000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return provider.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null); + }); + + assertEquals(ErrorCode.GENERAL, result.getErrorCode()); + assertTrue( + result.getErrorMessage().contains("timed out after 50ms"), + "Expected a timeout message, got: " + result.getErrorMessage()); + } finally { + executor.shutdownNow(); + } + } +}