From 9a66fe85f9f58bf36f871a79b1dc2b29b33f5fc4 Mon Sep 17 00:00:00 2001 From: Bruce Bujon Date: Tue, 20 Oct 2026 13:03:52 +0200 Subject: [PATCH 1/7] feat(ffe): Port the per-subject evaluation telemetry opt-out to the SDK Port master #12884 to the dd-openfeature SDK: the dd_evaluation_telemetry_opt_out context attribute suppresses evaluation events, evaluation metrics and span enrichment for the subject when enabled by Provider.Options.enableEvaluationTelemetryOptOutParameter or DD_FEATURE_FLAGS_EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED. Exposures are unchanged. Provider.Options now defaults to a 30 seconds initialization timeout, as on master. Co-Authored-By: Claude Opus 5.5 --- .../EvaluationTelemetryOptOutHook.java | 55 ++ .../com/datadog/openfeature/Provider.java | 42 +- .../openfeature/internal/config/Settings.java | 12 + .../EvaluationTelemetryOptOutTest.java | 766 ++++++++++++++++++ .../internal/config/SettingsTest.java | 31 + 5 files changed, 903 insertions(+), 3 deletions(-) create mode 100644 dd-openfeature/src/main/java/com/datadog/openfeature/EvaluationTelemetryOptOutHook.java create mode 100644 dd-openfeature/src/test/java/com/datadog/openfeature/EvaluationTelemetryOptOutTest.java diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/EvaluationTelemetryOptOutHook.java b/dd-openfeature/src/main/java/com/datadog/openfeature/EvaluationTelemetryOptOutHook.java new file mode 100644 index 00000000000..0916ea3d5f2 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/EvaluationTelemetryOptOutHook.java @@ -0,0 +1,55 @@ +package com.datadog.openfeature; + +import dev.openfeature.sdk.EvaluationContext; +import dev.openfeature.sdk.FlagEvaluationDetails; +import dev.openfeature.sdk.Hook; +import dev.openfeature.sdk.HookContext; +import dev.openfeature.sdk.Value; +import java.util.Map; +import java.util.Optional; + +/** Gates a Datadog finally-only evaluation telemetry hook, not exposure or application hooks. */ +final class EvaluationTelemetryOptOutHook implements Hook { + private static final String PARAMETER = "dd_evaluation_telemetry_opt_out"; + private static final String DECISION = "datadog.evaluationTelemetryOptOut"; + private final Hook delegate; + + EvaluationTelemetryOptOutHook(final Hook delegate) { + this.delegate = delegate; + } + + @Override + public Optional before( + final HookContext context, final Map hints) { + // Provider before hooks see the merged context, including application before hooks. + // HookData belongs to this evaluation, including nested and concurrent calls. + context.getHookData().set(DECISION, isOptedOut(context.getCtx())); + return Optional.empty(); + } + + @Override + public void error( + final HookContext context, final Exception error, final Map hints) { + // A failing application before hook can prevent our before hook from running. + // Preserve its decision before application error hooks can change the context. + if (context.getHookData().get(DECISION) == null) { + context.getHookData().set(DECISION, isOptedOut(context.getCtx())); + } + } + + @Override + public void finallyAfter( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + if (Boolean.TRUE.equals(context.getHookData().get(DECISION)) || isOptedOut(context.getCtx())) { + return; + } + delegate.finallyAfter(context, details, hints); + } + + private static boolean isOptedOut(final EvaluationContext context) { + final Value value = context == null ? null : context.getValue(PARAMETER); + return value != null && value.isBoolean() && Boolean.TRUE.equals(value.asBoolean()); + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java b/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java index cc421c310c4..1dd5ca59701 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java @@ -111,7 +111,13 @@ public Provider(final Options options) { if (spanEnrichmentHook != null) { hooks.add(spanEnrichmentHook); } - // Exposure hook: always registered last; no-op when the evaluator is not a Datadog one or is + if (options.isEvaluationTelemetryOptOutParameterEnabled(this.settings)) { + for (int i = 0; i < hooks.size(); i++) { + hooks.set(i, new EvaluationTelemetryOptOutHook(hooks.get(i))); + } + } + // Exposure hook: always registered last, and independent of evaluation telemetry opt-out; no-op + // when the evaluator is not a Datadog one or is // not initialized. hooks.add(new ExposureLoggingHook<>(this::datadogEvaluator)); this.providerHooks = @@ -316,8 +322,38 @@ private enum InitializationState { /** The provider options. */ public static class Options { - private long timeout; - private TimeUnit unit; + private long timeout = 30; + private TimeUnit unit = SECONDS; + private Boolean evaluationTelemetryOptOutParameterEnabled; + + /** + * Enables per-evaluation telemetry opt-out through the {@code dd_evaluation_telemetry_opt_out} + * context attribute. Overrides {@code + * DD_FEATURE_FLAGS_EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED}, including when false. + * Without an explicit option, the provider reads configuration at construction. Disabled by + * default. + * + *

Only boolean {@code true} suppresses new Datadog evaluation events, evaluation metrics, + * and automatic flag span enrichment. Missing or other values preserve telemetry. This does not + * change flag results, exposure logging, application hooks, or previously recorded data. + * + *

The attribute remains available for targeting and exposure reporting. Existing privacy and + * reporting settings apply. + * + * @param enabled whether to honor the attribute; configure before constructing the provider. + * @return these options. + */ + public Options enableEvaluationTelemetryOptOutParameter(final boolean enabled) { + this.evaluationTelemetryOptOutParameterEnabled = enabled; + return this; + } + + private boolean isEvaluationTelemetryOptOutParameterEnabled(final Settings settings) { + if (this.evaluationTelemetryOptOutParameterEnabled != null) { + return this.evaluationTelemetryOptOutParameterEnabled; + } + return settings.isEvaluationTelemetryPerSubjectOptOutEnabled(); + } /** * Sets how long provider initialization waits for the initial flag configuration. diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/Settings.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/Settings.java index 11bf5f4e91c..1213c5d2ca4 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/Settings.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/Settings.java @@ -36,6 +36,8 @@ public final class Settings { static final String EVALUATION_COUNTS_ENABLED = "flagging.evaluation.counts.enabled"; static final String EXPOSURES_DATADOG_LOGGING_ENABLED = "feature.flags.exposures.datadog.logging.enabled"; + static final String EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED = + "feature.flags.evaluation.telemetry.per.subject.opt.out.enabled"; static final String API_KEY = "api-key"; static final String SITE = "site"; static final String ENV = "env"; @@ -162,6 +164,16 @@ public boolean isExposuresDatadogLoggingEnabled() { return !Boolean.FALSE.equals(getBoolean(EXPOSURES_DATADOG_LOGGING_ENABLED)); } + /** + * Tells whether the {@code dd_evaluation_telemetry_opt_out} evaluation context attribute can opt + * subjects out of evaluation telemetry. The provider option takes precedence over this setting. + * + * @return whether per-subject evaluation telemetry opt-out is enabled (off by default). + */ + public boolean isEvaluationTelemetryPerSubjectOptOutEnabled() { + return Boolean.TRUE.equals(getBoolean(EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED)); + } + /** * @return the Datadog API key, or {@code null} if not set. */ diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/EvaluationTelemetryOptOutTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/EvaluationTelemetryOptOutTest.java new file mode 100644 index 00000000000..d5e3ab218c0 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/EvaluationTelemetryOptOutTest.java @@ -0,0 +1,766 @@ +package com.datadog.openfeature; + +import static java.util.Arrays.asList; +import static java.util.Collections.emptyList; +import static java.util.Collections.emptyMap; +import static java.util.Collections.singletonList; +import static java.util.Collections.singletonMap; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertAll; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.datadog.openfeature.internal.ConfigurationService; +import com.datadog.openfeature.internal.FeatureFlagsRuntime; +import com.datadog.openfeature.internal.RuntimeServices; +import com.datadog.openfeature.internal.bridge.AgentBridge; +import com.datadog.openfeature.internal.bridge.HealthMetrics; +import com.datadog.openfeature.internal.bridge.SpanEnricher; +import com.datadog.openfeature.internal.config.TestSettings; +import com.datadog.openfeature.internal.exposure.ExposureDeduplicationCache; +import com.datadog.openfeature.internal.exposure.ExposureEvent; +import com.datadog.openfeature.internal.exposure.ExposurePipeline; +import com.datadog.openfeature.internal.flagevaluation.FlagEvalEvent; +import com.datadog.openfeature.internal.flagevaluation.FlagEvaluationPipeline; +import com.datadog.openfeature.internal.ufc.Allocation; +import com.datadog.openfeature.internal.ufc.ConditionConfiguration; +import com.datadog.openfeature.internal.ufc.ConditionOperator; +import com.datadog.openfeature.internal.ufc.Flag; +import com.datadog.openfeature.internal.ufc.Rule; +import com.datadog.openfeature.internal.ufc.ServerConfiguration; +import com.datadog.openfeature.internal.ufc.Split; +import com.datadog.openfeature.internal.ufc.ValueType; +import com.datadog.openfeature.internal.ufc.Variant; +import dev.openfeature.sdk.Client; +import dev.openfeature.sdk.ErrorCode; +import dev.openfeature.sdk.EvaluationContext; +import dev.openfeature.sdk.FlagEvaluationDetails; +import dev.openfeature.sdk.Hook; +import dev.openfeature.sdk.HookContext; +import dev.openfeature.sdk.ImmutableContext; +import dev.openfeature.sdk.MutableContext; +import dev.openfeature.sdk.OpenFeatureAPI; +import dev.openfeature.sdk.Value; +import dev.openfeature.sdk.multiprovider.MultiProvider; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.NullSource; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * Exercises real OpenFeature resolution and the provider telemetry hooks, capturing each output + * boundary: flag evaluation events (EVP), OTel evaluation metrics, span enrichment, exposures and + * {@link ExposureHook} evaluations. + * + *

