feat: make reconciliation thread number configurable (#400)

Fixes #399.

Author: metacosm

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Operator.java

@@ -141,7 +141,12 @@ public <R extends CustomResource> void register(
 
       CustomResourceCache customResourceCache = new CustomResourceCache(objectMapper);
       DefaultEventHandler defaultEventHandler =
-          new DefaultEventHandler(customResourceCache, dispatcher, controllerName, retry);
+          new DefaultEventHandler(
+              customResourceCache,
+              dispatcher,
+              controllerName,
+              retry,
+              configurationService.concurrentReconciliationThreads());
       DefaultEventSourceManager eventSourceManager =
           new DefaultEventSourceManager(defaultEventHandler, retry != null);
       defaultEventHandler.setEventSourceManager(eventSourceManager);

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java

@@ -55,4 +55,15 @@ default Config getClientConfiguration() {
   default boolean checkCRDAndValidateLocalModel() {
     return true;
   }
+
+  int DEFAULT_RECONCILIATION_THREADS_NUMBER = 5;
+  /**
+   * Retrieves the maximum number of threads the operator can spin out to dispatch reconciliation
+   * requests to controllers
+   *
+   * @return the maximum number of concurrent reconciliation threads
+   */
+  default int concurrentReconciliationThreads() {
+    return DEFAULT_RECONCILIATION_THREADS_NUMBER;
+  }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/DefaultEventHandler.java

@@ -6,6 +6,7 @@
 
 import io.fabric8.kubernetes.client.CustomResource;
 import io.javaoperatorsdk.operator.api.RetryInfo;
+import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import io.javaoperatorsdk.operator.processing.event.DefaultEventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
@@ -44,14 +45,29 @@ public DefaultEventHandler(
       CustomResourceCache customResourceCache,
       EventDispatcher eventDispatcher,
       String relatedControllerName,
-      Retry retry) {
+      Retry retry,
+      int concurrentReconciliationThreads) {
     this.customResourceCache = customResourceCache;
     this.eventDispatcher = eventDispatcher;
     this.retry = retry;
     eventBuffer = new EventBuffer();
     executor =
         new ScheduledThreadPoolExecutor(
-            5, runnable -> new Thread(runnable, "EventHandler-" + relatedControllerName));
+            concurrentReconciliationThreads,
+            runnable -> new Thread(runnable, "EventHandler-" + relatedControllerName));
+  }
+
+  public DefaultEventHandler(
+      CustomResourceCache customResourceCache,
+      EventDispatcher eventDispatcher,
+      String relatedControllerName,
+      Retry retry) {
+    this(
+        customResourceCache,
+        eventDispatcher,
+        relatedControllerName,
+        retry,
+        ConfigurationService.DEFAULT_RECONCILIATION_THREADS_NUMBER);
   }
 
   public void setEventSourceManager(DefaultEventSourceManager eventSourceManager) {

operator-framework-spring-boot-starter/src/main/java/io/javaoperatorsdk/operator/springboot/starter/OperatorAutoConfiguration.java

@@ -134,4 +134,9 @@ public RetryConfiguration getRetryConfiguration() {
           .orElse(RetryConfiguration.DEFAULT);
     }
   }
+
+  @Override
+  public int concurrentReconciliationThreads() {
+    return configuration.getConcurrentReconciliationThreads();
+  }
 }

operator-framework-spring-boot-starter/src/main/java/io/javaoperatorsdk/operator/springboot/starter/OperatorConfigurationProperties.java

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.springboot.starter;
 
+import io.javaoperatorsdk.operator.api.config.ConfigurationService;
 import java.util.Collections;
 import java.util.Map;
 import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -10,6 +11,8 @@ public class OperatorConfigurationProperties {
   private KubernetesClientProperties client = new KubernetesClientProperties();
   private Map<String, ControllerProperties> controllers = Collections.emptyMap();
   private boolean checkCrdAndValidateLocalModel = true;
+  private int concurrentReconciliationThreads =
+      ConfigurationService.DEFAULT_RECONCILIATION_THREADS_NUMBER;
 
   public KubernetesClientProperties getClient() {
     return client;
@@ -34,4 +37,12 @@ public boolean getCheckCrdAndValidateLocalModel() {
   public void setCheckCrdAndValidateLocalModel(boolean checkCrdAndValidateLocalModel) {
     this.checkCrdAndValidateLocalModel = checkCrdAndValidateLocalModel;
   }
+
+  public int getConcurrentReconciliationThreads() {
+    return concurrentReconciliationThreads;
+  }
+
+  public void setConcurrentReconciliationThreads(int concurrentReconciliationThreads) {
+    this.concurrentReconciliationThreads = concurrentReconciliationThreads;
+  }
 }