Enable batch IR caching by default + fix vertx bt startup issue (#435)

## Summary
Enable batch IR caching by default & fix an issue where our Vertx init
code tries to connect to BT at startup and takes a second or two on the
worker threads (and results in the warning - 'Thread
Thread[vert.x-eventloop-thread-1,5,main] has been blocked for 2976 ms,
time limit is 2000 ms').

## Checklist
- [ ] Added Unit Tests
- [ ] Covered by existing CI
- [X] Integration tested
- [ ] Documentation update



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

- **Refactor**
- Streamlined caching configuration and logic with a consistent default
setting for improved behavior.
- Enhanced service startup by shifting to asynchronous initialization
with better error handling for a more robust launch.

- **Tests**
- Removed an outdated test case that validated previous caching
behavior.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: piyush-zlai

online/src/main/scala/ai/chronon/online/fetcher/FetchContext.scala

@@ -19,11 +19,6 @@ case class FetchContext(kvStore: KVStore,
       .exists(_.asInstanceOf[Boolean])
   }
 
-  def isCachingEnabled(groupByName: String): Boolean = {
-    Option(flagStore)
-      .exists(_.isSet("enable_fetcher_batch_ir_cache", Map("group_by_streaming_dataset" -> groupByName).toJava))
-  }
-
   def shouldStreamingDecodeThrow(groupByName: String): Boolean = {
     Option(flagStore)
       .exists(

online/src/main/scala/ai/chronon/online/fetcher/FetcherCache.scala

@@ -26,12 +26,19 @@ trait FetcherCache {
   @transient private lazy val logger: Logger = LoggerFactory.getLogger(getClass)
 
   val batchIrCacheName = "batch_cache"
-  val maybeBatchIrCache: Option[BatchIrCache] =
+  val defaultBatchIrCacheSize = "10000"
+
+  val configuredBatchIrCacheSize: Option[Int] =
     Option(System.getProperty("ai.chronon.fetcher.batch_ir_cache_size_elements"))
-      .map(size => new BatchIrCache(batchIrCacheName, size.toInt))
-      .orElse(None)
+      .orElse(Some(defaultBatchIrCacheSize))
+      .map(_.toInt)
+      .filter(_ > 0)
+
+  val maybeBatchIrCache: Option[BatchIrCache] =
+    configuredBatchIrCacheSize
+      .map(size => new BatchIrCache(batchIrCacheName, size))
 
-  // Caching needs to be configured globally
+  // Caching needs to be configured globally with a cache size > 0
   def isCacheSizeConfigured: Boolean = maybeBatchIrCache.isDefined
 
   // Caching needs to be enabled for the specific groupBy

online/src/main/scala/ai/chronon/online/fetcher/GroupByFetcher.scala

@@ -27,16 +27,16 @@ class GroupByFetcher(fetchContext: FetchContext, metadataStore: MetadataStore)
   @transient private implicit lazy val logger: Logger = LoggerFactory.getLogger(getClass)
 
   override def isCachingEnabled(groupBy: GroupBy): Boolean = {
-    if (!isCacheSizeConfigured || groupBy.getMetaData == null || groupBy.getMetaData.getName == null) return false
-
-    val gbName = groupBy.getMetaData.getName
-
-    val isCachingFlagEnabled = fetchContext.isCachingEnabled(gbName)
-
-    if (fetchContext.debug)
-      logger.info(s"Online IR caching is ${if (isCachingFlagEnabled) "enabled" else "disabled"} for $gbName")
+    if (fetchContext.debug) {
+      configuredBatchIrCacheSize match {
+        case Some(cacheSize) =>
+          logger.info(s"Online IR caching is enabled with cache size = $cacheSize")
+        case None =>
+          logger.info("Online IR caching is disabled")
+      }
+    }
 
-    isCachingFlagEnabled
+    isCacheSizeConfigured
   }
 
   /** Convert a groupBy request into a batch kv request and optionally a streaming kv request

online/src/test/scala/ai/chronon/online/test/FetcherBaseTest.scala

@@ -203,35 +203,6 @@ class FetcherBaseTest extends AnyFlatSpec with MockitoSugar with Matchers with M
     verify(ttlCache, never()).apply(any())
   }
 
-  it should "determine if caching is enabled correctly" in {
-    val flagStore: FlagStore = (flagName: String, attributes: java.util.Map[String, String]) => {
-      flagName match {
-        case "enable_fetcher_batch_ir_cache" =>
-          attributes.get("group_by_streaming_dataset") match {
-            case "test_groupby_2" => false
-            case "test_groupby_3" => true
-            case other @ _ =>
-              fail(s"Unexpected group_by_streaming_dataset: $other")
-              false
-          }
-        case _ => false
-      }
-    }
-
-    kvStore = mock[KVStore](Answers.RETURNS_DEEP_STUBS)
-    when(kvStore.executionContext).thenReturn(ExecutionContext.global)
-
-    val fetchContext = FetchContext(kvStore, flagStore = flagStore)
-
-    val fetcherBaseWithFlagStore =
-      spy[fetcher.JoinPartFetcher](new fetcher.JoinPartFetcher(fetchContext, new MetadataStore(fetchContext)))
-    when(fetcherBaseWithFlagStore.isCacheSizeConfigured).thenReturn(true)
-
-    // no name set
-    assertFalse(fetchContext.isCachingEnabled("test_groupby_2"))
-    assertTrue(fetchContext.isCachingEnabled("test_groupby_3"))
-  }
-
   it should "fetch in the happy case" in {
     val fetchContext = mock[FetchContext]
     val baseFetcher = new fetcher.JoinPartFetcher(fetchContext, mock[MetadataStore])

service/src/main/java/ai/chronon/service/FetcherVerticle.java

@@ -26,16 +26,27 @@ public class FetcherVerticle extends AbstractVerticle {
     @Override
     public void start(Promise<Void> startPromise) throws Exception {
         ConfigStore cfgStore = new ConfigStore(vertx);
-        startHttpServer(cfgStore.getServerPort(), cfgStore.encodeConfig(), ApiProvider.buildApi(cfgStore), startPromise);
+
+        Api api = ApiProvider.buildApi(cfgStore);
+
+        // Execute the blocking Bigtable initialization in a separate worker thread
+        vertx.executeBlocking(() -> api.buildJavaFetcher("feature-service", false))
+        .onSuccess(fetcher -> {
+            try {
+                // This code runs back on the event loop when the blocking operation completes
+                startHttpServer(cfgStore.getServerPort(), cfgStore.encodeConfig(), fetcher, startPromise);
+            } catch (Exception e) {
+                startPromise.fail(e);
+            }
+        })
+        .onFailure(startPromise::fail);
     }
 
-    protected void startHttpServer(int port, String configJsonString, Api api, Promise<Void> startPromise) throws Exception {
+    protected void startHttpServer(int port, String configJsonString, JavaFetcher fetcher, Promise<Void> startPromise) throws Exception {
         Router router = Router.router(vertx);
 
         // Define routes
 
-        JavaFetcher fetcher = api.buildJavaFetcher("feature-service", false);
-
         // Set up sub-routes for the various feature retrieval apis
         router.route("/v1/fetch/*").subRouter(FetchRouter.createFetchRoutes(vertx, fetcher));