The provider reads its settings from the real system properties and environment through {@link + * com.datadog.openfeature.internal.bridge.AgentBridges#detect()}, so the configuration scenarios + * use the system property; the environment fallback is covered by {@code SettingsTest}. + */ +class EvaluationTelemetryOptOutTest { + private static final String PARAMETER = "dd_evaluation_telemetry_opt_out"; + private static final String SYSTEM_PROPERTY = + "dd.feature.flags.evaluation.telemetry.per.subject.opt.out.enabled"; + + private final List events = new CopyOnWriteArrayList<>(); + private final List spans = new CopyOnWriteArrayList<>(); + private final List exposures = new CopyOnWriteArrayList<>(); + private final List customerEvaluations = new CopyOnWriteArrayList<>(); + private final AtomicInteger metrics = new AtomicInteger(); + private final ExposureDeduplicationCache exposureCache = new ExposureDeduplicationCache(); + private final SpanEnricher spanEnricher = + new SpanEnricher() { + @Override + public void serialId(final int serialId, final boolean doLog, final String targetingKey) { + EvaluationTelemetryOptOutTest.this.spans.add("serial-id:" + targetingKey); + } + + @Override + public void runtimeDefault(final String flagKey, final Object defaultValue) { + EvaluationTelemetryOptOutTest.this.spans.add("runtime-default:" + flagKey); + } + }; + private final OpenFeatureAPI api = OpenFeatureAPI.getInstance(); + private String previousProperty; + private FeatureFlagsRuntime runtime; + private CapturingEvaluator evaluator; + private Client client; + + @BeforeEach + void setUp() throws Exception { + this.previousProperty = System.getProperty(SYSTEM_PROPERTY); + System.clearProperty(SYSTEM_PROPERTY); + this.api.setEvaluationContext(new MutableContext()); + GlobalOpenTelemetry.resetForTest(); + final OpenTelemetry otel = mock(OpenTelemetry.class, RETURNS_DEEP_STUBS); + final LongCounter counter = mock(LongCounter.class); + when(otel.getMeterProvider() + .meterBuilder("ddtrace.openfeature") + .build() + .counterBuilder("feature_flag.evaluations") + .setUnit("{evaluation}") + .setDescription("Number of feature flag evaluations") + .build()) + .thenReturn(counter); + doAnswer(invocation -> this.metrics.incrementAndGet()) + .when(counter) + .add(eq(1L), any(Attributes.class)); + GlobalOpenTelemetry.set(otel); + this.runtime = capturingRuntime(new CapturingPipeline(this.events)); + this.api.addHooks(new ExposureHook(this.customerEvaluations::add)); + } + + @AfterEach + void tearDown() { + if (this.previousProperty == null) { + System.clearProperty(SYSTEM_PROPERTY); + } else { + System.setProperty(SYSTEM_PROPERTY, this.previousProperty); + } + this.api.clearHooks(); + this.api.setEvaluationContext(new MutableContext()); + this.api.shutdown(); + GlobalOpenTelemetry.resetForTest(); + } + + @Test + void systemPropertyEnablesPerSubjectOptOut() throws Exception { + System.setProperty(SYSTEM_PROPERTY, "true"); + start(new Provider.Options()); + final FlagEvaluationDetails details = evaluate("subject-1", new Value(true)); + assertEquals(1, details.getValue()); + assertEquals("on", details.getVariant()); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + evaluate("subject-2", new Value(false)); + evaluate("subject-3", null); + assertTelemetryCounts(2); + assertEquals(3, this.exposures.size()); + } + + @ParameterizedTest + @NullSource + @ValueSource(strings = {"false", "invalid", ""}) + void systemPropertyDefaultsToDisabled(final String value) throws Exception { + if (value != null) { + System.setProperty(SYSTEM_PROPERTY, value); + } + start(new Provider.Options()); + evaluate("subject-1", new Value(true)); + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + } + + @Test + void explicitFalseOverridesEnabledSystemProperty() throws Exception { + System.setProperty(SYSTEM_PROPERTY, "true"); + start(new Provider.Options().enableEvaluationTelemetryOptOutParameter(false)); + evaluate("subject-1", new Value(true)); + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + } + + @Test + void explicitTrueOverridesDisabledSystemProperty() throws Exception { + System.setProperty(SYSTEM_PROPERTY, "false"); + start(new Provider.Options().enableEvaluationTelemetryOptOutParameter(true)); + evaluate("subject-1", new Value(true)); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + } + + @Test + void systemPropertyIsCapturedAtProviderConstruction() throws Exception { + System.setProperty(SYSTEM_PROPERTY, "true"); + final Provider.Options options = new Provider.Options(); + start(options); + System.setProperty(SYSTEM_PROPERTY, "false"); + evaluate("subject-1", new Value(true)); + assertTelemetryCounts(0); + this.api.shutdown(); + start(options); + evaluate("subject-2", new Value(true)); + assertTelemetryCounts(1); + } + + @Test + void optedOutSubjectKeepsAssignmentAndExposureButEmitsNoEvaluationTelemetry() throws Exception { + start(); + + final FlagEvaluationDetails details = evaluate("subject-1", new Value(true)); + + assertNull(details.getErrorCode()); + assertEquals(1, details.getValue()); + assertEquals("on", details.getVariant()); + assertEquals(1, this.exposures.size()); + assertEquals(1, this.customerEvaluations.size()); + assertTelemetryCounts(0); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void telemetryAttributeRemainsAvailableForTargetingAndExposure(final boolean enabled) + throws Exception { + start(new Provider.Options().enableEvaluationTelemetryOptOutParameter(enabled)); + + for (final boolean observeFullEvaluationData : new boolean[] {false, true}) { + this.evaluator.accept(targetingConfiguration(observeFullEvaluationData)); + final String subject = "subject-" + observeFullEvaluationData; + final int previousEvents = this.events.size(); + final int previousExposures = this.exposures.size(); + + final FlagEvaluationDetails allowed = evaluate(subject, new Value(false)); + final FlagEvaluationDetails optedOut = evaluate(subject, new Value(true)); + final FlagEvaluationDetails repeated = evaluate(subject, new Value(true)); + final FlagEvaluationDetails stringValue = + evaluate(subject + "-string", new Value("true")); + + assertNull(allowed.getErrorCode()); + assertEquals(1, allowed.getValue()); + assertEquals("on", allowed.getVariant()); + for (final FlagEvaluationDetails details : asList(optedOut, repeated, stringValue)) { + assertNull(details.getErrorCode()); + assertEquals(2, details.getValue(), "the ordinary targeting rule sees the attribute"); + assertEquals("targeted", details.getVariant()); + } + assertEquals( + previousExposures + 3, this.exposures.size(), "normal exposure deduplication applies"); + assertEquals(false, exposureAttribute(previousExposures)); + assertEquals(true, exposureAttribute(previousExposures + 1)); + assertEquals("true", exposureAttribute(previousExposures + 2)); + assertTelemetryCounts(previousEvents + (enabled ? 2 : 4)); + + if (observeFullEvaluationData) { + assertEquals(false, this.events.get(previousEvents).attrs.get(PARAMETER)); + assertEquals("true", this.events.get(this.events.size() - 1).attrs.get(PARAMETER)); + if (!enabled) { + assertEquals(true, this.events.get(previousEvents + 1).attrs.get(PARAMETER)); + } + } else { + for (final FlagEvalEvent event : this.events.subList(previousEvents, this.events.size())) { + assertTrue(event.attrs.isEmpty(), "normal evaluation privacy rules still apply"); + } + } + } + } + + @Test + void allowedSubjectEmitsEachSignal() throws Exception { + start(); + + evaluate("subject-1", new Value(false)); + + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + assertEquals(1, this.customerEvaluations.size()); + } + + @ParameterizedTest + @MethodSource("nonOptOutValues") + void onlyBooleanTrueSuppressesTelemetry(final Value value) throws Exception { + start(); + + evaluate("subject-1", value); + + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + } + + static Stream nonOptOutValues() { + return Stream.of( + Arguments.of((Value) null), + Arguments.of(new Value()), + Arguments.of(new Value(false)), + Arguments.of(new Value("true")), + Arguments.of(new Value(1)), + Arguments.of(new Value(singletonList(new Value(true))))); + } + + @Test + void theParameterHasNoEffectUnlessEnabled() throws Exception { + start(new Provider.Options()); + + evaluate("subject-1", new Value(true)); + + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + } + + @Test + void theParameterIsCaseSensitiveAndDoesNotAcceptTheOldName() throws Exception { + start(); + + this.client.getIntegerDetails( + "flag", + 0, + new MutableContext("subject-1") + .add("dd_telemetry_opt_out", true) + .add("ddEvaluationTelemetryOptOut", true)); + + assertTelemetryCounts(1); + } + + @Test + void providerOptionsAreCapturedAtConstruction() throws Exception { + final Provider.Options options = + new Provider.Options().enableEvaluationTelemetryOptOutParameter(true); + start(options); + options.enableEvaluationTelemetryOptOutParameter(false); + + evaluate("subject-1", new Value(true)); + + assertTelemetryCounts(0); + } + + @Test + void eachEvaluationUsesItsOwnDecisionWithoutChangingExistingEvents() throws Exception { + start(); + + evaluate("same-subject", new Value(false)); + evaluate("same-subject", new Value(true)); + assertTelemetryCounts(1); + evaluate("same-subject", new Value(false)); + + assertTelemetryCounts(2); + assertEquals(1, this.exposures.size(), "exposure deduplication remains independent"); + assertEquals(3, this.customerEvaluations.size()); + } + + @Test + void missingAndWrongTypeFlagsKeepDefaultsWithoutEmittingTelemetry() throws Exception { + start(); + final EvaluationContext context = context("subject-1", new Value(true)); + + final FlagEvaluationDetails missing = + this.client.getIntegerDetails("missing", 42, context); + final FlagEvaluationDetails wrongType = + this.client.getStringDetails("flag", "default", context); + + assertEquals(ErrorCode.FLAG_NOT_FOUND, missing.getErrorCode()); + assertEquals(42, missing.getValue()); + assertEquals(ErrorCode.TYPE_MISMATCH, wrongType.getErrorCode()); + assertEquals("default", wrongType.getValue()); + assertTelemetryCounts(0); + assertEquals(0, this.exposures.size()); + assertEquals(2, this.customerEvaluations.size()); + + this.client.getIntegerDetails("missing", 42, context("subject-2", new Value(false))); + assertTelemetryCounts(1); + } + + @Test + void contextMergingHonorsThePerEvaluationOverride() throws Exception { + start(); + this.api.setEvaluationContext(new MutableContext().add(PARAMETER, true)); + + this.client.getIntegerDetails("flag", 0, new ImmutableContext("subject-1")); + evaluate("subject-2", new Value(false)); + + assertTelemetryCounts(1); + assertEquals("subject-2", this.events.get(0).targetingKey); + assertEquals(2, this.exposures.size()); + } + + @Test + void applicationBeforeHooksCanSupplyTheOptOutDecision() throws Exception { + start(); + this.api.addHooks( + new Hook() { + @Override + public Optional before( + final HookContext context, final Map hints) { + return Optional.of(new ImmutableContext(singletonMap(PARAMETER, new Value(true)))); + } + }); + + evaluate("subject-1", new Value(false)); + + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + } + + @Test + void applicationAfterHooksCannotUndoTheEvaluationDecision() throws Exception { + start(); + final MutableContext invocationContext = new MutableContext("subject-1").add(PARAMETER, true); + this.api.addHooks( + new Hook() { + @Override + public void after( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + invocationContext.add(PARAMETER, false); + } + }); + + final FlagEvaluationDetails details = + this.client.getIntegerDetails("flag", 0, invocationContext); + + assertNull(details.getErrorCode()); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + } + + @Test + void optingOutInAnAfterHookStopsTelemetryBeforeEmission() throws Exception { + start(); + final MutableContext invocationContext = new MutableContext("subject-1").add(PARAMETER, false); + this.api.addHooks( + new Hook() { + @Override + public void after( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + invocationContext.add(PARAMETER, true); + } + }); + + final FlagEvaluationDetails details = + this.client.getIntegerDetails("flag", 0, invocationContext); + + assertNull(details.getErrorCode()); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + } + + @Test + void aFailingBeforeHookDoesNotLeakErrorTelemetry() throws Exception { + start(); + final MutableContext invocationContext = new MutableContext("subject-1").add(PARAMETER, true); + this.api.addHooks( + new Hook() { + @Override + public Optional before( + final HookContext context, final Map hints) { + throw new IllegalStateException("application hook failed"); + } + + @Override + public void error( + final HookContext context, + final Exception error, + final Map hints) { + invocationContext.add(PARAMETER, false); + } + }); + + final FlagEvaluationDetails details = + this.client.getIntegerDetails("flag", 0, invocationContext); + + assertEquals(ErrorCode.GENERAL, details.getErrorCode()); + assertEquals(0, details.getValue()); + assertTelemetryCounts(0); + assertEquals(0, this.exposures.size()); + assertEquals(1, this.customerEvaluations.size()); + } + + @Test + void nestedEvaluationsKeepIndependentDecisions() throws Exception { + start(); + this.api.addHooks( + new Hook() { + @Override + public void after( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + if ("outer".equals(context.getCtx().getTargetingKey())) { + assertEquals(1, evaluate("inner", new Value(false)).getValue()); + } + } + }); + + assertEquals(1, evaluate("outer", new Value(true)).getValue()); + + assertTelemetryCounts(1); + assertEquals("inner", this.events.get(0).targetingKey); + assertEquals(2, this.exposures.size()); + assertEquals(2, this.customerEvaluations.size()); + } + + @Test + void concurrentSubjectsCannotChangeEachOthersDecision() throws Exception { + start(); + final CountDownLatch optedOutEvaluated = new CountDownLatch(1); + final CountDownLatch allowOptedOutToFinish = new CountDownLatch(1); + this.api.addHooks( + new Hook() { + @Override + public void after( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + if ("opted-out".equals(context.getCtx().getTargetingKey())) { + optedOutEvaluated.countDown(); + try { + assertTrue(allowOptedOutToFinish.await(10, SECONDS)); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } + } + }); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + final Future> pending = + executor.submit(() -> evaluate("opted-out", new Value(true))); + assertTrue(optedOutEvaluated.await(10, SECONDS)); + assertEquals(1, evaluate("allowed", new Value(false)).getValue()); + allowOptedOutToFinish.countDown(); + assertNull(pending.get(10, SECONDS).getErrorCode()); + + assertTelemetryCounts(1); + assertEquals("allowed", this.events.get(0).targetingKey); + assertEquals(2, this.exposures.size()); + } finally { + allowOptedOutToFinish.countDown(); + executor.shutdownNow(); + } + } + + @Test + void independentProvidersKeepTheirOwnActivationSetting() throws Exception { + start(); + this.api.setProviderAndWait( + "unrestricted", + provider(new Provider.Options(), new CapturingEvaluator(defaultConfiguration()))); + + evaluate("opted-out", new Value(true)); + this.api + .getClient("unrestricted") + .getIntegerDetails("flag", 0, context("unrestricted-subject", new Value(true))); + + assertTelemetryCounts(1); + assertEquals("unrestricted-subject", this.events.get(0).targetingKey); + assertEquals(2, this.exposures.size()); + } + + @Test + void optOutWorksInsideAMultiProvider() throws Exception { + start(); + this.api.setProviderAndWait( + new MultiProvider( + singletonList( + provider( + new Provider.Options().enableEvaluationTelemetryOptOutParameter(true), + new CapturingEvaluator(defaultConfiguration()))))); + + assertEquals(1, evaluate("opted-out", new Value(true)).getValue()); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + + assertEquals(1, evaluate("allowed", new Value(false)).getValue()); + assertTelemetryCounts(1); + assertEquals(2, this.exposures.size()); + } + + @Test + void missingConfigurationKeepsTheDefaultWithoutEmittingTelemetry() throws Exception { + start(); + this.evaluator.accept(null); + + final FlagEvaluationDetails details = evaluate("subject-1", new Value(true)); + + assertEquals(ErrorCode.PROVIDER_NOT_READY, details.getErrorCode()); + assertEquals(0, details.getValue()); + assertTelemetryCounts(0); + assertEquals(0, this.exposures.size()); + assertEquals(1, this.customerEvaluations.size()); + } + + private void start() throws Exception { + start(new Provider.Options().enableEvaluationTelemetryOptOutParameter(true)); + } + + private void start(final Provider.Options options) throws Exception { + this.evaluator = new CapturingEvaluator(defaultConfiguration()); + this.api.setProviderAndWait(provider(options, this.evaluator)); + this.client = this.api.getClient(); + } + + /** + * Creates a provider with span enrichment on, recording span enrichment to {@link #spanEnricher}. + */ + private Provider provider(final Provider.Options options, final DDEvaluator evaluator) + throws ReflectiveOperationException { + final Provider provider = new Provider(options, evaluator, Boolean.TRUE); + // No seam injects a span enricher: the provider takes it from the detected agent bridge. + final Field enricher = SpanEnrichmentHook.class.getDeclaredField("enricher"); + enricher.setAccessible(true); + enricher.set(provider.spanEnrichmentHook(), this.spanEnricher); + return provider; + } + + /** Creates a runtime only holding a flag evaluation pipeline, without starting it. */ + private static FeatureFlagsRuntime capturingRuntime(final FlagEvaluationPipeline evaluations) + throws ReflectiveOperationException { + final Constructor constructor = + FeatureFlagsRuntime.class.getDeclaredConstructor( + ConfigurationService.class, ExposurePipeline.class, FlagEvaluationPipeline.class); + constructor.setAccessible(true); + return constructor.newInstance(mock(ConfigurationService.class), null, evaluations); + } + + private static ServerConfiguration defaultConfiguration() { + final Allocation allocation = + new Allocation( + "allocation", + null, + null, + null, + singletonList(new Split(emptyList(), "on", emptyMap(), 7)), + true); + final Flag flag = + new Flag( + "flag", + true, + ValueType.INTEGER, + singletonMap("on", new Variant("on", 1)), + singletonList(allocation)); + return new ServerConfiguration("", "", false, null, singletonMap("flag", flag)); + } + + private static ServerConfiguration targetingConfiguration( + final boolean observeFullEvaluationData) { + final Rule rule = + new Rule( + singletonList( + new ConditionConfiguration( + ConditionOperator.ONE_OF, PARAMETER, singletonList(true)))); + final Allocation targeted = + new Allocation( + "targeted", + singletonList(rule), + null, + null, + singletonList(new Split(emptyList(), "targeted", emptyMap(), 7)), + true); + final Allocation fallback = + new Allocation( + "fallback", + null, + null, + null, + singletonList(new Split(emptyList(), "on", emptyMap(), 7)), + true); + final Map variants = new HashMap<>(); + variants.put("on", new Variant("on", 1)); + variants.put("targeted", new Variant("targeted", 2)); + final Flag flag = + new Flag("flag", true, ValueType.INTEGER, variants, asList(targeted, fallback)); + return new ServerConfiguration( + "", "", observeFullEvaluationData, null, singletonMap("flag", flag)); + } + + private FlagEvaluationDetails evaluate(final String subject, final Value optOut) { + return this.client.getIntegerDetails("flag", 0, context(subject, optOut)); + } + + private static EvaluationContext context(final String subject, final Value optOut) { + return new MutableContext( + subject, optOut == null ? emptyMap() : singletonMap(PARAMETER, optOut)); + } + + private Object exposureAttribute(final int index) { + return this.exposures.get(index).subject.attributes.get(PARAMETER); + } + + private void assertTelemetryCounts(final int expected) { + assertAll( + () -> assertEquals(expected, this.events.size(), "evaluation EVP enqueue count"), + () -> assertEquals(expected, this.metrics.get(), "OTel evaluation count"), + () -> assertEquals(expected, this.spans.size(), "automatic span enrichment count")); + } + + /** + * A Datadog evaluator holding a configuration, sharing the test runtime for flag evaluation + * events, and the test exposure cache and capture for exposures. + */ + private final class CapturingEvaluator extends DDEvaluator { + CapturingEvaluator(final ServerConfiguration configuration) { + super( + () -> {}, + AgentBridge.NONE, + TestSettings.of("experimental.flagging.provider.span.enrichment.enabled", "true")); + accept(configuration); + } + + @Override + public boolean initialize( + final long timeout, final TimeUnit unit, final EvaluationContext context) { + return hasConfiguration(); + } + + @Override + FeatureFlagsRuntime runtime() { + return EvaluationTelemetryOptOutTest.this.runtime; + } + + @Override + ExposureDeduplicationCache exposureCache() { + return EvaluationTelemetryOptOutTest.this.exposureCache; + } + + @Override + boolean isExposureDeliveryEnabled() { + return true; + } + + @Override + void recordExposure(final ExposureEvent event) { + EvaluationTelemetryOptOutTest.this.exposures.add(event); + } + } + + /** A flag evaluation pipeline capturing enqueued events, never started. */ + private static final class CapturingPipeline extends FlagEvaluationPipeline { + private final List events; + + CapturingPipeline(final List events) { + super( + 1, 1, SECONDS, (route, json) -> {}, emptyMap(), new RuntimeServices(HealthMetrics.NOOP)); + this.events = events; + } + + @Override + public boolean hasCapacityForEnqueue() { + return true; + } + + @Override + public void enqueue(final FlagEvalEvent event) { + this.events.add(event); + } + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java index 1cfd0a827c5..7af245430ba 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java @@ -106,6 +106,37 @@ void readsTheExposuresDatadogLoggingToggle(final String value, final boolean ena .isExposuresDatadogLoggingEnabled()); } + @TableTest({ + "scenario | agent | property | environment | enabled", + "default | | | | false ", + "environment enables | | | 'true' | true ", + "environment disabled | | | 'false' | false ", + "invalid environment | | | 'invalid' | false ", + "blank environment | | | '' | false ", + "property overrides environment | | 'false' | 'true' | false ", + "property enables | | 'true' | 'false' | true ", + "blank property falls back | | ' ' | 'true' | true ", + "agent overrides property | 'false' | 'true' | 'true' | false ", + "agent enables | 'true' | 'false' | 'false' | true " + }) + void readsTheEvaluationTelemetryPerSubjectOptOutToggle( + final String agent, final String property, final String environment, final boolean enabled) { + final Map bridge = new HashMap<>(); + putIfNotNull(bridge, "feature.flags.evaluation.telemetry.per.subject.opt.out.enabled", agent); + final Map properties = new HashMap<>(); + putIfNotNull( + properties, "dd.feature.flags.evaluation.telemetry.per.subject.opt.out.enabled", property); + final Map variables = new HashMap<>(); + putIfNotNull( + variables, + "DD_FEATURE_FLAGS_EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED", + environment); + final Settings settings = + new Settings(new SettingAgentBridge(bridge), properties::get, variables::get); + + assertEquals(enabled, settings.isEvaluationTelemetryPerSubjectOptOutEnabled()); + } + @TableTest({ "scenario | value | interval", "valid | '10' | 10 ", From 21d760077e2a67c8da13595cfca86b6b29cf8ed3 Mon Sep 17 00:00:00 2001 From: Bruce Bujon Date: Tue, 20 Oct 2026 13:03:54 +0200 Subject: [PATCH 2/7] feat(ffe): Send each Feature Flags event batch once Port the SDK part of master #12477: direct intake no longer retries, and the event proxy fallback switches to direct intake for good on the first proxy failure, replaying the batch only when the proxy definitively did not receive it. Co-Authored-By: Claude Opus 5.5 --- .../delivery/DirectIntakeTransport.java | 72 ++--------- .../internal/delivery/FallbackTransport.java | 17 ++- .../delivery/DirectIntakeTransportTest.java | 119 ++++-------------- .../delivery/FallbackTransportTest.java | 8 +- 4 files changed, 48 insertions(+), 168 deletions(-) diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java index 9c004fd54c6..bd2c0171a6b 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java @@ -6,40 +6,33 @@ import com.datadog.openfeature.internal.bridge.EventTransport; import com.datadog.openfeature.internal.config.Settings; import java.io.IOException; -import java.net.ConnectException; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; -import java.net.http.HttpTimeoutException; import java.time.Duration; import java.util.concurrent.ThreadLocalRandom; -import java.util.concurrent.TimeUnit; import javax.annotation.Nullable; -/** Posts events directly to the Datadog event platform intake, authenticated with an API key. */ +/** + * Posts events directly to the Datadog event platform intake, authenticated with an API key. Each + * batch is sent once, without retries or redirects, so an ambiguously accepted batch is never sent + * twice. + */ public final class DirectIntakeTransport implements EventTransport { - static final int MAX_RETRIES = 5; - private static final long INITIAL_RETRY_DELAY_MILLIS = 100; private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(10); private final URI intake; private final String apiKey; private final String traceId; private final HttpClient httpClient; - private final Retries.Sleeper retrySleeper; - DirectIntakeTransport( - final URI intake, - final String apiKey, - final HttpClient httpClient, - final Retries.Sleeper retrySleeper) { + DirectIntakeTransport(final URI intake, final String apiKey, final HttpClient httpClient) { this.intake = intake; this.apiKey = apiKey; // Intake requests are attributed to a stable, random trace id. this.traceId = Long.toUnsignedString(ThreadLocalRandom.current().nextLong() >>> 1); this.httpClient = httpClient; - this.retrySleeper = retrySleeper; } /** @@ -58,8 +51,7 @@ public static DirectIntakeTransport create(final Settings settings) { return new DirectIntakeTransport( URI.create("https://" + SiteHost.of("event-platform-intake", settings.site()) + "/api/v2/"), apiKey, - HttpClient.newBuilder().connectTimeout(REQUEST_TIMEOUT).followRedirects(NEVER).build(), - TimeUnit.MILLISECONDS::sleep); + HttpClient.newBuilder().connectTimeout(REQUEST_TIMEOUT).followRedirects(NEVER).build()); } @Override @@ -76,14 +68,8 @@ public void post(final String route, final byte[] json) throws IOException { final int status; try { status = - Retries.send( - () -> - new HttpCall<>( - this.httpClient, request, HttpResponse.BodyHandlers.discarding()), - new IntakeRetryPolicy(), - MAX_RETRIES + 1, - this.retrySleeper, - () -> false) + new HttpCall<>(this.httpClient, request, HttpResponse.BodyHandlers.discarding()) + .execute() .statusCode(); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); @@ -93,44 +79,4 @@ public void post(final String route, final byte[] json) throws IOException { throw new IOException("Request to " + route + " returned error response " + status); } } - - private static final class IntakeRetryPolicy implements Retries.Policy { - private boolean rateLimitRetry; - - @Override - public boolean shouldRetry(final IOException error) { - return error instanceof ConnectException || error instanceof HttpTimeoutException; - } - - @Override - public long retryDelayMillis(@Nullable final HttpResponse response, final int attempt) { - if (this.rateLimitRetry) { - return -1; - } - if (response != null) { - final int status = response.statusCode(); - if (status == 429) { - final String reset = response.headers().firstValue("X-RateLimit-Reset").orElse(null); - if (reset != null) { - try { - final long seconds = Long.parseLong(reset); - if (seconds > 10) { - return -1; - } - if (seconds >= 0) { - this.rateLimitRetry = true; - return TimeUnit.SECONDS.toMillis(seconds) - + ThreadLocalRandom.current().nextInt(401); - } - } catch (final NumberFormatException ignored) { - // An invalid reset header falls back to exponential backoff. - } - } - } else if (status < 500 || status > 599) { - return -1; - } - } - return INITIAL_RETRY_DELAY_MILLIS << (attempt - 1); - } - } } diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/FallbackTransport.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/FallbackTransport.java index e01bf019672..3c4ca077f0b 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/FallbackTransport.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/FallbackTransport.java @@ -7,8 +7,9 @@ import org.slf4j.LoggerFactory; /** - * Posts through the event proxy, switching to direct intake for good once the proxy is definitively - * unavailable. Ambiguous proxy failures are not replayed through direct intake. + * Posts through the event proxy, switching to direct intake for good on the first proxy failure. + * Only a definitively rejected batch is replayed through direct intake, an ambiguously failed batch + * is not to avoid sending it twice. */ public final class FallbackTransport implements EventTransport { private static final Logger LOGGER = LoggerFactory.getLogger(FallbackTransport.class); @@ -29,11 +30,17 @@ public void post(final String route, final byte[] json) throws IOException { this.proxy.post(route, json); return; } catch (final EventProxyUnavailableException e) { - LOGGER.debug( - "Switching Feature Flags event delivery from the event proxy to direct intake"); - this.proxyUnavailable = true; + switchToDirectIntake(); + } catch (final IOException e) { + switchToDirectIntake(); + throw e; } } this.direct.post(route, json); } + + private void switchToDirectIntake() { + LOGGER.debug("Switching Feature Flags event delivery from the event proxy to direct intake"); + this.proxyUnavailable = true; + } } diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java index 8ff26333df8..ed3f92e35cc 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java @@ -5,26 +5,21 @@ 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.ArgumentMatchers.any; import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import com.datadog.openfeature.internal.LocalHttpServer; import com.datadog.openfeature.internal.config.TestSettings; import java.io.IOException; import java.io.InterruptedIOException; import java.net.ConnectException; -import java.net.InetAddress; -import java.net.ServerSocket; -import java.net.URI; import java.net.UnknownHostException; import java.net.http.HttpClient; import java.net.http.HttpTimeoutException; -import java.util.List; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CopyOnWriteArrayList; import javax.net.ssl.SSLException; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -34,7 +29,6 @@ public class DirectIntakeTransportTest { private LocalHttpServer server; - private final List sleeps = new CopyOnWriteArrayList<>(); @BeforeEach void setUp() throws IOException { @@ -83,26 +77,14 @@ void postsJsonWithApiKeyAndTraceHeaders() throws Exception { "scenario | status", "server error | 500 ", "unavailable | 503 ", - "too many requests | 429 " + "too many requests | 429 ", + "bad request | 400 ", + "forbidden | 403 ", + "not found | 404 ", + "timeout | 408 ", + "redirect | 302 " }) - void retriesRetryableStatus(final int status) throws Exception { - this.server.enqueue(status, null).enqueue(status, null).enqueue(200, null); - - transport().post("flagevaluation", new byte[] {'{', '}'}); - - assertEquals(3, this.server.requests().size()); - assertEquals(List.of(100L, 200L), this.sleeps); - } - - @TableTest({ - "scenario | status", - "bad request | 400 ", - "forbidden | 403 ", - "not found | 404 ", - "timeout | 408 ", - "redirect | 302 " - }) - void doesNotRetryOtherFailures(final int status) { + void sendsFailedBatchesOnce(final int status) { this.server.otherwise(status, null); assertThrows(IOException.class, () -> transport().post("exposures", new byte[] {'{', '}'})); @@ -110,72 +92,28 @@ void doesNotRetryOtherFailures(final int status) { assertEquals(1, this.server.requests().size()); } - @Test - void givesUpAfterMaxRetries() { - this.server.otherwise(503, null); - - assertThrows(IOException.class, () -> transport().post("exposures", new byte[] {'{', '}'})); - - assertEquals(DirectIntakeTransport.MAX_RETRIES + 1, this.server.requests().size()); - assertEquals(DirectIntakeTransport.MAX_RETRIES, this.sleeps.size()); - } - - @TableTest({ - "scenario | reset | requests | minDelay | maxDelay", - "immediate | 0 | 2 | 0 | 400 ", - "one second | 1 | 2 | 1000 | 1400 ", - "maximum wait | 10 | 2 | 10000 | 10400 ", - "too long | 11 | 1 | 0 | 0 ", - "overflow | 9223372036854775807 | 1 | 0 | 0 ", - "invalid | invalid | 3 | 100 | 100 ", - "negative | -2 | 3 | 100 | 100 ", - "empty | '' | 3 | 100 | 100 " - }) - void honorsRateLimitReset( - final String reset, final int requests, final long minDelay, final long maxDelay) - throws Exception { - this.server - .enqueue(429, null, "X-RateLimit-Reset", reset) - .enqueue(503, null) - .enqueue(200, null); - - if (requests == 3) { - transport().post("exposures", new byte[] {'{', '}'}); - } else { - assertThrows(IOException.class, () -> transport().post("exposures", new byte[] {'{', '}'})); - } - - assertEquals(requests, this.server.requests().size()); - assertEquals(requests - 1, this.sleeps.size()); - if (!this.sleeps.isEmpty()) { - assertTrue(this.sleeps.get(0) >= minDelay && this.sleeps.get(0) <= maxDelay); - } - } - @TableTest({ - "scenario | failure | attempts", - "connection | connection | 6 ", - "timeout | timeout | 6 ", - "TLS failure | tls | 1 ", - "DNS failure | dns | 1 ", - "interrupted IO | interrupted | 1 ", - "other IO | other | 1 " + "scenario | failure ", + "connection | connection ", + "timeout | timeout ", + "TLS failure | tls ", + "DNS failure | dns ", + "interrupted IO | interrupted", + "other IO | other " }) - void retriesOnlyConnectionAndTimeoutFailures(final IOException failure, final int attempts) - throws Exception { + void doesNotRetryIoFailures(final IOException failure) { final HttpClient client = mock(HttpClient.class); doReturn(CompletableFuture.failedFuture(failure)).when(client).sendAsync(any(), any()); - doThrow(failure).when(client).send(any(), any()); final DirectIntakeTransport transport = - new DirectIntakeTransport(this.server.uri("/api/v2/"), "secret", client, this.sleeps::add); + new DirectIntakeTransport(this.server.uri("/api/v2/"), "secret", client); assertThrows(failure.getClass(), () -> transport.post("exposures", new byte[] {'{', '}'})); - assertEquals(attempts - 1, this.sleeps.size()); + verify(client, times(1)).sendAsync(any(), any()); } /** - * Creates an I/O failure for a retry scenario. + * Creates an I/O failure for a send scenario. * * @param value the failure kind. * @return the failure. @@ -198,23 +136,8 @@ public static IOException failure(final String value) { } } - @Test - void retriesConnectionFailures() throws Exception { - final URI unreachable; - try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { - unreachable = URI.create("http://127.0.0.1:" + socket.getLocalPort() + "/api/v2/"); - } - final DirectIntakeTransport transport = - new DirectIntakeTransport( - unreachable, "secret", HttpClient.newHttpClient(), this.sleeps::add); - - assertThrows(IOException.class, () -> transport.post("exposures", new byte[] {'{', '}'})); - - assertEquals(DirectIntakeTransport.MAX_RETRIES, this.sleeps.size()); - } - private DirectIntakeTransport transport() { return new DirectIntakeTransport( - this.server.uri("/api/v2/"), "secret", HttpClient.newHttpClient(), this.sleeps::add); + this.server.uri("/api/v2/"), "secret", HttpClient.newHttpClient()); } } diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/FallbackTransportTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/FallbackTransportTest.java index ce05e06d685..585236831f4 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/FallbackTransportTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/FallbackTransportTest.java @@ -42,11 +42,15 @@ void switchesToDirectIntakeForGoodWhenTheProxyIsUnavailable() throws IOException } @Test - void doesNotReplayAmbiguousProxyFailures() throws IOException { + void switchesToDirectIntakeForGoodWithoutReplayingAmbiguousProxyFailures() throws IOException { doThrow(new IOException("timeout")).when(this.proxy).post("exposures", PAYLOAD); assertThrows(IOException.class, () -> this.transport.post("exposures", PAYLOAD)); - verify(this.direct, never()).post("exposures", PAYLOAD); + + this.transport.post("exposures", PAYLOAD); + + verify(this.proxy, times(1)).post("exposures", PAYLOAD); + verify(this.direct, times(1)).post("exposures", PAYLOAD); } } From 8b92504b6fa24e7d284f5da615a7c18d60d74282 Mon Sep 17 00:00:00 2001 From: Bruce Bujon Date: Tue, 20 Oct 2026 13:03:55 +0200 Subject: [PATCH 3/7] feat(ffe): Flush pending Feature Flags events at application shutdown The runtime now stops at application shutdown: through the new optional ShutdownSource bridge component when the Datadog Java agent provides it, so the agent flushes the SDK before the JVM shutdown hooks run, or from a JVM shutdown hook otherwise. Co-Authored-By: Claude Opus 5.5 --- .../internal/FeatureFlagsRuntime.java | 58 ++++++++++++++- .../internal/bridge/AgentBridge.java | 9 +++ .../internal/bridge/ShutdownSource.java | 12 ++++ .../internal/FeatureFlagsRuntimeTest.java | 70 +++++++++++++++++++ 4 files changed, 148 insertions(+), 1 deletion(-) create mode 100644 dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ShutdownSource.java diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java index 89a9b0fd4b1..c0f987e90bd 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java @@ -5,6 +5,7 @@ import com.datadog.openfeature.internal.bridge.AgentBridge; import com.datadog.openfeature.internal.bridge.ConfigurationSource; import com.datadog.openfeature.internal.bridge.EventTransport; +import com.datadog.openfeature.internal.bridge.ShutdownSource; import com.datadog.openfeature.internal.config.Settings; import com.datadog.openfeature.internal.delivery.CdnConfigurationSource; import com.datadog.openfeature.internal.delivery.DirectIntakeTransport; @@ -20,6 +21,7 @@ import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import javax.annotation.Nullable; import org.slf4j.Logger; @@ -27,7 +29,9 @@ /** * The Feature Flags runtime: configuration delivery and event pipelines. A single runtime is shared - * by all the providers of a class loader, and stopped when the last one releases it. + * by all the providers of a class loader, and stopped when the last one releases it, or when the + * application shuts down: by the Datadog Java agent before the JVM shutdown hooks run when + * attached, or by a JVM shutdown hook otherwise. */ public final class FeatureFlagsRuntime { private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlagsRuntime.class); @@ -42,6 +46,8 @@ public final class FeatureFlagsRuntime { @Nullable private final FlagEvaluationPipeline evaluations; private final List> listeners = new CopyOnWriteArrayList<>(); private volatile ServerConfiguration configuration; + private final AtomicBoolean stopped = new AtomicBoolean(); + @Nullable private volatile AutoCloseable shutdownRegistration; FeatureFlagsRuntime( final ConfigurationService source, @@ -66,6 +72,7 @@ public static FeatureFlagsRuntime acquire(final AgentBridge bridge, final Settin if (shared == null) { final FeatureFlagsRuntime created = create(bridge, settings); created.start(); + created.registerShutdown(bridge); shared = created; } references++; @@ -172,7 +179,56 @@ void start() { } } + /** + * Stops the runtime at application shutdown: through the agent when it supports it, otherwise + * through a JVM shutdown hook. + * + * @param bridge the bridge to the Datadog Java agent. + */ + void registerShutdown(final AgentBridge bridge) { + AutoCloseable registration = null; + try { + final ShutdownSource shutdown = bridge.shutdown(); + if (shutdown != null) { + registration = shutdown.subscribe(this::stop); + } + } catch (final RuntimeException | LinkageError e) { + LOGGER.debug("Could not register the Feature Flags shutdown with the Datadog Java agent", e); + } + if (registration == null) { + registration = addShutdownHook(); + } + this.shutdownRegistration = registration; + } + + @Nullable + private AutoCloseable addShutdownHook() { + final Thread hook = new Thread(this::stop, "dd-openfeature-shutdown"); + try { + Runtime.getRuntime().addShutdownHook(hook); + } catch (final IllegalStateException | SecurityException e) { + LOGGER.debug("Could not register the Feature Flags shutdown hook", e); + return null; + } + return () -> { + if (Thread.currentThread() != hook) { + Runtime.getRuntime().removeShutdownHook(hook); + } + }; + } + void stop() { + if (!this.stopped.compareAndSet(false, true)) { + return; + } + final AutoCloseable registration = this.shutdownRegistration; + if (registration != null) { + try { + registration.close(); + } catch (final Exception e) { + // The JVM is already shutting down, or the agent backend is gone. + } + } this.source.close(); if (this.evaluations != null) { this.evaluations.close(); diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java index 1ff84479cec..98dd74a96dd 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java @@ -38,4 +38,13 @@ public interface AgentBridge { * @return the span enricher. */ SpanEnricher spanEnricher(); + + /** + * @return the agent shutdown notifications, sent before the JVM shutdown hooks run, or {@code + * null} if not available, so the SDK must flush from its own JVM shutdown hook. + */ + @Nullable + default ShutdownSource shutdown() { + return null; + } } diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ShutdownSource.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ShutdownSource.java new file mode 100644 index 00000000000..145bf413b7b --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ShutdownSource.java @@ -0,0 +1,12 @@ +package com.datadog.openfeature.internal.bridge; + +/** A source of application shutdown notifications. */ +public interface ShutdownSource { + /** + * Subscribes to the application shutdown. + * + * @param listener the listener run at shutdown, typically flushing the pending events. + * @return the handle to close to unsubscribe. + */ + AutoCloseable subscribe(Runnable listener); +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java index 2d719897fce..9b9df1cd14f 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java @@ -15,6 +15,7 @@ import com.datadog.openfeature.internal.bridge.ConfigurationSource; import com.datadog.openfeature.internal.bridge.EventTransport; import com.datadog.openfeature.internal.bridge.HealthMetrics; +import com.datadog.openfeature.internal.bridge.ShutdownSource; import com.datadog.openfeature.internal.bridge.SpanEnricher; import com.datadog.openfeature.internal.config.TestSettings; import com.datadog.openfeature.internal.ufc.ServerConfiguration; @@ -239,6 +240,46 @@ void lastReleaseFlushesPendingExposures() { assertEquals(1, ((List) requests.get(0).get("exposures")).size()); } + @Test + void agentShutdownFlushesPendingExposuresOnce() { + final TestConfigurationSource source = new TestConfigurationSource(); + final List> requests = new CopyOnWriteArrayList<>(); + final TestAgentBridge bridge = + new TestAgentBridge(source, (route, json) -> requests.add(JsonReading.readObject(json))) + .withAgentShutdown(); + final FeatureFlagsRuntime runtime = + FeatureFlagsRuntime.acquire( + bridge, + TestSettings.of( + "feature.flags.configuration.source", "remote_config", + "flagging.evaluation.counts.enabled", "false")); + try { + assertNotNull(bridge.shutdownFlush, "the runtime registers its flush with the agent"); + runtime.recordExposure(exposure()); + + bridge.shutdownFlush.run(); + + assertTrue(source.closed); + assertTrue(bridge.shutdownUnregistered); + assertEquals(1, requests.size()); + } finally { + FeatureFlagsRuntime.release(); + } + assertEquals(1, requests.size(), "a release after the agent shutdown does not flush again"); + } + + @Test + void releaseUnregistersTheAgentShutdownFlush() { + final TestAgentBridge bridge = + new TestAgentBridge(new TestConfigurationSource(), null).withAgentShutdown(); + FeatureFlagsRuntime.acquire( + bridge, TestSettings.of("feature.flags.configuration.source", "remote_config")); + + FeatureFlagsRuntime.release(); + + assertTrue(bridge.shutdownUnregistered); + } + @Test void failedStartDoesNotCacheTheSharedRuntime() { final ConfigurationSource failing = @@ -280,6 +321,16 @@ void recordsExposuresOnlyWhenDeliveryIsEnabled() { null)); } + private static com.datadog.openfeature.internal.exposure.ExposureEvent exposure() { + return new com.datadog.openfeature.internal.exposure.ExposureEvent( + 1L, + new com.datadog.openfeature.internal.exposure.Allocation("a"), + new com.datadog.openfeature.internal.exposure.Flag("f"), + new com.datadog.openfeature.internal.exposure.Variant("v"), + new com.datadog.openfeature.internal.exposure.Subject("s", emptyMap()), + null); + } + private static final class TestConfigurationSource implements ConfigurationSource { volatile Consumer listener; volatile boolean closed; @@ -294,6 +345,9 @@ public AutoCloseable subscribe(final Consumer listener) { private static final class TestAgentBridge implements AgentBridge { private final ConfigurationSource remoteConfiguration; private final EventTransport eventProxy; + private boolean agentShutdown; + volatile Runnable shutdownFlush; + volatile boolean shutdownUnregistered; TestAgentBridge( final ConfigurationSource remoteConfiguration, final EventTransport eventProxy) { @@ -325,5 +379,21 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Override + public ShutdownSource shutdown() { + if (!this.agentShutdown) { + return AgentBridge.super.shutdown(); + } + return listener -> { + this.shutdownFlush = listener; + return () -> this.shutdownUnregistered = true; + }; + } + + TestAgentBridge withAgentShutdown() { + this.agentShutdown = true; + return this; + } } } From 44d04095541324afd1e9922cd1b524682e8f07dd Mon Sep 17 00:00:00 2001 From: Bruce Bujon Date: Tue, 20 Oct 2026 13:33:37 +0200 Subject: [PATCH 4/7] refactor(ffe): Make the bridge shutdown component abstract like the others No released agent implements the bridge yet, so the shutdown component needs no default method for compatibility. The no-op bridge returns null, like the other optional components. Co-Authored-By: Claude Opus 5.5 --- .../internal/FeatureFlagsRuntime.java | 16 +++------------- .../openfeature/internal/bridge/AgentBridge.java | 4 +--- .../internal/bridge/NoopAgentBridge.java | 6 ++++++ .../openfeature/DatadogExposureLoggingTest.java | 6 ++++++ .../com/datadog/openfeature/ProviderTest.java | 6 ++++++ .../internal/FeatureFlagsRuntimeTest.java | 2 +- .../internal/config/SettingsTest.java | 6 ++++++ 7 files changed, 29 insertions(+), 17 deletions(-) diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java index c0f987e90bd..91afc21875a 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java @@ -186,19 +186,9 @@ void start() { * @param bridge the bridge to the Datadog Java agent. */ void registerShutdown(final AgentBridge bridge) { - AutoCloseable registration = null; - try { - final ShutdownSource shutdown = bridge.shutdown(); - if (shutdown != null) { - registration = shutdown.subscribe(this::stop); - } - } catch (final RuntimeException | LinkageError e) { - LOGGER.debug("Could not register the Feature Flags shutdown with the Datadog Java agent", e); - } - if (registration == null) { - registration = addShutdownHook(); - } - this.shutdownRegistration = registration; + final ShutdownSource shutdown = bridge.shutdown(); + this.shutdownRegistration = + shutdown == null ? addShutdownHook() : shutdown.subscribe(this::stop); } @Nullable diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java index 98dd74a96dd..342c2c2bc8c 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java @@ -44,7 +44,5 @@ public interface AgentBridge { * null} if not available, so the SDK must flush from its own JVM shutdown hook. */ @Nullable - default ShutdownSource shutdown() { - return null; - } + ShutdownSource shutdown(); } diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/NoopAgentBridge.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/NoopAgentBridge.java index dab3e42db97..0a8ff1709ae 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/NoopAgentBridge.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/NoopAgentBridge.java @@ -30,4 +30,10 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Nullable + @Override + public ShutdownSource shutdown() { + return null; + } } diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/DatadogExposureLoggingTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/DatadogExposureLoggingTest.java index 846b4bc31cc..7f75c8ab611 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/DatadogExposureLoggingTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/DatadogExposureLoggingTest.java @@ -14,6 +14,7 @@ import com.datadog.openfeature.internal.bridge.ConfigurationSource; import com.datadog.openfeature.internal.bridge.EventTransport; import com.datadog.openfeature.internal.bridge.HealthMetrics; +import com.datadog.openfeature.internal.bridge.ShutdownSource; import com.datadog.openfeature.internal.bridge.SpanEnricher; import com.datadog.openfeature.internal.config.TestSettings; import com.datadog.openfeature.internal.exposure.ExposureEvent; @@ -150,5 +151,10 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Override + public ShutdownSource shutdown() { + return null; + } } } diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java index 9f08140abd0..65184e17ee9 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java @@ -25,6 +25,7 @@ import com.datadog.openfeature.internal.bridge.ConfigurationSource; import com.datadog.openfeature.internal.bridge.EventTransport; import com.datadog.openfeature.internal.bridge.HealthMetrics; +import com.datadog.openfeature.internal.bridge.ShutdownSource; import com.datadog.openfeature.internal.bridge.SpanEnricher; import com.datadog.openfeature.internal.config.TestSettings; import com.datadog.openfeature.internal.ufc.Allocation; @@ -609,5 +610,10 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Override + public ShutdownSource shutdown() { + return null; + } } } diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java index 9b9df1cd14f..e8935ff5a92 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java @@ -383,7 +383,7 @@ public SpanEnricher spanEnricher() { @Override public ShutdownSource shutdown() { if (!this.agentShutdown) { - return AgentBridge.super.shutdown(); + return null; } return listener -> { this.shutdownFlush = listener; diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java index 7af245430ba..80c3a9258b3 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java @@ -9,6 +9,7 @@ import com.datadog.openfeature.internal.bridge.ConfigurationSource; import com.datadog.openfeature.internal.bridge.EventTransport; import com.datadog.openfeature.internal.bridge.HealthMetrics; +import com.datadog.openfeature.internal.bridge.ShutdownSource; import com.datadog.openfeature.internal.bridge.SpanEnricher; import java.util.HashMap; import java.util.Map; @@ -230,5 +231,10 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Override + public ShutdownSource shutdown() { + return null; + } } } From ddbc4e648f7eabd4d4e1ae8347829821b4cfbeee Mon Sep 17 00:00:00 2001 From: Bruce Bujon Date: Tue, 20 Oct 2026 14:07:11 +0200 Subject: [PATCH 5/7] fix(ffe): Share one shutdown budget between the Feature Flags event pipelines The runtime closed the flag evaluation pipeline, waiting up to 5 seconds, before even signaling the exposure pipeline, then waited up to another 800 ms: a stalled final POST could hold the JVM exit for almost 6 seconds and starve the exposures flush. Port the two-phase shutdown of the agent writers: signal both pipelines, then await both against one 800 ms deadline, without interrupting a current or final HTTP request. Co-Authored-By: Claude Opus 5.5 --- .../openfeature/internal/EventPipeline.java | 56 +++++++++++++++++++ .../internal/FeatureFlagsRuntime.java | 12 +++- .../internal/exposure/ExposurePipeline.java | 23 ++++---- .../FlagEvaluationPipeline.java | 48 +++++++--------- .../internal/FeatureFlagsRuntimeTest.java | 39 +++++++++++++ 5 files changed, 136 insertions(+), 42 deletions(-) create mode 100644 dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventPipeline.java diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventPipeline.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventPipeline.java new file mode 100644 index 00000000000..dba55339c80 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventPipeline.java @@ -0,0 +1,56 @@ +package com.datadog.openfeature.internal; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +/** + * An event pipeline stopping in two phases, so that several pipelines can share one shutdown + * budget: every pipeline is first asked to flush, then each one is awaited until the same deadline. + */ +public interface EventPipeline extends AutoCloseable { + /** The time the pipelines get to flush their pending events when stopping. */ + long SHUTDOWN_TIMEOUT_MILLIS = 800; + + /** Stops accepting events and asks the worker to flush the pending ones, without waiting. */ + void requestShutdown(); + + /** + * Waits for the worker to flush the pending events and exit. + * + * @param deadlineNanos the {@link System#nanoTime()} value to wait until. + */ + void awaitTermination(long deadlineNanos); + + /** Stops the pipeline, waiting at most {@link #SHUTDOWN_TIMEOUT_MILLIS} for the final flush. */ + @Override + default void close() { + final long deadline = shutdownDeadline(); + requestShutdown(); + awaitTermination(deadline); + } + + /** + * @return the deadline of a shutdown starting now. + */ + static long shutdownDeadline() { + return System.nanoTime() + MILLISECONDS.toNanos(SHUTDOWN_TIMEOUT_MILLIS); + } + + /** + * Waits for a worker thread to exit. + * + * @param worker the worker thread. + * @param deadlineNanos the {@link System#nanoTime()} value to wait until. + */ + static void awaitThread(final Thread worker, final long deadlineNanos) { + final long remaining = deadlineNanos - System.nanoTime(); + if (worker == Thread.currentThread() || remaining <= 0) { + return; + } + try { + NANOSECONDS.timedJoin(worker, remaining); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java index 91afc21875a..b700a9f4c45 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java @@ -219,12 +219,20 @@ void stop() { // The JVM is already shutting down, or the agent backend is gone. } } + // Signal both pipelines before waiting for either, so they flush in parallel within one budget. + final long deadline = EventPipeline.shutdownDeadline(); + if (this.evaluations != null) { + this.evaluations.requestShutdown(); + } + if (this.exposures != null) { + this.exposures.requestShutdown(); + } this.source.close(); if (this.evaluations != null) { - this.evaluations.close(); + this.evaluations.awaitTermination(deadline); } if (this.exposures != null) { - this.exposures.close(); + this.exposures.awaitTermination(deadline); } } diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/exposure/ExposurePipeline.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/exposure/ExposurePipeline.java index 8397b208647..9577b1df211 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/exposure/ExposurePipeline.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/exposure/ExposurePipeline.java @@ -1,5 +1,8 @@ package com.datadog.openfeature.internal.exposure; +import static com.datadog.openfeature.internal.EventPipeline.awaitThread; + +import com.datadog.openfeature.internal.EventPipeline; import com.datadog.openfeature.internal.EventQueue; import com.datadog.openfeature.internal.RuntimeServices; import com.datadog.openfeature.internal.bridge.EventTransport; @@ -13,14 +16,13 @@ import org.slf4j.LoggerFactory; /** Batches exposure events to the event platform {@code exposures} route. */ -public class ExposurePipeline implements Consumer, AutoCloseable { +public class ExposurePipeline implements Consumer, EventPipeline { private static final Logger LOGGER = LoggerFactory.getLogger(ExposurePipeline.class); public static final int DEFAULT_CAPACITY = 1 << 16; // 65536 elements public static final int DEFAULT_FLUSH_INTERVAL_IN_SECONDS = 1; private static final int FLUSH_THRESHOLD = 100; private static final String EXPOSURES_ROUTE = "exposures"; - private static final long SHUTDOWN_TIMEOUT_MILLIS = 800; private final EventQueue queue; private final ExposureSerializingHandler serializer; @@ -51,22 +53,17 @@ public void start() { } @Override - public void close() { - final long deadline = - System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SHUTDOWN_TIMEOUT_MILLIS); + public void requestShutdown() { synchronized (this.lifecycleLock) { if (this.closed.compareAndSet(false, true)) { this.serializer.requestShutdown(); } } - final long remaining = deadline - System.nanoTime(); - if (Thread.currentThread() != this.serializerThread && remaining > 0) { - try { - TimeUnit.NANOSECONDS.timedJoin(this.serializerThread, remaining); - } catch (final InterruptedException e) { - Thread.currentThread().interrupt(); - } - } + } + + @Override + public void awaitTermination(final long deadlineNanos) { + awaitThread(this.serializerThread, deadlineNanos); } @Override diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationPipeline.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationPipeline.java index d2f3bd924fd..e3daf400dae 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationPipeline.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationPipeline.java @@ -1,5 +1,8 @@ package com.datadog.openfeature.internal.flagevaluation; +import static com.datadog.openfeature.internal.EventPipeline.awaitThread; + +import com.datadog.openfeature.internal.EventPipeline; import com.datadog.openfeature.internal.EventQueue; import com.datadog.openfeature.internal.RuntimeServices; import com.datadog.openfeature.internal.bridge.EventTransport; @@ -38,7 +41,7 @@ * the worker has been joined, counting any remainder as a closed drop so shutdown loss is * observable rather than silent. */ -public class FlagEvaluationPipeline implements FlagEvaluationWriter, AutoCloseable { +public class FlagEvaluationPipeline implements FlagEvaluationWriter, EventPipeline { private static final Logger LOGGER = LoggerFactory.getLogger(FlagEvaluationPipeline.class); @@ -122,37 +125,30 @@ int aggregatorFullTierSizeForTest() { } @Override - public void close() { - final boolean workerRunning; + public void requestShutdown() { synchronized (lifecycleLock) { - if (!closed.compareAndSet(false, true)) { - return; - } - workerRunning = this.serializerThread.isAlive(); - if (workerRunning) { - // Ask the worker to drain the queue and final-flush, then interrupt to wake it from poll(). + if (closed.compareAndSet(false, true)) { + // The worker polls every 100 ms: do not interrupt a current or final HTTP request. serializer.requestShutdown(); - this.serializerThread.interrupt(); } } + } + + @Override + public void awaitTermination(final long deadlineNanos) { if (Thread.currentThread() == this.serializerThread) { return; } - if (workerRunning) { - try { - // Bounded wait for the worker's final flush so queued events are not lost on shutdown. - this.serializerThread.join(TimeUnit.SECONDS.toMillis(5)); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } + awaitThread(this.serializerThread, deadlineNanos); // enqueue() is lock-free, so a producer that passed the closed check before close() ran can // still land an event after the worker's final drain. Sweep the remainder so that loss is // counted rather than silently stranded in a queue nobody polls again. Only safe once the // worker is gone: the queue is single-consumer, so sweeping alongside a live worker would - // break that contract. If join() timed out the worker is still draining, so skip the sweep. - if (!this.serializerThread.isAlive()) { - sweepAndCountResidualEvents(); + // break that contract. If the wait timed out the worker is still draining, so skip the sweep. + synchronized (lifecycleLock) { + if (!this.serializerThread.isAlive()) { + sweepAndCountResidualEvents(); + } } } @@ -294,12 +290,10 @@ public void run() { // On exit (interrupt or shutdown request), drain everything still buffered and flush it so // queued events are not lost on shutdown. // - // close() interrupts this thread to break it out of poll(). The final flush does socket - // I/O, which fails fast on a thread whose interrupt flag is set, so clear the flag - // for the duration of the drain and restore it afterwards. Without this the flush that - // close() exists to guarantee throws IOException, the aggregated rows are discarded by - // the clear() in flush(), and the loss is invisible to the drop counters because those - // rows already left the queue. + // Normal close only signals shutdown. If an unrelated interrupt ends the duty cycle, clear + // it during the final socket I/O and restore it afterwards: the flush otherwise fails + // fast, the aggregated rows are discarded by the clear() in flush(), and the loss is + // invisible to the drop counters because those rows already left the queue. final boolean wasInterrupted = Thread.interrupted(); try { drainAndFlush(); diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java index e8935ff5a92..ee5a4ab4d96 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java @@ -1,6 +1,8 @@ package com.datadog.openfeature.internal; import static java.util.Collections.emptyMap; +import static java.util.concurrent.TimeUnit.NANOSECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -18,11 +20,13 @@ import com.datadog.openfeature.internal.bridge.ShutdownSource; import com.datadog.openfeature.internal.bridge.SpanEnricher; import com.datadog.openfeature.internal.config.TestSettings; +import com.datadog.openfeature.internal.flagevaluation.FlagEvalEvent; import com.datadog.openfeature.internal.ufc.ServerConfiguration; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; import java.util.function.Consumer; import org.junit.jupiter.api.Test; @@ -280,6 +284,41 @@ void releaseUnregistersTheAgentShutdownFlush() { assertTrue(bridge.shutdownUnregistered); } + @Test + void stopFlushesBothPipelinesWithinOneShutdownBudget() throws Exception { + final CountDownLatch exposuresPosted = new CountDownLatch(1); + final CountDownLatch evaluationsPosted = new CountDownLatch(1); + final CountDownLatch releasePosts = new CountDownLatch(1); + final EventTransport stalledTransport = + (route, json) -> { + ("exposures".equals(route) ? exposuresPosted : evaluationsPosted).countDown(); + try { + releasePosts.await(10, SECONDS); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + }; + final FeatureFlagsRuntime runtime = + FeatureFlagsRuntime.create( + new TestAgentBridge(new TestConfigurationSource(), stalledTransport), + TestSettings.of("feature.flags.configuration.source", "remote_config")); + runtime.start(); + try { + runtime.recordExposure(exposure()); + runtime.evaluations().enqueue(new FlagEvalEvent("f", "v", "a", "s", 1L, emptyMap())); + + final long started = System.nanoTime(); + runtime.stop(); + final long elapsedMillis = NANOSECONDS.toMillis(System.nanoTime() - started); + + assertEquals(0, evaluationsPosted.getCount(), "the evaluations final flush started"); + assertEquals(0, exposuresPosted.getCount(), "the exposures final flush started"); + assertTrue(elapsedMillis < 1300, "Both pipeline waits took " + elapsedMillis + " ms"); + } finally { + releasePosts.countDown(); + } + } + @Test void failedStartDoesNotCacheTheSharedRuntime() { final ConfigurationSource failing = From 341b380aa87c5e2f39c3dd2330b965e7a311eeee Mon Sep 17 00:00:00 2001 From: Bruce Bujon Date: Tue, 20 Oct 2026 14:07:12 +0200 Subject: [PATCH 6/7] fix(ffe): Fall back to a JVM shutdown hook when the agent shutdown subscription fails A failing agent shutdown subscription escaped the shared runtime acquisition after the runtime started, leaking it and failing the provider initialization. Fall back to the JVM shutdown hook instead. Co-Authored-By: Claude Opus 5.5 --- .../internal/FeatureFlagsRuntime.java | 13 +++++++--- .../internal/FeatureFlagsRuntimeTest.java | 25 +++++++++++++++++++ 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java index b700a9f4c45..58ab706a4af 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java @@ -186,9 +186,16 @@ void start() { * @param bridge the bridge to the Datadog Java agent. */ void registerShutdown(final AgentBridge bridge) { - final ShutdownSource shutdown = bridge.shutdown(); - this.shutdownRegistration = - shutdown == null ? addShutdownHook() : shutdown.subscribe(this::stop); + AutoCloseable registration = null; + try { + final ShutdownSource shutdown = bridge.shutdown(); + if (shutdown != null) { + registration = shutdown.subscribe(this::stop); + } + } catch (final RuntimeException | LinkageError e) { + LOGGER.debug("Could not subscribe to the agent shutdown, using a JVM shutdown hook", e); + } + this.shutdownRegistration = registration == null ? addShutdownHook() : registration; } @Nullable diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java index ee5a4ab4d96..b140ebbfd7c 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java @@ -319,6 +319,22 @@ void stopFlushesBothPipelinesWithinOneShutdownBudget() throws Exception { } } + @Test + void failingAgentShutdownSubscriptionFallsBackToTheJvmShutdownHook() { + final TestConfigurationSource source = new TestConfigurationSource(); + final TestAgentBridge bridge = + new TestAgentBridge(source, null) + .withFailingAgentShutdown(new IllegalStateException("Shutdown in progress")); + + final FeatureFlagsRuntime runtime = + FeatureFlagsRuntime.acquire( + bridge, TestSettings.of("feature.flags.configuration.source", "remote_config")); + assertNotNull(runtime); + FeatureFlagsRuntime.release(); + + assertTrue(source.closed, "the started runtime is stopped on release"); + } + @Test void failedStartDoesNotCacheTheSharedRuntime() { final ConfigurationSource failing = @@ -385,6 +401,7 @@ private static final class TestAgentBridge implements AgentBridge { private final ConfigurationSource remoteConfiguration; private final EventTransport eventProxy; private boolean agentShutdown; + private RuntimeException shutdownFailure; volatile Runnable shutdownFlush; volatile boolean shutdownUnregistered; @@ -425,6 +442,9 @@ public ShutdownSource shutdown() { return null; } return listener -> { + if (this.shutdownFailure != null) { + throw this.shutdownFailure; + } this.shutdownFlush = listener; return () -> this.shutdownUnregistered = true; }; @@ -434,5 +454,10 @@ TestAgentBridge withAgentShutdown() { this.agentShutdown = true; return this; } + + TestAgentBridge withFailingAgentShutdown(final RuntimeException failure) { + this.shutdownFailure = failure; + return withAgentShutdown(); + } } } From 289b36c1d7996405363f1243c882d57b7d1553b4 Mon Sep 17 00:00:00 2001 From: Sarah Chen Date: Tue, 20 Oct 2026 16:06:38 -0500 Subject: [PATCH 7/7] fix(ffe): Prevent startup failures with older Java agents by returning caller defaults when the RC connector is unavailable --- .../com/datadog/openfeature/Provider.java | 12 +++- .../internal/FeatureFlagsRuntime.java | 22 +++++-- .../com/datadog/openfeature/ProviderTest.java | 58 ++++++++++++++++--- .../internal/FeatureFlagsRuntimeTest.java | 2 +- 4 files changed, 80 insertions(+), 14 deletions(-) diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java b/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java index 1dd5ca59701..b8b72493c9d 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java @@ -3,6 +3,7 @@ import static java.util.concurrent.TimeUnit.SECONDS; import com.datadog.openfeature.internal.FeatureFlagsRuntime; +import com.datadog.openfeature.internal.FeatureFlagsRuntime.RemoteConfigurationUnavailableException; import com.datadog.openfeature.internal.bridge.AgentBridge; import com.datadog.openfeature.internal.bridge.AgentBridges; import com.datadog.openfeature.internal.config.Settings; @@ -150,6 +151,14 @@ public void initialize(final EvaluationContext context) throws Exception { throw new ProviderNotReadyError( "Provider timed-out while waiting for initial configuration"); } + } catch (final RemoteConfigurationUnavailableException e) { + // No RC connector means no configuration can arrive in this JVM. Complete registration + // without waiting or switching sources. The evaluator has no runtime/configuration and + // returns caller defaults with PROVIDER_NOT_READY, never successful flag assignments. + // OpenFeature marks the completed registration READY; that does not imply RC is available. + initializationState.set(InitializationState.FALLBACK); + log.warn("{}; returning caller defaults", e.getMessage()); + log.debug("Feature Flags Remote Configuration connector unavailable", e); } catch (final OpenFeatureError e) { markInitializationError(); throw e; @@ -316,7 +325,8 @@ private enum InitializationState { INITIALIZING, INITIAL_CONFIG_RECEIVED, READY, - ERROR + ERROR, + FALLBACK } /** The provider options. */ diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java index 58ab706a4af..4ff8fecadac 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java @@ -104,11 +104,14 @@ static FeatureFlagsRuntime create(final AgentBridge bridge, final Settings setti final EventTransport transport; if (remoteConfig) { // TODO Support Remote Configuration without the Datadog Java agent. - final ConfigurationSource remoteConfiguration = bridge.remoteConfiguration(); + final ConfigurationSource remoteConfiguration; + try { + remoteConfiguration = bridge.remoteConfiguration(); + } catch (final LinkageError e) { + throw new RemoteConfigurationUnavailableException(e); + } if (remoteConfiguration == null) { - throw new IllegalStateException( - "Feature Flags Remote Configuration source requires the Datadog Java agent with" - + " Remote Configuration enabled"); + throw new RemoteConfigurationUnavailableException(null); } source = new RemoteConfigurationService(remoteConfiguration); // Remote Configuration never falls back to direct intake. @@ -149,6 +152,17 @@ static FeatureFlagsRuntime create(final AgentBridge bridge, final Settings setti return new FeatureFlagsRuntime(source, exposures, evaluations); } + /** Signals an unavailable RC connector before any runtime resources have been started. */ + public static final class RemoteConfigurationUnavailableException extends IllegalStateException { + private RemoteConfigurationUnavailableException(@Nullable final LinkageError cause) { + super( + "Feature Flags Remote Configuration requires a compatible Datadog Java agent with" + + " Remote Configuration enabled; attach a compatible agent or configure" + + " DD_FEATURE_FLAGS_CONFIGURATION_SOURCE=agentless for CDN delivery", + cause); + } + } + private static Map eventContext(final Settings settings) { final Map context = new HashMap<>(); putIfNotNull(context, "service", settings.service()); diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java index 65184e17ee9..97325594548 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java @@ -7,7 +7,9 @@ import static java.util.concurrent.TimeUnit.SECONDS; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertFalse; +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.ArgumentMatchers.any; @@ -370,18 +372,58 @@ public void testDisabledProductFailsInitialization() { assertThat(error.getCause().getMessage(), equalTo("Feature Flags is disabled")); } + @MethodSource("providerMethods") + @ParameterizedTest + public void testRemoteConfigurationWithoutCompatibleAgentUsesDefaults( + final String flag, final E defaultValue, final EvaluateMethod method) throws Exception { + final Provider provider = newRemoteConfigurationProvider(AgentBridge.NONE); + final OpenFeatureAPI api = OpenFeatureAPI.getInstance(); + + assertDoesNotThrow(() -> api.setProviderAndWait(provider)); + // READY means initialization completed; evaluation details still report unavailable config. + assertThat(api.getClient().getProviderState(), equalTo(ProviderState.READY)); + assertThat(initializationState(provider), equalTo("FALLBACK")); + assertFalse(this.evaluator.hasConfiguration()); + assertNull(this.evaluator.runtime(), "no polling or event delivery starts without RC"); + final FlagEvaluationDetails result = method.evaluate(api.getClient(), flag, defaultValue); + assertThat(result.getValue(), equalTo(defaultValue)); + assertThat(result.getErrorCode(), equalTo(ErrorCode.PROVIDER_NOT_READY)); + assertThat(result.getReason(), equalTo("ERROR")); + assertNull(result.getVariant()); + assertTrue(this.cdn.requests().isEmpty(), "explicit RC must not silently switch to CDN"); + } + @Test - public void testRemoteConfigurationWithoutAgentFailsInitialization() { + public void testIncompatibleRemoteConfigurationBridgeUsesDefaults() throws Exception { + final AgentBridge bridge = mock(AgentBridge.class); + when(bridge.healthMetrics()).thenReturn(HealthMetrics.NOOP); + when(bridge.remoteConfiguration()).thenThrow(new NoSuchMethodError("older agent RC bridge")); + final Provider provider = newRemoteConfigurationProvider(bridge); + final OpenFeatureAPI api = OpenFeatureAPI.getInstance(); + + assertDoesNotThrow(() -> api.setProviderAndWait(provider)); + assertThat(initializationState(provider), equalTo("FALLBACK")); + final FlagEvaluationDetails result = + api.getClient().getStringDetails("missing", "caller-default"); + assertThat(result.getValue(), equalTo("caller-default")); + assertThat(result.getErrorCode(), equalTo(ErrorCode.PROVIDER_NOT_READY)); + assertNull(this.evaluator.runtime()); + assertDoesNotThrow(api::shutdown); + } + + private Provider newRemoteConfigurationProvider(final AgentBridge bridge) { final Provider[] providerRef = new Provider[1]; - final DDEvaluator remoteConfig = + this.evaluator = new DDEvaluator( () -> providerRef[0].onConfigurationChange(), - AgentBridge.NONE, - TestSettings.of("feature.flags.configuration.source", "remote_config")); - providerRef[0] = new Provider(new Options().initTimeout(10, MILLISECONDS), remoteConfig); - - assertThrows( - FatalError.class, () -> OpenFeatureAPI.getInstance().setProviderAndWait(providerRef[0])); + bridge, + TestSettings.of( + "feature.flags.configuration.source", + "remote_config", + "feature.flags.configuration.source.agentless.base.url", + this.cdn.uri("/").toString())); + providerRef[0] = new Provider(new Options().initTimeout(10, MILLISECONDS), this.evaluator); + return providerRef[0]; } @Test diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java index b140ebbfd7c..995174b8fa4 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java @@ -56,7 +56,7 @@ void remoteConfigurationRequiresTheJavaAgent() { AgentBridge.NONE, TestSettings.of("feature.flags.configuration.source", "remote_config"))); - assertTrue(error.getMessage().contains("requires the Datadog Java agent")); + assertTrue(error.getMessage().contains("requires a compatible Datadog Java agent")); } @Test