zentol commented on code in PR #22848:
URL: https://github.com/apache/flink/pull/22848#discussion_r1252806552
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServices.java:
##########
@@ -69,75 +59,72 @@ public class
KubernetesMultipleComponentLeaderElectionHaServices extends Abstrac
private final String lockIdentity;
- private final FatalErrorHandler fatalErrorHandler;
-
- @Nullable
- @GuardedBy("lock")
- private DefaultMultipleComponentLeaderElectionService
multipleComponentLeaderElectionService =
- null;
-
KubernetesMultipleComponentLeaderElectionHaServices(
FlinkKubeClient kubeClient,
- Executor executor,
- Configuration config,
- BlobStoreService blobStoreService,
- FatalErrorHandler fatalErrorHandler)
- throws IOException {
+ Executor ioExecutor,
+ Configuration configuration,
+ BlobStoreService blobStoreService)
+ throws Exception {
+ this(
+ kubeClient,
+ kubeClient.createConfigMapSharedWatcher(
+ KubernetesUtils.getConfigMapLabels(
+
configuration.get(KubernetesConfigOptions.CLUSTER_ID),
+ LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY)),
+ Executors.newCachedThreadPool(
+ new ExecutorThreadFactory("config-map-watch-handler")),
+ ioExecutor,
+ configuration.get(KubernetesConfigOptions.CLUSTER_ID),
+ UUID.randomUUID().toString(),
+ configuration,
+ blobStoreService);
+ }
+ private KubernetesMultipleComponentLeaderElectionHaServices(
+ FlinkKubeClient kubeClient,
+ KubernetesConfigMapSharedWatcher configMapSharedWatcher,
+ ExecutorService watchExecutorService,
+ Executor ioExecutor,
+ String clusterId,
+ String lockIdentity,
+ Configuration configuration,
+ BlobStoreService blobStoreService)
+ throws Exception {
super(
- config,
- executor,
+ configuration,
+ createDriverFactory(
+ kubeClient,
+ configMapSharedWatcher,
+ watchExecutorService,
+ clusterId,
+ lockIdentity,
+ configuration),
+ ioExecutor,
blobStoreService,
- FileSystemJobResultStore.fromConfiguration(config));
- this.kubeClient = checkNotNull(kubeClient);
- this.clusterId =
checkNotNull(config.get(KubernetesConfigOptions.CLUSTER_ID));
- this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
+ FileSystemJobResultStore.fromConfiguration(configuration));
- this.configMapSharedWatcher =
- this.kubeClient.createConfigMapSharedWatcher(
- KubernetesUtils.getConfigMapLabels(
- clusterId,
LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
- this.watchExecutorService =
- Executors.newCachedThreadPool(
- new ExecutorThreadFactory("config-map-watch-handler"));
-
- lockIdentity = UUID.randomUUID().toString();
+ this.kubeClient = checkNotNull(kubeClient);
+ this.clusterId = checkNotNull(clusterId);
+ this.configMapSharedWatcher = checkNotNull(configMapSharedWatcher);
+ this.watchExecutorService = checkNotNull(watchExecutorService);
Review Comment:
Where is this executor closed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
