This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 581c67e8a [FLINK-41031] Close operator-owned resources on shutdown and
exit on startup failure (#1224)
581c67e8a is described below
commit 581c67e8aa6946928ad1a3aeee652b26f6c2fa4f
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Thu Oct 29 12:59:43 2026 +0200
[FLINK-41031] Close operator-owned resources on shutdown and exit on
startup failure (#1224)
---
.../flink/kubernetes/operator/FlinkOperator.java | 31 +++++++++++++++++++---
.../operator/config/FlinkConfigManager.java | 11 ++++++--
.../operator/health/CanaryResourceManager.java | 12 +++++++--
.../operator/health/OperatorHealthService.java | 11 ++++++--
.../metrics/KubernetesOperatorMetricGroup.java | 12 ++++++++-
.../service/FlinkResourceContextFactory.java | 10 +++++--
.../operator/config/FlinkConfigManagerTest.java | 9 +++++++
.../operator/health/HealthProbeTest.java | 15 ++++++++++-
.../metrics/KubernetesOperatorMetricGroupTest.java | 20 ++++++++++++++
.../service/FlinkResourceContextFactoryTest.java | 14 ++++++++++
.../operator/admission/FlinkOperatorWebhook.java | 26 +++++++++++++++---
.../admission/informer/InformerManager.java | 11 +++++++-
.../admission/informer/InformerManagerTest.java | 14 ++++++++++
13 files changed, 178 insertions(+), 18 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index 8a5af0e6b..4ee1db6bf 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
@@ -57,6 +57,7 @@ import
org.apache.flink.kubernetes.operator.utils.KubernetesClientUtils;
import org.apache.flink.kubernetes.operator.utils.OperatorPluginUtils;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
+import org.apache.flink.util.IOUtils;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.Operator;
@@ -69,6 +70,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
@@ -199,6 +201,7 @@ public class FlinkOperator {
var reconcilerFactory = new ReconcilerFactory(eventRecorder,
statusRecorder, autoscaler);
var observerFactory = new
FlinkDeploymentObserverFactory(eventRecorder);
var canaryResourceManager = new
CanaryResourceManager<FlinkDeployment>(configManager);
+ canaryResourceManagers.add(canaryResourceManager);
HealthProbe.INSTANCE.registerCanaryResourceManager(canaryResourceManager);
canaryResourceManagers.add(canaryResourceManager);
@@ -224,6 +227,7 @@ public class FlinkOperator {
var reconciler = new SessionJobReconciler(eventRecorder,
statusRecorder, autoscaler);
var observer = new FlinkSessionJobObserver(eventRecorder);
var canaryResourceManager = new
CanaryResourceManager<FlinkSessionJob>(configManager);
+ canaryResourceManagers.add(canaryResourceManager);
HealthProbe.INSTANCE.registerCanaryResourceManager(canaryResourceManager);
canaryResourceManagers.add(canaryResourceManager);
@@ -297,22 +301,41 @@ public class FlinkOperator {
registerSessionJobController();
registerSnapshotController();
registerBlueGreenController();
- operator.installShutdownHook(
-
baseConfig.get(KubernetesOperatorConfigOptions.OPERATOR_TERMINATION_TIMEOUT));
+ Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
operator.start();
if (operatorHealthService != null) {
HealthProbe.INSTANCE.setRuntimeInfo(operator.getRuntimeInfo());
- Runtime.getRuntime().addShutdownHook(new
Thread(operatorHealthService::stop));
operatorHealthService.start();
}
}
public void stop() {
+ // in-flight reconciliations are drained first
operator.stop();
+
+ var closeables = new ArrayList<AutoCloseable>();
+ closeables.add(ctxFactory);
+ closeables.addAll(canaryResourceManagers);
+ closeables.add(configManager);
+ closeables.add(metricGroup);
+ if (operatorHealthService != null) {
+ closeables.add(operatorHealthService);
+ }
+ try {
+ IOUtils.closeAll(closeables);
+ } catch (Exception e) {
+ LOG.warn("Failed to close operator components", e);
+ }
}
public static void main(String... args) {
EnvUtils.logEnvironmentInfo(LOG, "Flink Kubernetes Operator", args);
- new FlinkOperator(null).run();
+ try {
+ new FlinkOperator(null).run();
+ } catch (Throwable t) {
+ // Non-daemon threads started before the failure would keep the
JVM alive
+ LOG.error("Failed to start the operator", t);
+ System.exit(1);
+ }
}
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
index 9b9126bc5..ebbd8b557 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
@@ -74,11 +74,12 @@ import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConf
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.VERSION_CONF_PREFIX;
/** Configuration manager for the Flink operator. */
-public class FlinkConfigManager {
+public class FlinkConfigManager implements AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(FlinkConfigManager.class);
private static final ObjectMapper objectMapper = new ObjectMapper();
+ @VisibleForTesting final ScheduledExecutorService executorService;
private volatile Configuration defaultConfig;
private volatile FlinkOperatorConfiguration defaultOperatorConfiguration;
// TODO: From 1.11 release, snapshot CRD should be mandatory and this can
be removed.
@@ -133,7 +134,7 @@ public class FlinkConfigManager {
});
updateDefaultConfig(defaultConfig);
- ScheduledExecutorService executorService =
Executors.newSingleThreadScheduledExecutor();
+ this.executorService = Executors.newSingleThreadScheduledExecutor();
executorService.scheduleWithFixedDelay(
cache::cleanUp,
cacheTimeout.toMillis(),
@@ -149,6 +150,12 @@ public class FlinkConfigManager {
return snapshotCrdInstalled;
}
+ /** Stops the config cache cleanup and the dynamic config watcher. */
+ @Override
+ public void close() {
+ executorService.shutdownNow();
+ }
+
/**
* Update the base configuration for the operator. Newly generated configs
(observe, deploy,
* etc.) will use this as the base.
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManager.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManager.java
index e3eccc0b1..bd0d8660b 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManager.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManager.java
@@ -46,7 +46,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
* deployed, their reconciliation stops at this class.
*/
@RequiredArgsConstructor
-public class CanaryResourceManager<CR extends AbstractFlinkResource<?, ?>> {
+public class CanaryResourceManager<CR extends AbstractFlinkResource<?, ?>>
+ implements AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(CanaryResourceManager.class);
@@ -57,7 +58,8 @@ public class CanaryResourceManager<CR extends
AbstractFlinkResource<?, ?>> {
private final ConcurrentHashMap<ResourceID, CanaryResourceState>
canaryResources =
new ConcurrentHashMap<>();
- private final ScheduledExecutorService executorService =
Executors.newScheduledThreadPool(3);
+ @VisibleForTesting
+ final ScheduledExecutorService executorService =
Executors.newScheduledThreadPool(3);
private final FlinkConfigManager configManager;
@@ -243,6 +245,12 @@ public class CanaryResourceManager<CR extends
AbstractFlinkResource<?, ?>> {
return "true".equalsIgnoreCase(labels.getOrDefault(CANARY_LABEL,
"false"));
}
+ /** Stops the scheduled canary health checks. */
+ @Override
+ public void close() {
+ executorService.shutdownNow();
+ }
+
private class CanaryResourceState {
CR resource;
long bumpedGeneration;
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/OperatorHealthService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/OperatorHealthService.java
index c0e67e8b3..20007fcde 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/OperatorHealthService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/OperatorHealthService.java
@@ -24,7 +24,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/** Health probe service. */
-public class OperatorHealthService {
+public class OperatorHealthService implements AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(OperatorHealthService.class);
private final HealthProbe probe = HealthProbe.INSTANCE;
@@ -52,7 +52,14 @@ public class OperatorHealthService {
}
public void stop() {
- httpBootstrap.stop();
+ if (httpBootstrap != null) {
+ httpBootstrap.stop();
+ }
+ }
+
+ @Override
+ public void close() {
+ stop();
}
public static OperatorHealthService fromConfig(FlinkConfigManager
configManager) {
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
index fee905e8b..2f08646a6 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
@@ -23,13 +23,15 @@ import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.dump.QueryScopeInfo;
import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup;
import org.apache.flink.runtime.metrics.scope.ScopeFormat;
+import org.apache.flink.util.AutoCloseableAsync;
import io.fabric8.kubernetes.client.CustomResource;
import java.util.Map;
/** Base metric group for Flink Operator System metrics. */
-public class KubernetesOperatorMetricGroup extends
AbstractMetricGroup<AbstractMetricGroup<?>> {
+public class KubernetesOperatorMetricGroup extends
AbstractMetricGroup<AbstractMetricGroup<?>>
+ implements AutoCloseable {
protected final String namespace;
protected final String name;
@@ -76,6 +78,14 @@ public class KubernetesOperatorMetricGroup extends
AbstractMetricGroup<AbstractM
hostname);
}
+ @Override
+ public void close() {
+ super.close();
+ if (registry instanceof AutoCloseableAsync closeableRegistry) {
+ closeableRegistry.closeAsync();
+ }
+ }
+
@Override
protected final void putVariables(Map<String, String> variables) {
variables.put(KubernetesOperatorScopeFormat.NAMESPACE, namespace);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java
index 58be43e21..a3a029397 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java
@@ -53,7 +53,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/** Factory for creating the {@link FlinkResourceContext}. */
-public class FlinkResourceContextFactory {
+public class FlinkResourceContextFactory implements AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(FlinkResourceContextFactory.class);
@@ -71,7 +71,7 @@ public class FlinkResourceContextFactory {
private final FlinkConfigManager configManager;
private final ArtifactManager artifactManager;
- private final ExecutorService clientExecutorService;
+ @VisibleForTesting final ExecutorService clientExecutorService;
private final KubernetesOperatorMetricGroup operatorMetricGroup;
private final EventRecorder eventRecorder;
@@ -181,4 +181,10 @@ public class FlinkResourceContextFactory {
lastRecordedExceptionCache.remove(resourceId);
LOG.debug("Removed resource {} from last recorded exception cache",
resourceId);
}
+
+ /** Stops the executor used by the Flink REST cluster clients. */
+ @Override
+ public void close() {
+ clientExecutorService.shutdownNow();
+ }
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
index 715c072b8..e60ad317c 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
@@ -467,4 +467,13 @@ public class FlinkConfigManagerTest {
assertTrue(completed2.get());
assertTrue(completed3.get());
}
+
+ @Test
+ public void testCloseStopsExecutor() {
+ var configManager = new FlinkConfigManager(new Configuration());
+ assertFalse(configManager.executorService.isShutdown());
+
+ configManager.close();
+ assertTrue(configManager.executorService.isShutdown());
+ }
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
index 654476b1b..2878bcb18 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
@@ -37,6 +37,7 @@ import
io.javaoperatorsdk.operator.processing.event.ResourceID;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import java.net.ConnectException;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.Collections;
@@ -47,6 +48,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus.OK;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -81,7 +83,7 @@ public class HealthProbeTest {
assertNotNull(HealthProbe.INSTANCE.getRuntimeInfo());
} finally {
operator.stop();
- assertFalse(callHealthEndpoint(conf));
+ assertThrows(ConnectException.class, () ->
callHealthEndpoint(conf));
}
}
}
@@ -257,4 +259,15 @@ public class HealthProbeTest {
return () -> informers;
}
+
+ @Test
+ public void testCanaryResourceManagerCloseStopsExecutor() {
+ try (var configManager = new FlinkConfigManager(new Configuration())) {
+ var canaryManager = new
CanaryResourceManager<FlinkDeployment>(configManager);
+ assertFalse(canaryManager.executorService.isShutdown());
+
+ canaryManager.close();
+ assertTrue(canaryManager.executorService.isShutdown());
+ }
+ }
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
index 211bb65a4..164f71fd1 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
@@ -29,6 +29,8 @@ import java.util.Map;
import static
org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* @link KubernetesOperatorMetricGroup tests.
@@ -150,6 +152,24 @@ public class KubernetesOperatorMetricGroupTest {
registry.close();
}
+ @Test
+ public void testCloseClosesRegistry() {
+ var configuration = new Configuration();
+ var registry = new
MetricRegistryImpl(fromConfiguration(configuration));
+ var group =
+ KubernetesOperatorMetricGroup.create(
+ registry,
+ configuration,
+ "default",
+ "flink-kubernetes-operator",
+ "localhost");
+ assertFalse(registry.isShutdown());
+
+ group.close();
+ assertTrue(group.isClosed());
+ assertTrue(registry.isShutdown());
+ }
+
private static MetricRegistryConfiguration fromConfiguration(Configuration
configuration) {
return MetricRegistryConfiguration.fromConfiguration(configuration,
Long.MAX_VALUE);
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactoryTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactoryTest.java
index 1eb8fe4b1..ee464d142 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactoryTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactoryTest.java
@@ -25,6 +25,8 @@ import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptio
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/** Tests for {@link FlinkResourceContextFactory}. */
public class FlinkResourceContextFactoryTest {
@@ -56,4 +58,16 @@ public class FlinkResourceContextFactoryTest {
TestUtils.createTestMetricGroup(new
Configuration()),
null));
}
+
+ @Test
+ public void testCloseStopsClientExecutor() {
+ try (var configManager = new FlinkConfigManager(new Configuration());
+ var metricGroup = TestUtils.createTestMetricGroup(new
Configuration())) {
+ var contextFactory = new
FlinkResourceContextFactory(configManager, metricGroup, null);
+ assertFalse(contextFactory.clientExecutorService.isShutdown());
+
+ contextFactory.close();
+ assertTrue(contextFactory.clientExecutorService.isShutdown());
+ }
+ }
}
diff --git
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
index 673f9c9a8..cde403afb 100644
---
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
+++
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
@@ -32,6 +32,7 @@ import
org.apache.flink.kubernetes.operator.utils.KubernetesClientUtils;
import org.apache.flink.kubernetes.operator.utils.MutatorUtils;
import org.apache.flink.kubernetes.operator.utils.OperatorPluginUtils;
import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
+import org.apache.flink.util.IOUtils;
import org.apache.flink.shaded.netty4.io.netty.bootstrap.ServerBootstrap;
import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
@@ -53,6 +54,8 @@ import javax.annotation.Nullable;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.file.Path;
+import java.util.ArrayDeque;
+import java.util.Deque;
import java.util.Set;
/** Main Class for Flink native k8s operator. */
@@ -67,17 +70,23 @@ public class FlinkOperatorWebhook {
@VisibleForTesting final Set<FlinkResourceMutator> mutators;
@VisibleForTesting final AdmissionHandler admissionHandler;
+ private final Deque<AutoCloseable> ownedResources = new ArrayDeque<>();
+
@VisibleForTesting
FlinkOperatorWebhook(
@Nullable InformerManager informerManager, @Nullable
FlinkConfigManager configManager) {
if (informerManager == null) {
- informerManager = new InformerManager(new
KubernetesClientBuilder().build());
+ var kubernetesClient = new KubernetesClientBuilder().build();
+ ownedResources.push(kubernetesClient);
+ informerManager = new InformerManager(kubernetesClient);
+ ownedResources.push(informerManager);
}
if (configManager == null) {
configManager =
new FlinkConfigManager(
informerManager::setNamespaces,
KubernetesClientUtils.isCrdInstalled(FlinkStateSnapshot.class));
+ ownedResources.push(configManager);
}
var operatorConfig = configManager.getOperatorConfiguration();
@@ -119,12 +128,23 @@ public class FlinkOperatorWebhook {
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
+ try {
+ IOUtils.closeAll(ownedResources);
+ } catch (Exception e) {
+ LOG.warn("Failed to close webhook resources", e);
+ }
}
}
- public static void main(String[] args) throws Exception {
+ public static void main(String[] args) {
EnvUtils.logEnvironmentInfo(LOG, "Flink Kubernetes Webhook", args);
- new FlinkOperatorWebhook(null, null).run();
+ try {
+ new FlinkOperatorWebhook(null, null).run();
+ } catch (Throwable t) {
+ // Non-daemon threads started before the failure would keep the
JVM alive
+ LOG.error("Failed to start the webhook", t);
+ System.exit(1);
+ }
}
private static int getPort() {
diff --git
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManager.java
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManager.java
index d2b69d3cf..50b84cbea 100644
---
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManager.java
+++
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManager.java
@@ -36,7 +36,7 @@ import java.util.concurrent.ConcurrentHashMap;
import static
io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_NAMESPACES_SET;
/** The manager of the informers. */
-public class InformerManager {
+public class InformerManager implements AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(InformerManager.class);
private final Set<String> watchedNamespaces =
ConcurrentHashMap.newKeySet();
@@ -138,6 +138,15 @@ public class InformerManager {
LOG.info("Setting namespaces to {}", watchedNamespaces);
this.watchedNamespaces.clear();
this.watchedNamespaces.addAll(watchedNamespaces);
+ stopInformers();
+ }
+
+ @Override
+ public void close() {
+ stopInformers();
+ }
+
+ private void stopInformers() {
if (flinkDepInformers != null) {
synchronized (this) {
if (flinkDepInformers != null) {
diff --git
a/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManagerTest.java
b/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManagerTest.java
index 1f167fe26..591f4a4d2 100644
---
a/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManagerTest.java
+++
b/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManagerTest.java
@@ -72,6 +72,20 @@ public class InformerManagerTest {
Assertions.assertNotNull(informerManager.getFlinkSessionJobInformer("ns3"));
}
+ @Test
+ public void testCloseStopsInformers() {
+ var informerManager = new InformerManager(kubernetesClient);
+ informerManager.setNamespaces(Set.of("ns1"));
+ var depInformer = informerManager.getFlinkDepInformer("ns1");
+ var sessionJobInformer =
informerManager.getFlinkSessionJobInformer("ns1");
+ Assertions.assertTrue(depInformer.isRunning());
+ Assertions.assertTrue(sessionJobInformer.isRunning());
+
+ informerManager.close();
+ Assertions.assertFalse(depInformer.isRunning());
+ Assertions.assertFalse(sessionJobInformer.isRunning());
+ }
+
@Test
public void testSetNamespacesStopsSessionJobInformers() {
var informerManager = new InformerManager(kubernetesClient);
