feat: reading cache just on reconciliation dispatching (#1640)

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java

@@ -151,11 +151,11 @@ private void submitReconciliationExecution(ResourceState state) {
         }
         state.setUnderProcessing(true);
         final var latest = maybeLatest.get();
-        ExecutionScope<P> executionScope = new ExecutionScope<>(latest, state.getRetry());
+        ExecutionScope<P> executionScope = new ExecutionScope<>(state.getRetry());
         state.unMarkEventReceived();
         metrics.reconcileCustomResource(latest, state.getRetry(), metricsMetadata);
         log.debug("Executing events for custom resource. Scope: {}", executionScope);
-        executor.execute(new ReconcilerExecutor(executionScope));
+        executor.execute(new ReconcilerExecutor(resourceID, executionScope));
       } else {
         log.debug(
             "Skipping executing controller for resource id: {}. Controller in execution: {}. Latest Resource present: {}",
@@ -388,9 +388,11 @@ private void handleAlreadyMarkedEvents() {
 
   private class ReconcilerExecutor implements Runnable {
     private final ExecutionScope<P> executionScope;
+    private final ResourceID resourceID;
 
-    private ReconcilerExecutor(ExecutionScope<P> executionScope) {
+    private ReconcilerExecutor(ResourceID resourceID, ExecutionScope<P> executionScope) {
       this.executionScope = executionScope;
+      this.resourceID = resourceID;
     }
 
     @Override
@@ -399,6 +401,13 @@ public void run() {
       final var thread = Thread.currentThread();
       final var name = thread.getName();
       try {
+        var actualResource = cache.get(resourceID);
+        if (actualResource.isEmpty()) {
+          log.debug("Skipping execution; primary resource missing from cache: {}",
+              resourceID);
+          return;
+        }
+        actualResource.ifPresent(executionScope::setResource);
         MDCUtils.addResourceInfo(executionScope.getResource());
         thread.setName("ReconcilerExecutor-" + controllerName() + "-" + thread.getId());
         PostExecutionControl<P> postExecutionControl =

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/ExecutionScope.java

@@ -6,14 +6,18 @@
 class ExecutionScope<R extends HasMetadata> {
 
   // the latest custom resource from cache
-  private final R resource;
+  private R resource;
   private final RetryInfo retryInfo;
 
-  ExecutionScope(R resource, RetryInfo retryInfo) {
-    this.resource = resource;
+  ExecutionScope(RetryInfo retryInfo) {
     this.retryInfo = retryInfo;
   }
 
+  public ExecutionScope<R> setResource(R resource) {
+    this.resource = resource;
+    return this;
+  }
+
   public R getResource() {
     return resource;
   }

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java

@@ -120,7 +120,9 @@ void ifExecutionInProgressWaitsUntilItsFinished() {
   void schedulesAnEventRetryOnException() {
     TestCustomResource customResource = testCustomResource();
 
-    ExecutionScope executionScope = new ExecutionScope(customResource, null);
+    ExecutionScope executionScope =
+        new ExecutionScope(null);
+    executionScope.setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.exceptionDuringExecution(new RuntimeException("test"));
 
@@ -254,7 +256,7 @@ void cancelScheduleOnceEventsOnSuccessfulExecution() {
     var crID = new ResourceID("test-cr", TEST_NAMESPACE);
     var cr = testCustomResource(crID);
 
-    eventProcessor.eventProcessingFinished(new ExecutionScope(cr, null),
+    eventProcessor.eventProcessingFinished(new ExecutionScope(null).setResource(cr),
         PostExecutionControl.defaultDispatch());
 
     verify(retryTimerEventSourceMock, times(1)).cancelOnceSchedule(eq(crID));
@@ -283,7 +285,8 @@ void startProcessedMarkedEventReceivedBefore() {
   @Test
   void updatesEventSourceHandlerIfResourceUpdated() {
     TestCustomResource customResource = testCustomResource();
-    ExecutionScope executionScope = new ExecutionScope(customResource, null);
+    ExecutionScope executionScope =
+        new ExecutionScope(null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.customResourceUpdated(customResource);
 
@@ -297,7 +300,8 @@ void updatesEventSourceHandlerIfResourceUpdated() {
   @Test
   void notUpdatesEventSourceHandlerIfResourceUpdated() {
     TestCustomResource customResource = testCustomResource();
-    ExecutionScope executionScope = new ExecutionScope(customResource, null);
+    ExecutionScope executionScope =
+        new ExecutionScope(null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.customResourceStatusPatched(customResource);
 
@@ -311,7 +315,8 @@ void notUpdatesEventSourceHandlerIfResourceUpdated() {
   void notReschedulesAfterTheFinalizerRemoveProcessed() {
     TestCustomResource customResource = testCustomResource();
     markForDeletion(customResource);
-    ExecutionScope executionScope = new ExecutionScope(customResource, null);
+    ExecutionScope executionScope =
+        new ExecutionScope(null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.customResourceFinalizerRemoved(customResource);
 
@@ -324,7 +329,8 @@ void notReschedulesAfterTheFinalizerRemoveProcessed() {
   void skipEventProcessingIfFinalizerRemoveProcessed() {
     TestCustomResource customResource = testCustomResource();
     markForDeletion(customResource);
-    ExecutionScope executionScope = new ExecutionScope(customResource, null);
+    ExecutionScope executionScope =
+        new ExecutionScope(null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.customResourceFinalizerRemoved(customResource);
 
@@ -341,7 +347,8 @@ void skipEventProcessingIfFinalizerRemoveProcessed() {
   void newResourceAfterMissedDeleteEvent() {
     TestCustomResource customResource = testCustomResource();
     markForDeletion(customResource);
-    ExecutionScope executionScope = new ExecutionScope(customResource, null);
+    ExecutionScope executionScope =
+        new ExecutionScope(null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.customResourceFinalizerRemoved(customResource);
     var newResource = testCustomResource();
@@ -377,7 +384,8 @@ void rateLimitsReconciliationSubmission() {
   @Test
   void schedulesRetryForMarReconciliationInterval() {
     TestCustomResource customResource = testCustomResource();
-    ExecutionScope executionScope = new ExecutionScope(customResource, null);
+    ExecutionScope executionScope =
+        new ExecutionScope(null).setResource(customResource);
     PostExecutionControl postExecutionControl =
         PostExecutionControl.defaultDispatch();
 
@@ -398,7 +406,8 @@ void schedulesRetryForMarReconciliationIntervalIfRetryExhausted() {
             eventSourceManagerMock,
             metricsMock));
     eventProcessorWithRetry.start();
-    ExecutionScope executionScope = new ExecutionScope(testCustomResource(), null);
+    ExecutionScope executionScope =
+        new ExecutionScope(null).setResource(testCustomResource());
     PostExecutionControl postExecutionControl =
         PostExecutionControl.exceptionDuringExecution(new RuntimeException());
     when(eventProcessorWithRetry.retryEventSource()).thenReturn(retryTimerEventSourceMock);

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java

@@ -374,7 +374,6 @@ void propagatesRetryInfoToContextIfFinalizerSet() {
 
     reconciliationDispatcher.handleExecution(
         new ExecutionScope(
-            testCustomResource,
             new RetryInfo() {
               @Override
               public int getAttemptCount() {
@@ -385,7 +384,7 @@ public int getAttemptCount() {
               public boolean isLastAttempt() {
                 return true;
               }
-            }));
+            }).setResource(testCustomResource));
 
     ArgumentCaptor<Context> contextArgumentCaptor =
         ArgumentCaptor.forClass(Context.class);
@@ -504,7 +503,6 @@ void callErrorStatusHandlerIfImplemented() {
 
     reconciliationDispatcher.handleExecution(
         new ExecutionScope(
-            testCustomResource,
             new RetryInfo() {
               @Override
               public int getAttemptCount() {
@@ -515,7 +513,7 @@ public int getAttemptCount() {
               public boolean isLastAttempt() {
                 return true;
               }
-            }));
+            }).setResource(testCustomResource));
 
     verify(customResourceFacade, times(1)).updateStatus(testCustomResource);
     verify(((ErrorStatusHandler) reconciler), times(1)).updateErrorStatus(eq(testCustomResource),
@@ -535,8 +533,7 @@ void callErrorStatusHandlerEvenOnFirstError() {
     };
 
     var postExecControl = reconciliationDispatcher.handleExecution(
-        new ExecutionScope(
-            testCustomResource, null));
+        new ExecutionScope(null).setResource(testCustomResource));
     verify(customResourceFacade, times(1)).updateStatus(testCustomResource);
     verify(((ErrorStatusHandler) reconciler), times(1)).updateErrorStatus(eq(testCustomResource),
         any(), any());
@@ -555,8 +552,7 @@ void errorHandlerCanInstructNoRetryWithUpdate() {
     };
 
     var postExecControl = reconciliationDispatcher.handleExecution(
-        new ExecutionScope(
-            testCustomResource, null));
+        new ExecutionScope(null).setResource(testCustomResource));
 
     verify(((ErrorStatusHandler) reconciler), times(1)).updateErrorStatus(eq(testCustomResource),
         any(), any());
@@ -576,8 +572,7 @@ void errorHandlerCanInstructNoRetryNoUpdate() {
     };
 
     var postExecControl = reconciliationDispatcher.handleExecution(
-        new ExecutionScope(
-            testCustomResource, null));
+        new ExecutionScope(null).setResource(testCustomResource));
 
     verify(((ErrorStatusHandler) reconciler), times(1)).updateErrorStatus(eq(testCustomResource),
         any(), any());
@@ -596,8 +591,7 @@ void errorStatusHandlerCanPatchResource() {
 
 
     reconciliationDispatcher.handleExecution(
-        new ExecutionScope(
-            testCustomResource, null));
+        new ExecutionScope(null).setResource(testCustomResource));
 
     verify(customResourceFacade, times(1)).patchStatus(eq(testCustomResource), any());
     verify(((ErrorStatusHandler) reconciler), times(1)).updateErrorStatus(eq(testCustomResource),
@@ -622,8 +616,7 @@ void ifRetryLimitedToZeroMaxAttemptsErrorHandlerGetsCorrectLastAttempt() {
     reconciler.errorHandler = mockErrorHandler;
 
     reconciliationDispatcher.handleExecution(
-        new ExecutionScope(
-            testCustomResource, null));
+        new ExecutionScope(null).setResource(testCustomResource));
 
     verify(mockErrorHandler, times(1)).updateErrorStatus(any(),
         ArgumentMatchers.argThat((ArgumentMatcher<Context<TestCustomResource>>) context -> {
@@ -667,7 +660,7 @@ private void removeFinalizers(CustomResource customResource) {
   }
 
   public <T extends HasMetadata> ExecutionScope<T> executionScopeWithCREvent(T resource) {
-    return new ExecutionScope<>(resource, null);
+    return (ExecutionScope<T>) new ExecutionScope<>(null).setResource(resource);
   }
 
   private class TestReconciler