fix: complete the submission when the primary is missing from cache at execution time

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

@@ -543,6 +543,7 @@ public void run() {
       // change thread name for easier debugging
       final var thread = Thread.currentThread();
       final var name = thread.getName();
+      boolean reconciliationStarted = false;
       try {
         // we try to get the most up-to-date resource from cache
         var actualResource = cache.get(resourceID);
@@ -567,19 +568,27 @@ public void run() {
             }
           } else {
             log.debug("Skipping execution; primary resource missing from cache");
+            // the submission has to be completed, otherwise the resource stays marked as under
+            // processing and no further reconciliation would ever be submitted for it
+            eventProcessingFinished(executionScope, PostExecutionControl.defaultDispatch());
             return;
           }
         }
         actualResource.ifPresent(executionScope::setResource);
         MDCUtils.addResourceInfo(executionScope.getResource());
         metrics.reconciliationStarted(executionScope.getResource(), metricsMetadata);
+        reconciliationStarted = true;
         thread.setName("ReconcilerExecutor-" + controllerName() + "-" + thread.getId());
         PostExecutionControl<P> postExecutionControl =
             reconciliationDispatcher.handleExecution(executionScope);
         eventProcessingFinished(executionScope, postExecutionControl);
       } finally {
-        metrics.reconciliationFinished(
-            executionScope.getResource(), executionScope.getRetryInfo(), metricsMetadata);
+        // only report the reconciliation as finished if it was reported as started, otherwise
+        // gauges tracking in-flight reconciliations drift on every skipped execution
+        if (reconciliationStarted) {
+          metrics.reconciliationFinished(
+              executionScope.getResource(), executionScope.getRetryInfo(), metricsMetadata);
+        }
         // restore original name
         thread.setName(name);
         MDCUtils.removeResourceInfo();

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

@@ -134,6 +134,53 @@ void skipProcessingIfLatestCustomResourceNotInCache() {
     verify(reconciliationDispatcherMock, timeout(50).times(0)).handleExecution(any());
   }
 
+  @Test
+  void completesSubmissionIfResourceDisappearsFromCacheBeforeExecution() {
+    var resourceID = new ResourceID(UUID.randomUUID().toString(), TEST_NAMESPACE);
+    var customResource = testCustomResource(resourceID);
+    // present when the execution is submitted, but gone by the time the executor thread runs
+    when(controllerEventSourceMock.get(eq(resourceID)))
+        .thenReturn(Optional.of(customResource))
+        .thenReturn(Optional.empty());
+
+    eventProcessor.handleEvent(
+        new ResourceEvent(ResourceAction.UPDATED, resourceID, customResource));
+
+    await()
+        .atMost(Duration.ofSeconds(1))
+        .untilAsserted(() -> assertThat(eventProcessor.isUnderProcessing(resourceID)).isFalse());
+    verify(reconciliationDispatcherMock, times(0)).handleExecution(any());
+  }
+
+  @Test
+  void doesNotReportReconciliationFinishedIfItNeverStarted() {
+    var processorWithMetrics =
+        spy(
+            new EventProcessor(
+                controllerConfiguration(null, rateLimiterMock),
+                reconciliationDispatcherMock,
+                eventSourceManagerMock,
+                metricsMock));
+    processorWithMetrics.start();
+    when(processorWithMetrics.retryEventSource()).thenReturn(retryTimerEventSourceMock);
+
+    var resourceID = new ResourceID(UUID.randomUUID().toString(), TEST_NAMESPACE);
+    var customResource = testCustomResource(resourceID);
+    when(controllerEventSourceMock.get(eq(resourceID)))
+        .thenReturn(Optional.of(customResource))
+        .thenReturn(Optional.empty());
+
+    processorWithMetrics.handleEvent(
+        new ResourceEvent(ResourceAction.UPDATED, resourceID, customResource));
+
+    await()
+        .atMost(Duration.ofSeconds(1))
+        .untilAsserted(
+            () -> assertThat(processorWithMetrics.isUnderProcessing(resourceID)).isFalse());
+    verify(metricsMock, times(0)).reconciliationStarted(any(), any());
+    verify(metricsMock, times(0)).reconciliationFinished(any(), any(), any());
+  }
+
   @Test
   void ifExecutionInProgressWaitsUntilItsFinished() {
     ResourceID resourceUid = eventAlreadyUnderProcessing();