feat: moving cache sync timeout to controller level (#1576)

csviri · csviri · commit f3e42c254195 · 2022-10-31T09:39:32.000+01:00
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java
@@ -252,6 +252,12 @@ public List<DependentResourceSpec> getDependentResources() {
     return specs;
   }
 
+  @Override
+  public Duration cacheSyncTimeout() {
+    var cacheSyncTimeout = annotation.cacheSyncTimeout();
+    return Duration.of(cacheSyncTimeout.timeout(), cacheSyncTimeout.timeUnit().toChronoUnit());
+  }
+
   private String getName(Dependent dependent, Class<? extends DependentResource> dependentType) {
     var name = dependent.name();
     if (name.isBlank()) {
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java
@@ -1,6 +1,5 @@
 package io.javaoperatorsdk.operator.api.config;
 
-import java.time.Duration;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -169,15 +168,6 @@ default boolean stopOnInformerErrorDuringStartup() {
     return true;
   }
 
-  /**
-   * Timeout for cache sync in milliseconds. In other words source start timeout. Note that is
-   * "stopOnInformerErrorDuringStartup" is true the operator will stop on timeout. Default is 2
-   * minutes.
-   */
-  default Duration cacheSyncTimeout() {
-    return Duration.ofMinutes(2);
-  }
-
   /**
    * Handler for an informer stop. Informer stops if there is a non-recoverable error. Like received
    * a resource that cannot be deserialized.
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java
@@ -1,6 +1,5 @@
 package io.javaoperatorsdk.operator.api.config;
 
-import java.time.Duration;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -29,7 +28,6 @@ public class ConfigurationServiceOverrider {
   private LeaderElectionConfiguration leaderElectionConfiguration;
   private InformerStoppedHandler informerStoppedHandler;
   private Boolean stopOnInformerErrorDuringStartup;
-  private Duration cacheSyncTimeout;
 
   ConfigurationServiceOverrider(ConfigurationService original) {
     this.original = original;
@@ -108,11 +106,6 @@ public ConfigurationServiceOverrider withStopOnInformerErrorDuringStartup(
     return this;
   }
 
-  public ConfigurationServiceOverrider withCacheSyncTimeout(Duration cacheSyncTimeout) {
-    this.cacheSyncTimeout = cacheSyncTimeout;
-    return this;
-  }
-
   public ConfigurationService build() {
     return new BaseConfigurationService(original.getVersion(), cloner, objectMapper) {
       @Override
@@ -191,11 +184,6 @@ public boolean stopOnInformerErrorDuringStartup() {
         return stopOnInformerErrorDuringStartup != null ? stopOnInformerErrorDuringStartup
             : super.stopOnInformerErrorDuringStartup();
       }
-
-      @Override
-      public Duration cacheSyncTimeout() {
-        return cacheSyncTimeout != null ? cacheSyncTimeout : super.cacheSyncTimeout();
-      }
     };
   }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfiguration.java
@@ -17,7 +17,7 @@
 import io.javaoperatorsdk.operator.processing.retry.GradualRetry;
 import io.javaoperatorsdk.operator.processing.retry.Retry;
 
-public interface ControllerConfiguration<R extends HasMetadata> extends ResourceConfiguration<R> {
+public interface ControllerConfiguration<P extends HasMetadata> extends ResourceConfiguration<P> {
 
   @SuppressWarnings("rawtypes")
   RateLimiter DEFAULT_RATE_LIMITER = LinearRateLimiter.deactivatedRateLimiter();
@@ -71,7 +71,7 @@ default RateLimiter getRateLimiter() {
    *
    * @return filter
    */
-  default ResourceEventFilter<R> getEventFilter() {
+  default ResourceEventFilter<P> getEventFilter() {
     return ResourceEventFilters.passthrough();
   }
 
@@ -91,8 +91,8 @@ default ConfigurationService getConfigurationService() {
 
   @SuppressWarnings("unchecked")
   @Override
-  default Class<R> getResourceClass() {
-    return (Class<R>) Utils.getFirstTypeArgumentFromSuperClassOrInterface(getClass(),
+  default Class<P> getResourceClass() {
+    return (Class<P>) Utils.getFirstTypeArgumentFromSuperClassOrInterface(getClass(),
         ControllerConfiguration.class);
   }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java
@@ -38,6 +38,7 @@ public class ControllerConfigurationOverrider<R extends HasMetadata> {
   private OnUpdateFilter<R> onUpdateFilter;
   private GenericFilter<R> genericFilter;
   private RateLimiter rateLimiter;
+  private Duration cacheSyncTimeout;
 
   private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     finalizer = original.getFinalizerName();
@@ -56,6 +57,7 @@ private ControllerConfigurationOverrider(ControllerConfiguration<R> original) {
     dependentResources.forEach(drs -> namedDependentResourceSpecs.put(drs.getName(), drs));
     this.original = original;
     this.rateLimiter = original.getRateLimiter();
+    this.cacheSyncTimeout = original.cacheSyncTimeout();
   }
 
   public ControllerConfigurationOverrider<R> withFinalizer(String finalizer) {
@@ -176,6 +178,11 @@ public ControllerConfigurationOverrider<R> replacingNamedDependentResourceConfig
     return this;
   }
 
+  public ControllerConfigurationOverrider<R> withCacheSyncTimeout(Duration cacheSyncTimeout) {
+    this.cacheSyncTimeout = cacheSyncTimeout;
+    return this;
+  }
+
   public ControllerConfiguration<R> build() {
     final var hasModifiedNamespaces = !original.getNamespaces().equals(namespaces);
     final var newDependentSpecs = namedDependentResourceSpecs.values().stream()
@@ -208,6 +215,7 @@ public ControllerConfiguration<R> build() {
         onUpdateFilter,
         genericFilter,
         rateLimiter,
+        cacheSyncTimeout,
         newDependentSpecs);
   }
 
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/DefaultControllerConfiguration.java
@@ -31,6 +31,7 @@ public class DefaultControllerConfiguration<R extends HasMetadata>
   private final List<DependentResourceSpec> dependents;
   private final Duration reconciliationMaxInterval;
   private final RateLimiter rateLimiter;
+  private final Duration cacheSyncTimeout;
 
   // NOSONAR constructor is meant to provide all information
   public DefaultControllerConfiguration(
@@ -49,6 +50,7 @@ public DefaultControllerConfiguration(
       OnUpdateFilter<R> onUpdateFilter,
       GenericFilter<R> genericFilter,
       RateLimiter rateLimiter,
+      Duration cacheSyncTimeout,
       List<DependentResourceSpec> dependents) {
     super(labelSelector, resourceClass, onAddFilter, onUpdateFilter, genericFilter, namespaces);
     this.associatedControllerClassName = associatedControllerClassName;
@@ -65,6 +67,7 @@ public DefaultControllerConfiguration(
     this.rateLimiter =
         rateLimiter != null ? rateLimiter : LinearRateLimiter.deactivatedRateLimiter();
     this.dependents = dependents != null ? dependents : Collections.emptyList();
+    this.cacheSyncTimeout = cacheSyncTimeout;
   }
 
   @Override
@@ -116,4 +119,8 @@ public Optional<Duration> maxReconciliationInterval() {
   public RateLimiter getRateLimiter() {
     return rateLimiter;
   }
+
+  public Duration cacheSyncTimeout() {
+    return cacheSyncTimeout;
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ResourceConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ResourceConfiguration.java
@@ -1,12 +1,14 @@
 package io.javaoperatorsdk.operator.api.config;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Optional;
 import java.util.Set;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.ReconcilerUtils;
+import io.javaoperatorsdk.operator.api.reconciler.CacheSyncTimeout;
 import io.javaoperatorsdk.operator.api.reconciler.Constants;
 import io.javaoperatorsdk.operator.processing.event.source.filter.GenericFilter;
 import io.javaoperatorsdk.operator.processing.event.source.filter.OnAddFilter;
@@ -108,4 +110,13 @@ default Set<String> getEffectiveNamespaces() {
     }
     return targetNamespaces;
   }
+
+  /**
+   * Timeout for cache sync. In other words event source start timeout. Note that is
+   * "stopOnInformerErrorDuringStartup" is true the operator will stop on timeout. Default is 2
+   * minutes.
+   */
+  default Duration cacheSyncTimeout() {
+    return Duration.ofMinutes(CacheSyncTimeout.DEFAULT_TIMEOUT);
+  }
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/CacheSyncTimeout.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/CacheSyncTimeout.java
@@ -0,0 +1,19 @@
+package io.javaoperatorsdk.operator.api.reconciler;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.util.concurrent.TimeUnit;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE})
+public @interface CacheSyncTimeout {
+
+  int DEFAULT_TIMEOUT = 2;
+
+  int timeout();
+
+  TimeUnit timeUnit() default TimeUnit.MINUTES;
+
+}
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ControllerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/ControllerConfiguration.java
@@ -118,4 +118,7 @@ MaxReconciliationInterval maxReconciliationInterval() default @MaxReconciliation
    *         accessible no-arg constructor.
    */
   Class<? extends RateLimiter> rateLimiter() default LinearRateLimiter.class;
+
+  CacheSyncTimeout cacheSyncTimeout() default @CacheSyncTimeout(
+      timeout = CacheSyncTimeout.DEFAULT_TIMEOUT);
 }
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java
@@ -48,6 +48,7 @@ public EventSourceManager(Controller<P> controller) {
 
   public void postProcessDefaultEventSourcesAfterProcessorInitializer() {
     eventSources.controllerResourceEventSource().setEventHandler(controller.getEventProcessor());
+
     eventSources.retryEventSource().setEventHandler(controller.getEventProcessor());
   }
 
@@ -119,6 +120,7 @@ public final void registerEventSource(EventSource eventSource) throws OperatorEx
     registerEventSource(null, eventSource);
   }
 
+  @SuppressWarnings("unchecked")
   public final synchronized void registerEventSource(String name, EventSource eventSource)
       throws OperatorException {
     Objects.requireNonNull(eventSource, "EventSource must not be null");
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
@@ -97,7 +97,8 @@ public void changeNamespaces(Set<String> namespaces) {
   private InformerWrapper<T> createEventSource(
       FilterWatchListDeletable<T, KubernetesResourceList<T>, Resource<T>> filteredBySelectorClient,
       ResourceEventHandler<T> eventHandler, String key) {
-    var source = new InformerWrapper<>(filteredBySelectorClient.runnableInformer(0));
+    var source = new InformerWrapper<>(filteredBySelectorClient.runnableInformer(0),
+        configuration);
     source.addEventHandler(eventHandler);
     sources.put(key, source);
     return source;
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java
@@ -21,6 +21,7 @@
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.ReconcilerUtils;
 import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
+import io.javaoperatorsdk.operator.api.config.ResourceConfiguration;
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
@@ -32,10 +33,13 @@ class InformerWrapper<T extends HasMetadata>
 
   private final SharedIndexInformer<T> informer;
   private final Cache<T> cache;
+  private final ResourceConfiguration<?> resourceConfiguration;
 
-  public InformerWrapper(SharedIndexInformer<T> informer) {
+  public InformerWrapper(SharedIndexInformer<T> informer,
+      ResourceConfiguration<?> resourceConfiguration) {
     this.informer = informer;
     this.cache = (Cache<T>) informer.getStore();
+    this.resourceConfiguration = resourceConfiguration;
   }
 
   @Override
@@ -69,7 +73,7 @@ public void start() throws OperatorException {
         // note that in case we don't put here timeout and stopOnInformerErrorDuringStartup is
         // false, and there is a rbac issue the get never returns; therefore operator never really
         // starts
-        start.toCompletableFuture().get(configService.cacheSyncTimeout().toMillis(),
+        start.toCompletableFuture().get(resourceConfiguration.cacheSyncTimeout().toMillis(),
             TimeUnit.MILLISECONDS);
       } catch (TimeoutException | ExecutionException e) {
         if (configService.stopOnInformerErrorDuringStartup()) {
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/ControllerManagerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/ControllerManagerTest.java
@@ -58,7 +58,7 @@ private static class TestControllerConfiguration<R extends HasMetadata>
     public TestControllerConfiguration(Reconciler<R> controller, Class<R> crClass) {
       super(null, getControllerName(controller),
           CustomResource.getCRDName(crClass), null, false, null, null, null, null, crClass,
-          null, null, null, null, null, null);
+          null, null, null, null, null, null, null);
       this.controller = controller;
     }
 
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/ReconciliationDispatcherTest.java
@@ -25,14 +25,7 @@
 import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
 import io.javaoperatorsdk.operator.api.config.MockControllerConfiguration;
-import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
-import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
-import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
-import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
-import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
-import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.*;
 import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.CustomResourceFacade;
 import io.javaoperatorsdk.operator.processing.retry.GenericRetry;
@@ -106,7 +99,8 @@ private <R extends HasMetadata> ReconciliationDispatcher<R> init(R customResourc
     final Class<R> resourceClass = (Class<R>) customResource.getClass();
     configuration = configuration == null ? MockControllerConfiguration.forResource(resourceClass)
         : configuration;
-
+    when(configuration.cacheSyncTimeout())
+        .thenReturn(Duration.ofMinutes(CacheSyncTimeout.DEFAULT_TIMEOUT));
     when(configuration.getFinalizerName()).thenReturn(DEFAULT_FINALIZER);
     when(configuration.getName()).thenReturn("EventDispatcherTestController");
     when(configuration.getResourceClass()).thenReturn(resourceClass);
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilterTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/ResourceEventFilterTest.java
@@ -145,7 +145,7 @@ public ControllerConfig(String finalizer, boolean generationAware,
           eventFilter,
           customResourceClass,
           null,
-          null, null, null, null, null);
+          null, null, null, null, null, null);
     }
   }
 
diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/controller/ControllerResourceEventSourceTest.java
@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.processing.event.source.controller;
 
+import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.List;
 
@@ -9,6 +10,7 @@
 import io.javaoperatorsdk.operator.MockKubernetesClient;
 import io.javaoperatorsdk.operator.TestUtils;
 import io.javaoperatorsdk.operator.api.config.DefaultControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.CacheSyncTimeout;
 import io.javaoperatorsdk.operator.processing.Controller;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
@@ -188,7 +190,9 @@ public TestConfiguration(boolean generationAware, OnAddFilter<TestCustomResource
           null,
           TestCustomResource.class,
           null,
-          onAddFilter, onUpdateFilter, genericFilter, null, null);
+          onAddFilter, onUpdateFilter, genericFilter, null,
+          Duration.ofMinutes(CacheSyncTimeout.DEFAULT_TIMEOUT),
+          null);
     }
   }
 }
diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/InformerRelatedBehaviorITS.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/InformerRelatedBehaviorITS.java
@@ -205,12 +205,15 @@ Operator startOperator(boolean stopOnInformerErrorDuringStartup, boolean addStop
     Operator operator = new Operator(clientUsingServiceAccount(),
         co -> {
           co.withStopOnInformerErrorDuringStartup(stopOnInformerErrorDuringStartup);
-          co.withCacheSyncTimeout(Duration.ofMillis(3000));
           if (addStopHandler) {
-            co.withInformerStoppedHandler((informer, ex) -> replacementStopHandlerCalled = true);
+            co.withInformerStoppedHandler((informer, ex) -> {
+              replacementStopHandlerCalled = true;
+            });
           }
         });
-    operator.register(reconciler);
+    operator.register(reconciler, co -> {
+      co.withCacheSyncTimeout(Duration.ofMillis(3000));
+    });
     operator.installShutdownHook();
     operator.start();
     return operator;

Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,7 @@`
`17`	`17`	`import io.javaoperatorsdk.operator.processing.retry.GradualRetry;`
`18`	`18`	`import io.javaoperatorsdk.operator.processing.retry.Retry;`
`19`	`19`
`20`		`-public interface ControllerConfiguration<R extends HasMetadata> extends ResourceConfiguration<R> {`
	`20`	`+public interface ControllerConfiguration<P extends HasMetadata> extends ResourceConfiguration<P> {`
`21`	`21`
`22`	`22`	`@SuppressWarnings("rawtypes")`
`23`	`23`	`RateLimiter DEFAULT_RATE_LIMITER = LinearRateLimiter.deactivatedRateLimiter();`
`@@ -71,7 +71,7 @@ default RateLimiter getRateLimiter() {`
`71`	`71`	`*`
`72`	`72`	`* @return filter`
`73`	`73`	`*/`
`74`		`- default ResourceEventFilter<R> getEventFilter() {`
	`74`	`+ default ResourceEventFilter<P> getEventFilter() {`
`75`	`75`	`return ResourceEventFilters.passthrough();`
`76`	`76`	`}`
`77`	`77`
`@@ -91,8 +91,8 @@ default ConfigurationService getConfigurationService() {`
`91`	`91`
`92`	`92`	`@SuppressWarnings("unchecked")`
`93`	`93`	`@Override`
`94`		`- default Class<R> getResourceClass() {`
`95`		`- return (Class<R>) Utils.getFirstTypeArgumentFromSuperClassOrInterface(getClass(),`
	`94`	`+ default Class<P> getResourceClass() {`
	`95`	`+ return (Class<P>) Utils.getFirstTypeArgumentFromSuperClassOrInterface(getClass(),`
`96`	`96`	`ControllerConfiguration.class);`
`97`	`97`	`}`
`98`	`98`	`}`
Original file line number	Diff line number	Diff line change
`@@ -118,4 +118,7 @@ MaxReconciliationInterval maxReconciliationInterval() default @MaxReconciliation`
`118`	`118`	`* accessible no-arg constructor.`
`119`	`119`	`*/`
`120`	`120`	`Class<? extends RateLimiter> rateLimiter() default LinearRateLimiter.class;`
	`121`	`+`
	`122`	`+ CacheSyncTimeout cacheSyncTimeout() default @CacheSyncTimeout(`
	`123`	`+ timeout = CacheSyncTimeout.DEFAULT_TIMEOUT);`
`121`	`124`	`}`
Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,7 @@ public EventSourceManager(Controller<P> controller) {`
`48`	`48`
`49`	`49`	`public void postProcessDefaultEventSourcesAfterProcessorInitializer() {`
`50`	`50`	`eventSources.controllerResourceEventSource().setEventHandler(controller.getEventProcessor());`
	`51`	`+`
`51`	`52`	`eventSources.retryEventSource().setEventHandler(controller.getEventProcessor());`
`52`	`53`	`}`
`53`	`54`
`@@ -119,6 +120,7 @@ public final void registerEventSource(EventSource eventSource) throws OperatorEx`
`119`	`120`	`registerEventSource(null, eventSource);`
`120`	`121`	`}`
`121`	`122`
	`123`	`+ @SuppressWarnings("unchecked")`
`122`	`124`	`public final synchronized void registerEventSource(String name, EventSource eventSource)`
`123`	`125`	`throws OperatorException {`
`124`	`126`	`Objects.requireNonNull(eventSource, "EventSource must not be null");`