From 72e7bbc4cef205b56f06939f6bb68ded12d30241 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Attila=20M=C3=A9sz=C3=A1ros?= Date: Thu, 22 Oct 2026 15:13:04 +0200 Subject: [PATCH] fix: avoid deadlock when changeNamespaces joins a populated shared informer Fabric8 replays an already populated informer's cache to a newly added handler synchronously, on the registering thread. changeNamespaces registered the handler on an informer starter thread while the event source held its monitor waiting for that thread, so the replay blocked on the monitor and deadlocked. Register the handlers on the calling thread and only start the informers in parallel. Fixes #3650 --- .../source/informer/InformerManager.java | 22 ++++-- ...nformerEventSourceNamespaceChangeTest.java | 71 ++++++++++++++++--- 2 files changed, 80 insertions(+), 13 deletions(-) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java index 6caf39ccd9..d53b65920e 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java @@ -120,19 +120,31 @@ public void changeNamespaces(Set namespaces) { return; } + // Registering the handler on an already populated shared informer replays its cache to the + // handler synchronously, on the registering thread. The event source holds its monitor while + // waiting for the informers below to start, so the handler has to be registered on this thread: + // a replay from an informer starter thread would block on that monitor and deadlock. + var newSources = newNamespaces.stream().map(this::getEventSourceForNamespace).toList(); controllerConfiguration .getConfigurationService() .getExecutorServiceManager() .boundedExecuteAndWaitForAllToComplete( - newNamespaces.stream(), - ns -> { - final var source = getEventSourceForNamespace(ns); + newSources.stream(), + source -> { // block until the informer's cache is synced (or the sync timeout elapses) start(source); - log.debug("Registered new {} -> {} for namespace: {}", this, source, ns); + log.debug( + "Registered new {} -> {} for namespace: {}", + this, + source, + source.getTargetNamespace()); return null; }, - ns -> "InformerStarter-" + ns + "-" + configuration.getResourceClass().getSimpleName()); + source -> + "InformerStarter-" + + source.getTargetNamespace() + + "-" + + configuration.getResourceClass().getSimpleName()); } private void start(InformerWrapper informerWrapper) { diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSourceNamespaceChangeTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSourceNamespaceChangeTest.java index a1ee558036..7a2ee4ee16 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSourceNamespaceChangeTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/ManagedInformerEventSourceNamespaceChangeTest.java @@ -15,13 +15,16 @@ */ package io.javaoperatorsdk.operator.processing.event.source.informer; +import java.time.Duration; import java.util.Set; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; import io.fabric8.kubernetes.api.model.apps.Deployment; import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.informers.ResourceEventHandler; import io.fabric8.kubernetes.client.informers.SharedIndexInformer; import io.fabric8.kubernetes.client.informers.cache.Cache; import io.javaoperatorsdk.operator.MockKubernetesClient; @@ -37,8 +40,11 @@ import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** @@ -100,22 +106,71 @@ void namespaceChangeIsStillAppliedWhileRunning() { assertThat(pool.numberOfInformersForResource(Deployment.class)).isZero(); } + /** + * Fabric8 replays the cache of an already populated informer to a newly added handler + * synchronously, on the thread registering it. That replay enters the event source's monitor, + * which the namespace change holds while waiting for the informers to start, so the handler must + * be registered on the thread holding the monitor rather than on an informer starter thread. + */ + @Test + void namespaceChangeJoiningAPopulatedInformerDoesNotDeadlock() { + var eventHandler = mock(EventHandler.class); + var eventSource = buildEventSource(eventHandler); + eventSource.start(); + replayCacheOnHandlerRegistration(); + + assertTimeoutPreemptively( + Duration.ofSeconds(10), () -> eventSource.changeNamespaces(Set.of("ns1", "ns2"))); + + assertThat(eventSource.manager().isWatchingNamespace("ns2")).isTrue(); + verify(eventHandler).handleEvent(any()); + eventSource.stop(); + } + + private void replayCacheOnHandlerRegistration() { + var informer = informer(); + var existing = new Deployment(); + existing.setMetadata( + new ObjectMetaBuilder() + .withName("existing") + .withNamespace("ns2") + .withResourceVersion("1") + .build()); + doAnswer( + invocation -> { + ResourceEventHandler handler = invocation.getArgument(0); + handler.onAdd(existing); + return null; + }) + .when(informer) + .addEventHandler(any()); + } + /** * A successfully started {@link InformerEventSource} lists its cache to populate the * primary-to-secondary index, and the mock client leaves the informer's store unstubbed. */ private void stubInformerStore() { - SharedIndexInformer informer = - clientMock - .resources(Deployment.class) - .inNamespace("ns1") - .withLabelSelector((String) null) - .withShardSelector(null) - .runnableInformer(0); + var informer = informer(); when(informer.getStore()).thenReturn(mock(Cache.class)); } + /** The mock client hands out the very same informer for every namespace. */ + private SharedIndexInformer informer() { + return clientMock + .resources(Deployment.class) + .inNamespace("ns1") + .withLabelSelector((String) null) + .withShardSelector(null) + .runnableInformer(0); + } + private InformerEventSource buildEventSource() { + return buildEventSource(mock(EventHandler.class)); + } + + private InformerEventSource buildEventSource( + EventHandler eventHandler) { ConfigurationService configurationService = ConfigurationService.newOverriddenConfigurationService( new BaseConfigurationService(), @@ -125,7 +180,7 @@ private InformerEventSource buildEventSource() { when(controllerConfiguration.getName()).thenReturn("controller"); var eventSource = new InformerEventSource(configuration); - eventSource.setEventHandler(mock(EventHandler.class)); + eventSource.setEventHandler(eventHandler); eventSource.setControllerConfiguration(controllerConfiguration); return eventSource; }