From fb8fae03ac81d741c61c917bcbfbd67280ee9966 Mon Sep 17 00:00:00 2001 From: Jonathan Norris Date: Thu, 6 Aug 2026 09:36:12 -0400 Subject: [PATCH 1/2] feat: add ComparisonStrategy to multiprovider Signed-off-by: Jonathan Norris --- .../sdk/multiprovider/ComparisonStrategy.java | 254 ++++++++++++++++ .../multiprovider/ComparisonStrategyTest.java | 283 ++++++++++++++++++ 2 files changed, 537 insertions(+) create mode 100644 src/main/java/dev/openfeature/sdk/multiprovider/ComparisonStrategy.java create mode 100644 src/test/java/dev/openfeature/sdk/multiprovider/ComparisonStrategyTest.java 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(); + } + } +} From abc02f69fa062ce0b14c0facc6140f36cdd7ca41 Mon Sep 17 00:00:00 2001 From: Jonathan Norris Date: Mon, 10 Aug 2026 14:32:22 -0400 Subject: [PATCH 2/2] refactor: narrow ComparisonStrategy public surface and isolate timeouts Signed-off-by: Jonathan Norris --- .../sdk/multiprovider/ComparisonStrategy.java | 138 +++++++++++++----- .../multiprovider/ComparisonStrategyTest.java | 126 ++++++++++++++-- 2 files changed, 215 insertions(+), 49 deletions(-) diff --git a/src/main/java/dev/openfeature/sdk/multiprovider/ComparisonStrategy.java b/src/main/java/dev/openfeature/sdk/multiprovider/ComparisonStrategy.java index 69b030727..8c55d6e16 100644 --- a/src/main/java/dev/openfeature/sdk/multiprovider/ComparisonStrategy.java +++ b/src/main/java/dev/openfeature/sdk/multiprovider/ComparisonStrategy.java @@ -4,6 +4,7 @@ import dev.openfeature.sdk.EvaluationContext; import dev.openfeature.sdk.FeatureProvider; import dev.openfeature.sdk.ProviderEvaluation; +import dev.openfeature.sdk.internal.ConfigurableThreadFactory; import java.util.ArrayList; import java.util.Collections; import java.util.LinkedHashMap; @@ -14,7 +15,7 @@ import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; -import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; @@ -30,11 +31,26 @@ * 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. + * + *

Providers that do not respond before the internal timeout do not fail the evaluation. The + * fallback provider's result is still returned, carrying a {@link ProviderError} for each provider + * that timed out, so that a slow provider under comparison cannot degrade evaluations. Only a + * timeout of the fallback provider itself produces an error result. */ public class ComparisonStrategy implements Strategy { private static final long DEFAULT_TIMEOUT_MS = 30_000; + /** + * Shared pool used when no executor is supplied. + * + *

