From 96811784bdd14e95a2815a7305455277636769a9 Mon Sep 17 00:00:00 2001 From: Antonio Fernandez Alhambra Date: Thu, 17 Sep 2026 18:57:33 +0200 Subject: [PATCH 1/2] fix: evaluate activation conditions outside the workflow executor's monitor Signed-off-by: Antonio Fernandez Alhambra --- .../workflow/AbstractWorkflowExecutor.java | 27 ++- .../workflow/WorkflowReconcileExecutor.java | 38 ++-- .../WorkflowReconcileExecutorTest.java | 188 +++++++++++++++++- 3 files changed, 233 insertions(+), 20 deletions(-) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java index 665d80063b..796a6e61b7 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java @@ -132,6 +132,20 @@ protected synchronized void handleExceptionInExecutor( createOrGetResultFor(dependentResourceNode).withError(e); } + /** + * Called with the monitor held, after the node's execution mark is cleared and before the + * workflow may observe that no execution is left. Anything scheduled from here keeps its own + * execution mark, so the workflow does not complete while it is still running. + * + *

Implementations must not block, as the monitor is held, and must schedule any follow up work + * synchronously before returning. Scheduling it from another thread would allow that work to + * finish before its execution mark is set, leaving behind a mark that is never cleared. + * + *

