From 9df9c50a691ae6f16839d51d821b01debae3aa92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Wed, 29 Jul 2026 18:01:12 +0200 Subject: [PATCH] fix: complete the submission when the primary is missing from cache at execution time `ReconcilerExecutor.run` re-reads the primary from the cache and bails out if it is gone. On the non-triggerOnAllEvents path that bail-out just returned: } else { log.debug("Skipping execution; primary resource missing from cache"); return; } Two consequences. The resource stays wedged. `submitReconciliationExecution` has already called `state.setUnderProcessing(true)` and `state.unMarkEventReceived()`, so returning without `eventProcessingFinished` leaves the state marked as under processing with no event pending. `isControllerUnderExecution` then suppresses every future submission for that resource. Usually a DELETED event drops the whole `ResourceState` and hides this, but if the resource leaves the informer cache without a delete event reaching the processor (a narrowed label selector, `changeNamespaces` dropping a namespace, a re-list whose synthetic delete is filtered by an `onDeleteFilter`) the resource is never reconciled again. Metrics drift. The `return` is inside the `try`, so the `finally` still reported `reconciliationFinished` even though `reconciliationStarted` was never reached. With `MicrometerMetricsV2` that decrements `reconciliations.active` without a matching increment, and leaves the `reconciliations.queue` increment from `reconciliationSubmitted` outstanding, so both gauges drift monotonically. The same applied to the "delete event received meanwhile" branch. Now completes the submission via `eventProcessingFinished` (as the sibling branch already did) and only reports `reconciliationFinished` when `reconciliationStarted` was reported. Adds regression tests for both; they fail without this change. Note: `reconciliations.queue` is incremented by `reconciliationSubmitted` but only decremented by `reconciliationStarted`, so a submission that is abandoned still leaves the queue gauge one too high. Pairing those properly needs a decision on the `Metrics` contract (decrement in `reconciliationFinished`, or add an explicit "submission abandoned" callback) and is deliberately not changed here. --- .../processing/event/EventProcessor.java | 13 ++++- .../processing/event/EventProcessorTest.java | 47 +++++++++++++++++++ 2 files changed, 58 insertions(+), 2 deletions(-) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java index ddc0f73a27..374beb91e9 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventProcessor.java +++ b/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

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(); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java index f7864f2f16..941da8be4c 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/EventProcessorTest.java +++ b/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();