Provider evaluations block, so they are kept off {@link java.util.concurrent.ForkJoinPool + * #commonPool()} to avoid starving unrelated parallel work in the host application. Threads are + * daemon threads, so this pool never prevents JVM shutdown. + */ + private static final ExecutorService DEFAULT_EXECUTOR = + Executors.newCachedThreadPool(new ConfigurableThreadFactory("openfeature-comparison-strategy", true)); + @Getter private final String fallbackProvider; @@ -45,8 +61,6 @@ public class ComparisonStrategy implements Strategy { /** * 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 */ @@ -57,8 +71,6 @@ public ComparisonStrategy(String fallbackProvider) { /** * 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 @@ -66,11 +78,14 @@ public ComparisonStrategy(String fallbackProvider) { */ public ComparisonStrategy( String fallbackProvider, BiConsumer>> onMismatch) { - this(fallbackProvider, onMismatch, ForkJoinPool.commonPool(), DEFAULT_TIMEOUT_MS); + this(fallbackProvider, onMismatch, DEFAULT_EXECUTOR, DEFAULT_TIMEOUT_MS); } /** - * Constructs a comparison strategy with a caller-supplied executor. + * Constructs a comparison strategy with a caller-supplied executor and timeout. + * + *

Intentionally not public: the executor and timeout are implementation details, and the + * public surface is kept aligned with the js-sdk reference implementation. * * @param fallbackProvider provider name to use as fallback when successful * providers disagree @@ -80,7 +95,7 @@ public ComparisonStrategy( * @param timeoutMs maximum time in milliseconds to wait for all * providers to complete */ - public ComparisonStrategy( + ComparisonStrategy( String fallbackProvider, BiConsumer>> onMismatch, ExecutorService executorService, @@ -111,72 +126,92 @@ public ProviderEvaluation evaluate( int capacity = providers.size() * 4 / 3 + 1; Map> successfulResults = new ConcurrentHashMap<>(capacity); Map providerErrors = new ConcurrentHashMap<>(capacity); + Map timeoutErrors = new ConcurrentHashMap<>(capacity); Optional> runFailure = - runEvaluations(providers, providerFunction, successfulResults, providerErrors); + runEvaluations(providers, providerFunction, successfulResults, providerErrors, timeoutErrors); if (runFailure.isPresent()) { return runFailure.get(); } if (!providerErrors.isEmpty()) { - return errorResult("Provider errors during comparison", providers, providerErrors); + return errorResult( + "Provider errors during comparison", orderedErrors(providers, providerErrors, timeoutErrors)); } ProviderEvaluation fallbackResult = successfulResults.get(fallbackProvider); if (fallbackResult == null) { return errorResult( - "Fallback provider did not return a successful evaluation: " + fallbackProvider, - providers, - providerErrors); + fallbackFailureMessage(timeoutErrors), orderedErrors(providers, providerErrors, timeoutErrors)); + } + + if (onMismatch != null && !allEvaluationsMatch(successfulResults)) { + onMismatch.accept(key, orderedResults(providers, successfulResults)); } - if (allEvaluationsMatch(successfulResults)) { + if (timeoutErrors.isEmpty()) { return fallbackResult; } + // A provider under comparison was too slow. Report it, but keep serving the fallback result. + return withProviderErrors(fallbackResult, orderedErrors(providers, providerErrors, timeoutErrors)); + } - if (onMismatch != null) { - onMismatch.accept(key, orderedResults(providers, successfulResults)); + private String fallbackFailureMessage(Map timeoutErrors) { + if (timeoutErrors.containsKey(fallbackProvider)) { + return "Fallback provider did not respond within " + timeoutMs + "ms: " + fallbackProvider; } - return fallbackResult; + return "Fallback provider did not return a successful evaluation: " + fallbackProvider; } /** - * Evaluates every provider in parallel, recording each outcome into {@code successfulResults} or - * {@code providerErrors}. + * Evaluates every provider in parallel, recording each outcome into {@code successfulResults}, + * {@code providerErrors}, or, for providers that did not finish in time, {@code timeoutErrors}. * - * @return an error evaluation if the parallel run itself could not complete (timeout, - * interruption, or executor failure), otherwise {@link Optional#empty()} + * @return an error evaluation if the parallel run itself could not complete (interruption or + * executor failure), otherwise {@link Optional#empty()} */ private Optional> runEvaluations( Map providers, Function> providerFunction, Map> successfulResults, - Map providerErrors) { + Map providerErrors, + Map timeoutErrors) { try { + List providerNames = new ArrayList<>(providers.keySet()); List> tasks = new ArrayList<>(providers.size()); - for (Map.Entry entry : providers.entrySet()) { - String providerName = entry.getKey(); - FeatureProvider provider = entry.getValue(); + for (String providerName : providerNames) { + FeatureProvider provider = providers.get(providerName); tasks.add(() -> { recordEvaluation(providerName, provider, providerFunction, successfulResults, providerErrors); return null; }); } + // invokeAll returns futures in task submission order, which matches providerNames. List> futures = executorService.invokeAll(tasks, timeoutMs, TimeUnit.MILLISECONDS); - for (Future future : futures) { + for (int i = 0; i < futures.size(); i++) { + Future future = futures.get(i); + String providerName = providerNames.get(i); if (future.isCancelled()) { - return Optional.of(errorResult( - "Comparison strategy timed out after " + timeoutMs + "ms", providers, providerErrors)); + timeoutErrors.put( + providerName, + ProviderError.fromResult( + providerName, + ErrorCode.GENERAL, + "Provider did not respond within " + timeoutMs + "ms")); + } else { + future.get(); } - future.get(); } return Optional.empty(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - return Optional.of( - errorResult("Comparison strategy interrupted: " + e.getMessage(), providers, providerErrors)); + return Optional.of(errorResult( + "Comparison strategy interrupted: " + e.getMessage(), + orderedErrors(providers, providerErrors, timeoutErrors))); } catch (Exception e) { - return Optional.of(errorResult("Comparison strategy failed: " + e.getMessage(), providers, providerErrors)); + return Optional.of(errorResult( + "Comparison strategy failed: " + e.getMessage(), + orderedErrors(providers, providerErrors, timeoutErrors))); } } @@ -205,22 +240,47 @@ private void recordEvaluation( } } + /** Builds a {@link MultiProviderEvaluation} carrying per-provider error details. */ + private ProviderEvaluation errorResult(String baseMessage, List orderedErrors) { + return MultiProviderEvaluation.builder() + .errorCode(ErrorCode.GENERAL) + .errorMessage(ProviderError.buildAggregateMessage(baseMessage, orderedErrors)) + .providerErrors(orderedErrors) + .build(); + } + /** - * Builds a {@link MultiProviderEvaluation} carrying per-provider error details, ordered by the - * provider registration order so the aggregate message is stable across runs. + * Merges the recorded errors into a single list ordered by provider registration order, so that + * aggregate messages are stable across runs. */ - private ProviderEvaluation errorResult( - String baseMessage, Map providers, Map providerErrors) { - List orderedErrors = new ArrayList<>(providerErrors.size()); + private List orderedErrors( + Map providers, + Map providerErrors, + Map timeoutErrors) { + List orderedErrors = new ArrayList<>(providerErrors.size() + timeoutErrors.size()); for (String providerName : providers.keySet()) { ProviderError error = providerErrors.get(providerName); + if (error == null) { + error = timeoutErrors.get(providerName); + } if (error != null) { orderedErrors.add(error); } } + return orderedErrors; + } + + /** + * Returns the successful evaluation as a {@link MultiProviderEvaluation} carrying the given + * per-provider errors, so callers can see which providers were skipped or timed out. + */ + private ProviderEvaluation withProviderErrors( + ProviderEvaluation evaluation, List orderedErrors) { return MultiProviderEvaluation.builder() - .errorCode(ErrorCode.GENERAL) - .errorMessage(ProviderError.buildAggregateMessage(baseMessage, orderedErrors)) + .value(evaluation.getValue()) + .variant(evaluation.getVariant()) + .reason(evaluation.getReason()) + .flagMetadata(evaluation.getFlagMetadata()) .providerErrors(orderedErrors) .build(); } diff --git a/src/test/java/dev/openfeature/sdk/multiprovider/ComparisonStrategyTest.java b/src/test/java/dev/openfeature/sdk/multiprovider/ComparisonStrategyTest.java index 5589da657..2a1751cfd 100644 --- a/src/test/java/dev/openfeature/sdk/multiprovider/ComparisonStrategyTest.java +++ b/src/test/java/dev/openfeature/sdk/multiprovider/ComparisonStrategyTest.java @@ -119,8 +119,8 @@ void shouldEvaluateProvidersConcurrently() { 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. + // A dedicated pool of exactly two threads, so the assertion below cannot depend on how the + // strategy's shared default pool happens to be sized or already occupied. ExecutorService executor = Executors.newFixedThreadPool(2); try { ComparisonStrategy strategy = new ComparisonStrategy("provider1", null, executor, 5_000); @@ -251,7 +251,7 @@ void shouldTreatNullEvaluationAsProviderError() { } @Test - void shouldReturnTimeoutErrorWhenProvidersExceedTheTimeout() { + void shouldReturnErrorWhenTheFallbackProviderExceedsTheTimeout() { setupProviderSuccess(mockProvider1, "ok"); setupProviderSuccess(mockProvider2, "ok"); @@ -264,20 +264,126 @@ void shouldReturnTimeoutErrorWhenProvidersExceedTheTimeout() { 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(); - } + sleepUninterruptibly(5_000); 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()); + result.getErrorMessage().contains("Fallback provider did not respond within 50ms: provider1"), + "Expected a fallback timeout message, got: " + result.getErrorMessage()); + } finally { + executor.shutdownNow(); + } + } + + @Test + void shouldReturnFallbackResultAndReportTimeoutWhenComparedProviderIsTooSlow() { + setupProviderSuccess(mockProvider1, "fast"); + setupProviderSuccess(mockProvider2, "slow"); + + 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, 100); + ProviderEvaluation result = + strategy.evaluate(providers, FLAG_KEY, DEFAULT_STRING, null, provider -> { + if (provider == mockProvider2) { + sleepUninterruptibly(5_000); + } + return provider.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null); + }); + + assertNull(result.getErrorCode(), "a slow compared provider must not fail the evaluation"); + assertEquals("fast", result.getValue()); + + List providerErrors = ((MultiProviderEvaluation) result).getProviderErrors(); + assertEquals(1, providerErrors.size()); + assertEquals("provider2", providerErrors.get(0).getProviderName()); + assertEquals( + "Provider did not respond within 100ms", + providerErrors.get(0).getErrorMessage()); + } finally { + executor.shutdownNow(); + } + } + + @Test + void shouldNotInvokeMismatchCallbackWhenTheOnlyOtherProviderTimedOut() { + setupProviderSuccess(mockProvider1, "fast"); + setupProviderSuccess(mockProvider2, "slow"); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + providers.put("provider2", mockProvider2); + + ExecutorService executor = Executors.newFixedThreadPool(2); + AtomicInteger callbackCount = new AtomicInteger(); + try { + ComparisonStrategy strategy = new ComparisonStrategy( + "provider1", (key, evaluations) -> callbackCount.incrementAndGet(), executor, 100); + strategy.evaluate(providers, FLAG_KEY, DEFAULT_STRING, null, provider -> { + if (provider == mockProvider2) { + sleepUninterruptibly(5_000); + } + return provider.getStringEvaluation(FLAG_KEY, DEFAULT_STRING, null); + }); + + assertEquals(0, callbackCount.get(), "a timed-out provider has no value to disagree with"); } finally { executor.shutdownNow(); } } + + @Test + void shouldNotInvokeMismatchCallbackWhenProvidersAgree() { + setupProviderSuccess(mockProvider1, "same"); + setupProviderSuccess(mockProvider2, "same"); + + 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("same", result.getValue()); + assertNull(result.getErrorCode()); + assertEquals(0, callbackCount.get()); + } + + @Test + void shouldReturnErrorWhenTheExecutorRejectsTheEvaluations() { + setupProviderSuccess(mockProvider1, "ok"); + + Map providers = new LinkedHashMap<>(); + providers.put("provider1", mockProvider1); + + ExecutorService executor = Executors.newFixedThreadPool(1); + executor.shutdown(); + + ComparisonStrategy strategy = new ComparisonStrategy("provider1", null, executor, 1_000); + 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("Comparison strategy failed"), + "Expected a strategy failure message, got: " + result.getErrorMessage()); + } + + private static void sleepUninterruptibly(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } }