fix: race condition in workflow reconciler (#2549)

Signed-off-by: Chris Laprun <[email protected]>

Author: metacosm

Diff for: operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
@@ -36,7 +37,7 @@ protected AbstractWorkflowExecutor(DefaultWorkflow<P> workflow, P primary, Conte
     this.context = context;
     this.primaryID = ResourceID.fromResource(primary);
     executorService = context.getWorkflowExecutorService();
-    results = new ConcurrentHashMap<>(workflow.getDependentResourcesByName().size());
+    results = new HashMap<>(workflow.getDependentResourcesByName().size());
   }
 
   protected abstract Logger logger();
@@ -84,13 +85,13 @@ protected boolean isMarkedForDelete(DependentResourceNode<?, P> drn) {
     return getResultFlagFor(drn, WorkflowResult.DetailBuilder::isMarkedForDelete);
   }
 
-  protected WorkflowResult.DetailBuilder createOrGetResultFor(
+  protected synchronized WorkflowResult.DetailBuilder createOrGetResultFor(
       DependentResourceNode<?, P> dependentResourceNode) {
     return results.computeIfAbsent(dependentResourceNode,
         unused -> new WorkflowResult.DetailBuilder());
   }
 
-  protected Optional<WorkflowResult.DetailBuilder<?>> getResultFor(
+  protected synchronized Optional<WorkflowResult.DetailBuilder<?>> getResultFor(
       DependentResourceNode<?, P> dependentResourceNode) {
     return Optional.ofNullable(results.get(dependentResourceNode));
   }
@@ -115,8 +116,8 @@ protected synchronized void handleExceptionInExecutor(
     createOrGetResultFor(dependentResourceNode).withError(e);
   }
 
-  protected boolean isNotReady(DependentResourceNode<?, P> dependentResourceNode) {
-    return getResultFlagFor(dependentResourceNode, WorkflowResult.DetailBuilder::isNotReady);
+  protected boolean isReady(DependentResourceNode<?, P> dependentResourceNode) {
+    return getResultFlagFor(dependentResourceNode, WorkflowResult.DetailBuilder::isReady);
   }
 
   protected boolean isInError(DependentResourceNode<?, P> dependentResourceNode) {
@@ -132,15 +133,17 @@ protected synchronized void handleNodeExecutionFinish(
     }
   }
 
-  @SuppressWarnings("unchecked")
+  @SuppressWarnings({"unchecked", "OptionalUsedAsFieldOrParameterType"})
   protected <R> boolean isConditionMet(
       Optional<ConditionWithType<R, P, ?>> condition,
       DependentResourceNode<R, P> dependentResource) {
     final var dr = dependentResource.getDependentResource();
     return condition.map(c -> {
       final DetailedCondition.Result<?> r = c.detailedIsMet(dr, primary, context);
-      results.computeIfAbsent(dependentResource, unused -> new WorkflowResult.DetailBuilder())
-          .withResultForCondition(c, r);
+      synchronized (this) {
+        results.computeIfAbsent(dependentResource, unused -> new WorkflowResult.DetailBuilder())
+            .withResultForCondition(c, r);
+      }
       return r;
     }).orElse(DetailedCondition.Result.metWithoutResult).isSuccess();
   }
@@ -170,7 +173,7 @@ protected <R> void registerOrDeregisterEventSourceBasedOnActivation(
     }
   }
 
-  protected Map<DependentResource, WorkflowResult.Detail<?>> asDetails() {
+  protected synchronized Map<DependentResource, WorkflowResult.Detail<?>> asDetails() {
     return results.entrySet().stream()
         .collect(
             Collectors.toMap(e -> e.getKey().getDependentResource(), e -> e.getValue().build()));

Diff for: operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutor.java

@@ -120,7 +120,7 @@ private synchronized void handleDelete(DependentResourceNode dependentResourceNo
 
   private boolean allDependentsDeletedAlready(DependentResourceNode<?, P> dependentResourceNode) {
     var dependents = dependentResourceNode.getParents();
-    return dependents.stream().allMatch(d -> alreadyVisited(d) && !isNotReady(d)
+    return dependents.stream().allMatch(d -> alreadyVisited(d) && isReady(d)
         && !isInError(d) && !postDeleteConditionNotMet(d));
   }
 
@@ -231,7 +231,7 @@ private void markDependentsForDelete(DependentResourceNode<?, P> dependentResour
   private boolean allParentsReconciledAndReady(DependentResourceNode<?, ?> dependentResourceNode) {
     return dependentResourceNode.getDependsOn().isEmpty()
         || dependentResourceNode.getDependsOn().stream()
-            .allMatch(d -> alreadyVisited(d) && !isNotReady(d));
+            .allMatch(d -> alreadyVisited(d) && isReady(d));
   }
 
   private boolean hasErroredParent(DependentResourceNode<?, ?> dependentResourceNode) {

Diff for: operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowResult.java

@@ -178,8 +178,8 @@ public boolean hasPostDeleteConditionNotMet() {
       return deletePostconditionResult != null && !deletePostconditionResult.isSuccess();
     }
 
-    public boolean isNotReady() {
-      return readyPostconditionResult != null && !readyPostconditionResult.isSuccess();
+    public boolean isReady() {
+      return readyPostconditionResult == null || readyPostconditionResult.isSuccess();
     }
 
     DetailBuilder<R> markAsVisited() {

Diff for: operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/workflow/WorkflowReconcileExecutorTest.java

@@ -450,7 +450,8 @@ void readyConditionNotMetInOneParent() {
   void diamondShareWithReadyCondition() {
     var workflow = new WorkflowBuilder<TestCustomResource>()
         .addDependentResource(dr1)
-        .addDependentResourceAndConfigure(dr2).toDependOn(dr1)
+        .addDependentResourceAndConfigure(dr2)
+        .toDependOn(dr1)
         .withReadyPostcondition(notMetCondition)
         .addDependentResourceAndConfigure(dr3).toDependOn(dr1)
         .addDependentResourceAndConfigure(dr4).toDependOn(dr2, dr3)