From 63367a5945e9010c5841c37e7a7f7b887bad00d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Wed, 28 Oct 2026 10:30:36 +0100 Subject: [PATCH] fix: compute desired states outside of ConcurrentHashMap.computeIfAbsent Applying the desired state aspects inside computeIfAbsent failed with "Recursive update" when an aspect looked up the actual resource of the dependent (same key) or of another dependent whose key landed in the same bin. Compute with a lock per dependent instead, and hand the thread already computing a desired state the state as it is so far. Co-Authored-By: Claude Sonnet 5.5 --- .../api/reconciler/DefaultContext.java | 52 ++++++- .../DesiredStateAspectReentrancyTest.java | 145 ++++++++++++++++++ 2 files changed, 189 insertions(+), 8 deletions(-) create mode 100644 operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/DesiredStateAspectReentrancyTest.java diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java index 50e61526c4..328ef651c9 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java @@ -21,6 +21,8 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -50,6 +52,9 @@ public class DefaultContext

implements Context

{ private final boolean primaryResourceDeleted; private final boolean primaryResourceFinalStateUnknown; private final Map, Object> desiredStates = new ConcurrentHashMap<>(); + private final Map, DesiredStateInProgress> desiredStatesInProgress = + new ConcurrentHashMap<>(); + private final Map, Lock> desiredStateLocks = new ConcurrentHashMap<>(); private final ResourceOperations

resourceOperations; public DefaultContext( @@ -257,14 +262,43 @@ public DefaultContext

setRetryInfo(RetryInfo retryInfo) { @SuppressWarnings("unchecked") public R getOrComputeDesiredStateFor( DependentResource dependentResource, Function desiredStateComputer) { - return (R) - desiredStates.computeIfAbsent( - dependentResource, - ignored -> { - final var desired = desiredStateComputer.apply(getPrimaryResource()); - applyDesiredStateAspects(desired, dependentResource); - return desired; - }); + // an aspect can look up the actual resource of its dependent, which asks for the desired state + // being computed: the thread computing it is handed the state as it is so far, instead of + // re-entering the computation + final var computing = desiredStatesInProgress.get(dependentResource); + if (computing != null && computing.thread() == Thread.currentThread()) { + return (R) computing.desired(); + } + final var existing = desiredStates.get(dependentResource); + if (existing != null) { + return (R) existing; + } + // not computed within ConcurrentHashMap.computeIfAbsent, since an aspect can look up the actual + // resource of another dependent, which re-enters this method for a different key + final var lock = + desiredStateLocks.computeIfAbsent(dependentResource, ignored -> new ReentrantLock()); + lock.lock(); + try { + final var computedMeanwhile = desiredStates.get(dependentResource); + if (computedMeanwhile != null) { + return (R) computedMeanwhile; + } + final var desired = desiredStateComputer.apply(getPrimaryResource()); + if (desired == null) { + return null; + } + desiredStatesInProgress.put( + dependentResource, new DesiredStateInProgress(Thread.currentThread(), desired)); + try { + applyDesiredStateAspects(desired, dependentResource); + } finally { + desiredStatesInProgress.remove(dependentResource); + } + desiredStates.put(dependentResource, desired); + return desired; + } finally { + lock.unlock(); + } } /** @@ -280,4 +314,6 @@ private void applyDesiredStateAspects(Object desired, DependentResource de .forEach(aspect -> aspect.apply(hasMetadata, dependentResource, this)); } } + + private record DesiredStateInProgress(Thread thread, Object desired) {} } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/DesiredStateAspectReentrancyTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/DesiredStateAspectReentrancyTest.java new file mode 100644 index 0000000000..a7c85d928a --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/DesiredStateAspectReentrancyTest.java @@ -0,0 +1,145 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.dependent; + +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.junit.jupiter.api.Test; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.ConfigMapBuilder; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; +import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.DefaultContext; +import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource; +import io.javaoperatorsdk.operator.api.reconciler.dependent.DesiredStateAspect; +import io.javaoperatorsdk.operator.processing.Controller; +import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource; +import io.javaoperatorsdk.operator.processing.event.EventSourceManager; +import io.javaoperatorsdk.operator.processing.event.source.EventSource; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +@SuppressWarnings({"unchecked", "rawtypes"}) +class DesiredStateAspectReentrancyTest { + + private static final ConfigMap PRIMARY = + new ConfigMapBuilder() + .withNewMetadata() + .withName("primary") + .withNamespace("ns") + .endMetadata() + .build(); + + @Test + void aspectCanLookUpActualResourceOfTheDependent() { + final DesiredStateAspect aspect = + (desired, dependentResource, ctx) -> + ((DependentResource) dependentResource) + .getSecondaryResource(ctx.getPrimaryResource(), ctx); + final var context = contextWith(List.of(aspect)); + + final var dependent = new ChildConfigMapDependent(); + + assertThatCode(() -> dependent.getSecondaryResource(PRIMARY, context)) + .doesNotThrowAnyException(); + } + + /** + * The desired state of the other dependent is asked for while the one of the first is still being + * computed. Dependents that land in the same bin of the map holding the desired states used to + * fail with "Recursive update". + */ + @Test + void aspectCanLookUpActualResourceOfAnotherDependent() { + final var other = new ChildConfigMapDependent(); + final var dependent = new ChildConfigMapDependent(); + final DesiredStateAspect aspect = + (desired, dependentResource, ctx) -> { + if (dependentResource == dependent) { + ((DependentResource) other).getSecondaryResource(ctx.getPrimaryResource(), ctx); + } + }; + final var context = contextWith(List.of(aspect)); + + assertThatCode(() -> dependent.getSecondaryResource(PRIMARY, context)) + .doesNotThrowAnyException(); + } + + @Test + void desiredStateIsComputedOnce() { + final var context = contextWith(List.of()); + final var dependent = new ChildConfigMapDependent(); + + final var first = dependent.getOrComputeDesired(context); + + assertThat(dependent.getOrComputeDesired(context)).isSameAs(first); + } + + private static DefaultContext contextWith(List aspects) { + final ConfigurationService configurationService = mock(); + when(configurationService.desiredStateAspects()).thenReturn(aspects); + final ControllerConfiguration controllerConfiguration = mock(); + when(controllerConfiguration.getConfigurationService()).thenReturn(configurationService); + final Controller controller = mock(); + when(controller.getConfiguration()).thenReturn(controllerConfiguration); + final EventSourceManager esm = mock(); + final EventSource es = mock(); + when(es.getSecondaryResources(any())).thenReturn(Set.of(configMap("actual"))); + when(esm.getEventSourcesFor(ConfigMap.class)).thenReturn(List.of(es)); + when(controller.getEventSourceManager()).thenReturn(esm); + return new DefaultContext<>(null, controller, PRIMARY, false, false); + } + + private static ConfigMap configMap(String value) { + return new ConfigMapBuilder() + .withNewMetadata() + .withName("child") + .withNamespace("ns") + .endMetadata() + .withData(Map.of("key", value)) + .build(); + } + + private static class ChildConfigMapDependent + extends KubernetesDependentResource { + ChildConfigMapDependent() { + super(ConfigMap.class); + } + + @Override + public int hashCode() { + return 1; + } + + @Override + protected ConfigMap desired(ConfigMap primary, Context context) { + return configMap("desired"); + } + + @Override + public ConfigMap getOrComputeDesired(Context context) { + return super.getOrComputeDesired(context); + } + } +}