style: Apply scalafix and scalafmt changes

Author: piyush-zlai

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

@@ -72,13 +72,12 @@ object Fetcher {
     context.distribution(Metrics.Name.FetchCount, responseMap.size)
   }
 
-  /**
-   * Response for a join schema request
-   * @param joinName - Name of the join
-   * @param keySchema - Avro schema string for the key
-   * @param valueSchema - Avro schema string for the value
-   * @param schemaHash - Hash of the join schema payload (used to track updates to key / value schema fields or types)
-   */
+  /** Response for a join schema request
+    * @param joinName - Name of the join
+    * @param keySchema - Avro schema string for the key
+    * @param valueSchema - Avro schema string for the value
+    * @param schemaHash - Hash of the join schema payload (used to track updates to key / value schema fields or types)
+    */
   case class JoinSchemaResponse(joinName: String, keySchema: String, valueSchema: String, schemaHash: String)
 }
 
@@ -431,14 +430,18 @@ class Fetcher(val kvStore: KVStore,
 
     val joinCodecTry = joinCodecCache(joinName)
 
-    val joinSchemaResponse = joinCodecTry.map { joinCodec =>
-      JoinSchemaResponse(joinName, joinCodec.keyCodec.schemaStr, joinCodec.valueCodec.schemaStr, joinCodec.loggingSchemaHash)
-    }.recover {
-      case exception: Throwable =>
+    val joinSchemaResponse = joinCodecTry
+      .map { joinCodec =>
+        JoinSchemaResponse(joinName,
+                           joinCodec.keyCodec.schemaStr,
+                           joinCodec.valueCodec.schemaStr,
+                           joinCodec.loggingSchemaHash)
+      }
+      .recover { case exception: Throwable =>
         logger.error(s"Failed to fetch join schema for $joinName", exception)
         ctx.incrementException(exception)
         throw exception
-    }
+      }
 
     joinSchemaResponse.foreach(_ => ctx.distribution("response.latency.millis", System.currentTimeMillis() - startTime))
     joinSchemaResponse

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

@@ -181,47 +181,44 @@ class MetadataStore(fetchContext: FetchContext) {
 
     def parseJoins(response: ListResponse): Seq[String] = {
       val result = response.values
-        .map {
-          seqListValues =>
-
-            seqListValues
-              .map(kv => new String(kv.valueBytes, StandardCharsets.UTF_8))
-              .map(v => ThriftJsonCodec.fromJsonStr[Join](v, check = false, classOf[Join]))
-              .filter(_.join.metaData.online == isOnline)
-              .map(_.metaData.name)
-
-        }.recover {
-            case e: Exception =>
-              logger.error("Failed to list & parse joins from list response", e)
-              context.withSuffix("join_list").increment(Metrics.Name.Exception)
-              throw e
+        .map { seqListValues =>
+          seqListValues
+            .map(kv => new String(kv.valueBytes, StandardCharsets.UTF_8))
+            .map(v => ThriftJsonCodec.fromJsonStr[Join](v, check = false, classOf[Join]))
+            .filter(_.join.metaData.online == isOnline)
+            .map(_.metaData.name)
+
+        }
+        .recover { case e: Exception =>
+          logger.error("Failed to list & parse joins from list response", e)
+          context.withSuffix("join_list").increment(Metrics.Name.Exception)
+          throw e
         }
 
       result.get
     }
 
-    def doRetrieveAllListConfs(acc: mutable.ArrayBuffer[String], paginationKey: Option[Any] = None): Future[Seq[String]] = {
+    def doRetrieveAllListConfs(acc: mutable.ArrayBuffer[String],
+                               paginationKey: Option[Any] = None): Future[Seq[String]] = {
       val propsMap = {
         paginationKey match {
           case Some(key) => Map(ListEntityType -> JoinKeyword, ContinuationKey -> key)
-          case None => Map(ListEntityType -> JoinKeyword)
+          case None      => Map(ListEntityType -> JoinKeyword)
         }
       }
 
       val listRequest = ListRequest(fetchContext.metadataDataset, propsMap)
-      fetchContext.kvStore.list(listRequest).flatMap {
-        response =>
-
-          val joinSeq: Seq[String] = parseJoins(response)
-          val newAcc = acc ++ joinSeq
-          if (response.resultProps.contains(ContinuationKey)) {
-            doRetrieveAllListConfs(newAcc, response.resultProps.get(ContinuationKey))
-          } else {
-            context
-              .withSuffix("join_list")
-              .distribution(Metrics.Name.LatencyMillis, System.currentTimeMillis() - startTimeMs)
-            Future.successful(newAcc)
-          }
+      fetchContext.kvStore.list(listRequest).flatMap { response =>
+        val joinSeq: Seq[String] = parseJoins(response)
+        val newAcc = acc ++ joinSeq
+        if (response.resultProps.contains(ContinuationKey)) {
+          doRetrieveAllListConfs(newAcc, response.resultProps.get(ContinuationKey))
+        } else {
+          context
+            .withSuffix("join_list")
+            .distribution(Metrics.Name.LatencyMillis, System.currentTimeMillis() - startTimeMs)
+          Future.successful(newAcc)
+        }
       }
     }
 

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

@@ -49,7 +49,7 @@ class ListJoinsTest extends AnyFlatSpec with MockitoSugar with BeforeAndAfter wi
   it should "fail the call on internal issues" in {
     val metadataStore = new MetadataStore(FetchContext(kvStore))
     when(kvStore.list(any())).thenReturn(generateBrokenListResponse())
-    an [Exception] should be thrownBy Await.result(metadataStore.listJoins(), 10.seconds)
+    an[Exception] should be thrownBy Await.result(metadataStore.listJoins(), 10.seconds)
   }
 
   it should "paginate list calls" in {
@@ -61,7 +61,8 @@ class ListJoinsTest extends AnyFlatSpec with MockitoSugar with BeforeAndAfter wi
     val listResponseValues: Seq[Try[Seq[ListValue]]] = responses.map(v => Success(Seq(v)))
 
     // first response will have a continuation key
-    val first = Future(ListResponse(ListRequest(MetadataDataset, Map.empty), listResponseValues.head, Map(ContinuationKey -> "1")))
+    val first = Future(
+      ListResponse(ListRequest(MetadataDataset, Map.empty), listResponseValues.head, Map(ContinuationKey -> "1")))
     // second response will not have a continuation key
     val second = Future(ListResponse(ListRequest(MetadataDataset, Map.empty), listResponseValues.last, Map.empty))
 
@@ -77,7 +78,7 @@ class ListJoinsTest extends AnyFlatSpec with MockitoSugar with BeforeAndAfter wi
       // first is online = false
       "joins/user_transactions.txn_join_a",
       // this one is online = true
-      "joins/user_transactions.txn_join_d",
+      "joins/user_transactions.txn_join_d"
     )
 
     paths.map { path =>