OSS909 -- Make fetcher thread pool separate and pluggable (#282)

## Summary

Manual cherry-pick from https://github.com/airbnb/chronon/pull/909/files

Should be rebased and tested off of
#276

## Checklist
- [ ] Added Unit Tests
- [ ] Covered by existing CI
- [ ] Integration tested
- [ ] Documentation update
- [x] Not yet tested, should test after merging 276



<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Added a builder pattern for `JavaFetcher` to enhance object creation
flexibility.
- Introduced an optional `ExecutionContext` parameter across multiple
classes to improve execution context management.

- **Improvements**
- Updated constructor signatures in Java and Scala classes to support
more configurable execution contexts.
- Enhanced instantiation process with more structured object creation
methods.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: ezvz <[email protected]>
Co-authored-by: Nikhil Simha <[email protected]>
Co-authored-by: Thomas Chow <[email protected]>

Author: 4 people

online/src/main/java/ai/chronon/online/JavaFetcher.java

@@ -24,6 +24,7 @@
 import scala.collection.mutable.ArrayBuffer;
 import scala.compat.java8.FutureConverters;
 import scala.concurrent.Future;
+import scala.concurrent.ExecutionContext;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -35,15 +36,87 @@ public class JavaFetcher {
   Fetcher fetcher;
 
   public JavaFetcher(KVStore kvStore, String metaDataSet, Long timeoutMillis, Consumer<LoggableResponse> logFunc, ExternalSourceRegistry registry, String callerName, Boolean disableErrorThrows) {
-    this.fetcher = new Fetcher(kvStore, metaDataSet, timeoutMillis, logFunc, false, registry, callerName, null, disableErrorThrows);
+    this.fetcher = new Fetcher(kvStore, metaDataSet, timeoutMillis, logFunc, false, registry, callerName, null, disableErrorThrows, null);
   }
 
   public JavaFetcher(KVStore kvStore, String metaDataSet, Long timeoutMillis, Consumer<LoggableResponse> logFunc, ExternalSourceRegistry registry) {
-    this.fetcher = new Fetcher(kvStore, metaDataSet, timeoutMillis, logFunc, false, registry, null, null, false);
+    this.fetcher = new Fetcher(kvStore, metaDataSet, timeoutMillis, logFunc, false, registry, null, null, false, null);
   }
 
   public JavaFetcher(KVStore kvStore, String metaDataSet, Long timeoutMillis, Consumer<LoggableResponse> logFunc, ExternalSourceRegistry registry, String callerName, FlagStore flagStore, Boolean disableErrorThrows) {
-    this.fetcher = new Fetcher(kvStore, metaDataSet, timeoutMillis, logFunc, false, registry, callerName, flagStore, disableErrorThrows);
+    this.fetcher = new Fetcher(kvStore, metaDataSet, timeoutMillis, logFunc, false, registry, callerName, flagStore, disableErrorThrows, null);
+  }
+
+    /* user builder pattern to create JavaFetcher
+    example way to create the java fetcher
+    JavaFetcher fetcher = new JavaFetcher.Builder(kvStore, metaDataSet, timeoutMillis, logFunc, registry)
+                                        .callerName(callerName)
+                                        .flagStore(flagStore)
+                                        .disableErrorThrows(disableErrorThrows)
+                                        .build();
+     */
+  private JavaFetcher(Builder builder) {
+    this.fetcher = new Fetcher(builder.kvStore,
+            builder.metaDataSet,
+            builder.timeoutMillis,
+            builder.logFunc,
+            builder.debug,
+            builder.registry,
+            builder.callerName,
+            builder.flagStore,
+            builder.disableErrorThrows,
+            builder.executionContextOverride);
+  }
+
+  public static class Builder {
+    private KVStore kvStore;
+    private String metaDataSet;
+    private Long timeoutMillis;
+    private Consumer<LoggableResponse> logFunc;
+    private ExternalSourceRegistry registry;
+    private String callerName;
+    private Boolean debug;
+    private FlagStore flagStore;
+    private Boolean disableErrorThrows;
+    private ExecutionContext executionContextOverride;
+
+    public Builder(KVStore kvStore, String metaDataSet, Long timeoutMillis,
+                   Consumer<LoggableResponse> logFunc, ExternalSourceRegistry registry) {
+      this.kvStore = kvStore;
+      this.metaDataSet = metaDataSet;
+      this.timeoutMillis = timeoutMillis;
+      this.logFunc = logFunc;
+      this.registry = registry;
+    }
+
+    public Builder callerName(String callerName) {
+      this.callerName = callerName;
+      return this;
+    }
+    
+    public Builder flagStore(FlagStore flagStore) {
+      this.flagStore = flagStore;
+      return this;
+    }
+    
+    public Builder disableErrorThrows(Boolean disableErrorThrows) {
+      this.disableErrorThrows = disableErrorThrows;
+      return this;
+    }
+    
+    public Builder debug(Boolean debug) {
+      this.debug = debug;
+      return this;
+    }
+
+    public Builder executionContextOverride(ExecutionContext executionContextOverride) {
+      this.executionContextOverride = executionContextOverride;
+      return this;
+    }
+
+    public JavaFetcher build() {
+      return new JavaFetcher(this);
+    }
   }
 
 

online/src/main/scala/ai/chronon/online/Fetcher.scala

@@ -43,6 +43,7 @@ import scala.collection.JavaConverters._
 import scala.collection.Seq
 import scala.collection.mutable
 import scala.collection.mutable.ListBuffer
+import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.util.Failure
 import scala.util.Success
@@ -95,8 +96,15 @@ class Fetcher(val kvStore: KVStore,
               val externalSourceRegistry: ExternalSourceRegistry = null,
               callerName: String = null,
               flagStore: FlagStore = null,
-              disableErrorThrows: Boolean = false)
-    extends FetcherBase(kvStore, metaDataSet, timeoutMillis, debug, flagStore, disableErrorThrows) {
+              disableErrorThrows: Boolean = false,
+              executionContextOverride: ExecutionContext = null)
+    extends FetcherBase(kvStore,
+                        metaDataSet,
+                        timeoutMillis,
+                        debug,
+                        flagStore,
+                        disableErrorThrows,
+                        executionContextOverride) {
 
   private def reportCallerNameFetcherVersion(): Unit = {
     val message = s"CallerName: ${Option(callerName).getOrElse("N/A")}, FetcherVersion: ${BuildInfo.version}"

online/src/main/scala/ai/chronon/online/FetcherBase.scala

@@ -44,6 +44,7 @@ import com.google.gson.Gson
 import java.util
 import scala.collection.JavaConverters._
 import scala.collection.Seq
+import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.util.Failure
 import scala.util.Success
@@ -58,8 +59,9 @@ class FetcherBase(kvStore: KVStore,
                   timeoutMillis: Long = 10000,
                   debug: Boolean = false,
                   flagStore: FlagStore = null,
-                  disableErrorThrows: Boolean = false)
-    extends MetadataStore(kvStore, metaDataSet, timeoutMillis)
+                  disableErrorThrows: Boolean = false,
+                  executionContextOverride: ExecutionContext = null)
+    extends MetadataStore(kvStore, metaDataSet, timeoutMillis, executionContextOverride)
     with FetcherCache {
   import FetcherBase._
 

online/src/main/scala/ai/chronon/online/MetadataStore.scala

@@ -41,7 +41,10 @@ import scala.util.Try
 // [timestamp -> {metric name -> metric value}]
 case class DataMetrics(series: Seq[(Long, SortedMap[String, Any])])
 
-class MetadataStore(kvStore: KVStore, val dataset: String = MetadataDataset, timeoutMillis: Long) {
+class MetadataStore(kvStore: KVStore,
+                    val dataset: String = MetadataDataset,
+                    timeoutMillis: Long,
+                    executionContextOverride: ExecutionContext = null) {
   @transient implicit lazy val logger: Logger = LoggerFactory.getLogger(getClass)
   private var partitionSpec = PartitionSpec(format = "yyyy-MM-dd", spanMillis = WindowUtils.Day.millis)
   private val CONF_BATCH_SIZE = 50
@@ -56,7 +59,8 @@ class MetadataStore(kvStore: KVStore, val dataset: String = MetadataDataset, tim
     partitionSpec = PartitionSpec(format = format, spanMillis = partitionSpec.spanMillis)
   }
 
-  implicit val executionContext: ExecutionContext = kvStore.executionContext
+  implicit val executionContext: ExecutionContext =
+    Option(executionContextOverride).getOrElse(FlexibleExecutionContext.buildExecutionContext)
 
   def getConf[T <: TBase[_, _]: Manifest](confPathOrName: String): Try[T] = {
     val clazz = implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]]