An exception thrown here is recorded as an error on the node instead of being propagated, so + * that the workflow waiting on the monitor is notified in any case. + */ + protected void onNodeExecutionFinished(DependentResourceNode dependentResourceNode) {} + protected boolean isReady(DependentResourceNode dependentResourceNode) { return getResultFlagFor(dependentResourceNode, BaseWorkflowResult.DetailBuilder::isReady); } @@ -144,8 +158,17 @@ protected synchronized void handleNodeExecutionFinish( DependentResourceNode dependentResourceNode) { logger().trace("Finished execution for: {} primary: {}", dependentResourceNode, primaryID); actualExecutions.remove(dependentResourceNode); - if (noMoreExecutionsScheduled()) { - this.notifyAll(); + try { + onNodeExecutionFinished(dependentResourceNode); + } catch (Exception e) { + handleExceptionInExecutor(dependentResourceNode, e); + } catch (Error e) { + logger().error("java.lang.Error during execution finish", e); + throw e; + } finally { + if (noMoreExecutionsScheduled()) { + this.notifyAll(); + } } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java index d544891aba..939f4160d8 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java @@ -18,6 +18,7 @@ import java.util.ArrayList; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,6 +35,9 @@ class WorkflowReconcileExecutor

extends AbstractWorkflowE private static final String RECONCILE = "reconcile"; private static final String DELETE = "delete"; + private final Set> conditionNotMetNodes = + ConcurrentHashMap.newKeySet(); + public WorkflowReconcileExecutor(DefaultWorkflow

workflow, P primary, Context

context) { super(workflow, primary, context); } @@ -51,6 +55,13 @@ protected Logger logger() { return log; } + @Override + protected void onNodeExecutionFinished(DependentResourceNode dependentResourceNode) { + if (conditionNotMetNodes.remove(dependentResourceNode)) { + handleReconcileOrActivationConditionNotMet(dependentResourceNode); + } + } + private synchronized void handleReconcile(DependentResourceNode dependentResourceNode) { log.debug("Considering for reconcile: {} primaryID: {}", dependentResourceNode, primaryID); @@ -90,20 +101,7 @@ private synchronized void handleReconcile(DependentResourceNode depend return; } - boolean activationConditionMet = - isConditionMet(dependentResourceNode.getActivationCondition(), dependentResourceNode); - registerOrDeregisterEventSourceBasedOnActivation(activationConditionMet, dependentResourceNode); - - boolean reconcileConditionMet = true; - if (activationConditionMet) { - reconcileConditionMet = - isConditionMet(dependentResourceNode.getReconcilePrecondition(), dependentResourceNode); - } - if (!reconcileConditionMet || !activationConditionMet) { - handleReconcileOrActivationConditionNotMet(dependentResourceNode); - } else { - submit(dependentResourceNode, new NodeReconcileExecutor<>(dependentResourceNode), RECONCILE); - } + submit(dependentResourceNode, new NodeReconcileExecutor<>(dependentResourceNode), RECONCILE); } private synchronized void handleDelete(DependentResourceNode dependentResourceNode) { @@ -155,6 +153,17 @@ private NodeReconcileExecutor(DependentResourceNode dependentResourceNode) @Override protected void doRun(DependentResourceNode dependentResourceNode) { + final var activationConditionMet = + isConditionMet(dependentResourceNode.getActivationCondition(), dependentResourceNode); + registerOrDeregisterEventSourceBasedOnActivation( + activationConditionMet, dependentResourceNode); + if (!activationConditionMet + || !isConditionMet( + dependentResourceNode.getReconcilePrecondition(), dependentResourceNode)) { + conditionNotMetNodes.add(dependentResourceNode); + return; + } + final var dependentResource = dependentResourceNode.getDependentResource(); log.debug("Reconciling for primary: {} node: {} ", primaryID, dependentResourceNode); ReconcileResult reconcileResult = dependentResource.reconcile(primary, context); @@ -238,6 +247,7 @@ private void handleReconcileOrActivationConditionNotMet( Set bottomNodes = new HashSet<>(); markDependentsForDelete(dependentResourceNode, bottomNodes); bottomNodes.forEach(this::handleDelete); + handleDelete(dependentResourceNode); } private void markDependentsForDelete( diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java index 117fe2afb8..fa3e703a2f 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java @@ -15,13 +15,21 @@ */ package io.javaoperatorsdk.operator.processing.dependent.workflow; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInfo; +import org.junit.jupiter.api.Timeout; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -45,13 +53,12 @@ class WorkflowReconcileExecutorTest extends AbstractWorkflowExecutorTest { Context mockContext = spy(Context.class); ExecutorService executorService = Executors.newCachedThreadPool(); - EventSourceRetriever eventSourceRetriever = mock(EventSourceRetriever.class); + EventSourceRetriever eventSourceRetriever = mock(); TestDependent dr3 = new TestDependent("DR_3"); TestDependent dr4 = new TestDependent("DR_4"); @BeforeEach - @SuppressWarnings("unchecked") void setup(TestInfo testInfo) { log.debug("==> Starting test {}", testInfo.getDisplayName()); when(mockContext.managedWorkflowAndDependentResourceContext()) @@ -673,6 +680,161 @@ void deletesDependentsOfNonActiveDependentButNotTheNonActive() { assertThat(executionHistory).deleted(drDeleter, drDeleter3).notReconciled(dr1, drDeleter2); } + @Test + @Timeout(10) + void deleteScheduledByUnmetPreconditionFinishesBeforeReconcileReturns() { + // the delete is scheduled from the completion hook, while the monitor is still held + var deleteFinished = new AtomicBoolean(false); + var slowDeleter = + new TestDeleterDependent("SLOW_DELETER") { + @Override + public void delete(TestCustomResource primary, Context context) { + try { + Thread.sleep(200); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + deleteFinished.set(true); + super.delete(primary, context); + } + }; + + var workflow = + new WorkflowBuilder() + .addDependentResourceAndConfigure(slowDeleter) + .withReconcilePrecondition(notMetCondition) + .build(); + + workflow.reconcile(new TestCustomResource(), mockContext); + + assertTrue(deleteFinished.get(), "reconcile() returned while the delete was still running"); + } + + @Test + @Timeout(10) + void deletedIfReconcilePreconditionNotMetWhileAnotherDeleteCascadeIsRunning() { + // the shared dependent's cascade reaches this one while it is still evaluating its precondition + var sharedDependentDeleted = new CountDownLatch(1); + TestDeleterDependent drDeleter2 = new TestDeleterDependent("DR_DELETER_2"); + var drDeleter3 = + new TestDeleterDependent("DR_DELETER_3") { + @Override + public void delete(TestCustomResource primary, Context context) { + super.delete(primary, context); + sharedDependentDeleted.countDown(); + } + }; + Condition blockedNotMetCondition = + (dependentResource, primary, context) -> { + try { + if (!sharedDependentDeleted.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("the shared dependent was never deleted"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + return false; + }; + + var workflow = + new WorkflowBuilder() + .addDependentResourceAndConfigure(drDeleter) + .withReconcilePrecondition(notMetCondition) + .addDependentResourceAndConfigure(drDeleter2) + .withReconcilePrecondition(blockedNotMetCondition) + .addDependentResourceAndConfigure(drDeleter3) + .dependsOn(drDeleter, drDeleter2) + .build(); + + var res = workflow.reconcile(new TestCustomResource(), mockContext); + + assertThat(executionHistory).deleted(drDeleter3, drDeleter, drDeleter2); + Assertions.assertThat(res.getErroredDependents()).isEmpty(); + } + + @Test + @Timeout(10) + void completesIfSchedulingDeleteForUnmetPreconditionIsRejected() { + // the delete is scheduled from a hook that already runs in the node executor's finally block + ExecutorService rejectingSecondSubmit = mock(); + when(rejectingSecondSubmit.submit(any(Runnable.class))) + .thenAnswer(invocation -> executorService.submit((Runnable) invocation.getArgument(0))) + .thenThrow(new RejectedExecutionException("executor is shut down")); + when(mockContext.getWorkflowExecutorService()).thenReturn(rejectingSecondSubmit); + + var workflow = + new WorkflowBuilder() + .addDependentResourceAndConfigure(drDeleter) + .withReconcilePrecondition(notMetCondition) + .build(); + + var exception = + assertThrows( + AggregatedOperatorException.class, + () -> workflow.reconcile(new TestCustomResource(), mockContext)); + + Assertions.assertThat(exception.getAggregatedExceptions()) + .containsOnlyKeys(drDeleter.name()) + .extractingByKey(drDeleter.name()) + .isInstanceOf(RejectedExecutionException.class); + } + + @Test + @Timeout(10) + void activationConditionsEvaluatedConcurrently() { + // they can only meet at the barrier if neither of them holds the executor's monitor + var rendezvousCondition = rendezvousCondition(new CyclicBarrier(2)); + + var workflow = + new WorkflowBuilder() + .addDependentResourceAndConfigure(dr1) + .withActivationCondition(rendezvousCondition) + .addDependentResourceAndConfigure(dr2) + .withActivationCondition(rendezvousCondition) + .build(); + + var res = workflow.reconcile(new TestCustomResource(), mockContext); + + assertThat(executionHistory).reconciled(dr1, dr2); + Assertions.assertThat(res.getErroredDependents()).isEmpty(); + } + + @Test + void activationConditionErrorAttributedToItsOwnDependent() { + var workflow = + new WorkflowBuilder() + .addDependentResource(dr1) + .addDependentResourceAndConfigure(dr2) + .dependsOn(dr1) + .withActivationCondition(throwingCondition()) + .addDependentResourceAndConfigure(dr3) + .dependsOn(dr1) + .withThrowExceptionFurther(false) + .build(); + + var res = workflow.reconcile(new TestCustomResource(), mockContext); + + Assertions.assertThat(res.getErroredDependents()).containsOnlyKeys(dr2); + assertThat(executionHistory).reconciled(dr1, dr3).notReconciled(dr2); + } + + @Test + void activationConditionErrorOnTopLevelDependentDoesNotStopOthers() { + var workflow = + new WorkflowBuilder() + .addDependentResourceAndConfigure(dr1) + .withActivationCondition(throwingCondition()) + .addDependentResource(dr2) + .withThrowExceptionFurther(false) + .build(); + + var res = workflow.reconcile(new TestCustomResource(), mockContext); + + Assertions.assertThat(res.getErroredDependents()).containsOnlyKeys(dr1); + assertThat(executionHistory).reconciled(dr2).notReconciled(dr1); + } + @Test @SuppressWarnings("unchecked") void activationConditionOnlyCalledOnceOnDeleteDependents() { @@ -695,7 +857,6 @@ void activationConditionOnlyCalledOnceOnDeleteDependents() { } @Test - @SuppressWarnings("unchecked") void activationConditionEventSourceRegistrationWithParentWithFalsePrecondition() { var workflow = new WorkflowBuilder() @@ -713,7 +874,6 @@ void activationConditionEventSourceRegistrationWithParentWithFalsePrecondition() } @Test - @SuppressWarnings("unchecked") void activationConditionEventSourceRegistration() { var workflow = new WorkflowBuilder() @@ -826,4 +986,24 @@ void shouldReturnEmptyIfNoConditionResultExists() { final var reconcileResult = workflow.reconcile(new TestCustomResource(), mockContext); assertTrue(reconcileResult.getNotReadyDependentResult(dr1, Integer.class).isEmpty()); } + + private Condition rendezvousCondition(CyclicBarrier barrier) { + return (dependentResource, primary, context) -> { + try { + barrier.await(5, TimeUnit.SECONDS); + return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } catch (BrokenBarrierException | TimeoutException e) { + throw new IllegalStateException("activation conditions were serialized", e); + } + }; + } + + private Condition throwingCondition() { + return (dependentResource, primary, context) -> { + throw new IllegalStateException("Test exception"); + }; + } } From e8a5562f9754d7d8b99f7ab958b9a8a32bf6b861 Mon Sep 17 00:00:00 2001 From: Antonio Fernandez Alhambra Date: Fri, 18 Sep 2026 08:50:27 +0200 Subject: [PATCH 2/2] refactor: express the delete postcondition check positively Signed-off-by: Antonio Fernandez Alhambra --- .../dependent/workflow/AbstractWorkflowExecutor.java | 4 ++-- .../dependent/workflow/WorkflowCleanupExecutor.java | 2 +- .../dependent/workflow/WorkflowReconcileExecutor.java | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java index 796a6e61b7..d757eff4e4 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java @@ -92,8 +92,8 @@ protected boolean alreadyVisited(DependentResourceNode dependentResourceNo return getResultFlagFor(dependentResourceNode, BaseWorkflowResult.DetailBuilder::isVisited); } - protected boolean postDeleteConditionNotMet(DependentResourceNode drn) { - return getResultFlagFor(drn, BaseWorkflowResult.DetailBuilder::hasPostDeleteConditionNotMet); + protected boolean postDeleteConditionMet(DependentResourceNode drn) { + return !getResultFlagFor(drn, BaseWorkflowResult.DetailBuilder::hasPostDeleteConditionNotMet); } protected boolean isMarkedForDelete(DependentResourceNode drn) { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java index 1132d09350..ccc61edc39 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowCleanupExecutor.java @@ -137,7 +137,7 @@ private synchronized void handleDependentCleaned( private boolean allDependentsCleaned(DependentResourceNode dependentResourceNode) { List parents = dependentResourceNode.getParents(); return parents.isEmpty() - || parents.stream().allMatch(d -> alreadyVisited(d) && !postDeleteConditionNotMet(d)); + || parents.stream().allMatch(d -> alreadyVisited(d) && postDeleteConditionMet(d)); } @SuppressWarnings("unchecked") diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java index 939f4160d8..6917a95291 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java @@ -142,7 +142,7 @@ private boolean allDependentsDeletedAlready(DependentResourceNode dependen var dependents = dependentResourceNode.getParents(); return dependents.stream() .allMatch( - d -> alreadyVisited(d) && isReady(d) && !isInError(d) && !postDeleteConditionNotMet(d)); + d -> alreadyVisited(d) && isReady(d) && !isInError(d) && postDeleteConditionMet(d)); } private class NodeReconcileExecutor extends NodeExecutor {