wangyang0918 commented on a change in pull request #16607:
URL: https://github.com/apache/flink/pull/16607#discussion_r677359438
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesSharedInformer.java
##########
@@ -47,47 +46,37 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
/** Base class for shared watcher based on {@link SharedIndexInformer}. */
-public abstract class KubernetesSharedInformer<
- T extends HasMetadata, TList extends
KubernetesResourceList<T>, R>
+public abstract class KubernetesSharedInformer<T extends HasMetadata, R>
implements KubernetesSharedWatcher<R> {
protected final Logger log = LoggerFactory.getLogger(getClass());
private final NamespacedKubernetesClient client;
- private final SharedInformerFactory sharedInformerFactory;
private final SharedIndexInformer<T> sharedIndexInformer;
private final Function<T, R> eventWrapper;
+ private final ExecutorService informerExecutor;
+
private final AggregatedEventHandler aggregatedEventHandler;
public KubernetesSharedInformer(
NamespacedKubernetesClient client,
- Class<T> apiTypeClass,
- Class<TList> apiListTypeClass,
- Map<String, String> labels,
+ Informable<T> informable,
Function<T, R> eventWrapper) {
- Preconditions.checkArgument(
- !CollectionUtil.isNullOrEmpty(labels), "Labels must not be
null or empty");
this.client = client;
- final ExecutorService executorService =
+
+ informerExecutor =
Executors.newSingleThreadExecutor(
new
ExecutorThreadFactory("KubernetesClient-Informer"));
- this.sharedInformerFactory = client.informers(executorService);
- this.sharedInformerFactory.withLabels(labels);
Review comment:
`SharedInformerFactory` no longer inherits from `BaseOperation` and
suggest to use `Informable` instead. That's why I introduce such chanages.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
