resolve comments

Signed-off-by: Louis Chu <[email protected]>
opensearch-project · Oct 24, 2024 · 4dcc1ea · 4dcc1ea
1 parent 6cf9a45
commit 4dcc1ea
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 3 deletions.
diff --git a/flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSpark.scala b/flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSpark.scala
@@ -566,7 +566,7 @@ class FlintSpark(val spark: SparkSession) extends FlintSparkTransactionSupport w
     val indexName = index.name
     val indexLogEntry = index.latestLogEntry.get
     val internalSchedulingService =
-      new FlintSparkJobInternalSchedulingService(spark, flintIndexMonitor)
+      new FlintSparkJobInternalSchedulingService(spark, flintSparkConf, flintIndexMonitor)
     val externalSchedulingService =
       new FlintSparkJobExternalSchedulingService(flintAsyncQueryScheduler, flintSparkConf)
 

diff --git a/...n/scala/org/opensearch/flint/spark/scheduler/FlintSparkJobInternalSchedulingService.scala b/...n/scala/org/opensearch/flint/spark/scheduler/FlintSparkJobInternalSchedulingService.scala
@@ -30,6 +30,7 @@ import org.apache.spark.sql.flint.config.FlintSparkConf
  */
 class FlintSparkJobInternalSchedulingService(
     spark: SparkSession,
+    flintSparkConf: FlintSparkConf,
     flintIndexMonitor: FlintSparkIndexMonitor)
     extends FlintSparkJobSchedulingService
     with Logging {
@@ -78,7 +79,7 @@ class FlintSparkJobInternalSchedulingService(
   private def startRefreshingJob(index: FlintSparkIndex): Option[String] = {
     logInfo(s"Starting refreshing job for index ${index.name()}")
     val indexRefresh = FlintSparkIndexRefresh.create(index.name(), index)
-    val jobId = indexRefresh.start(spark, new FlintSparkConf(spark.conf.getAll.toMap.asJava))
+    val jobId = indexRefresh.start(spark, flintSparkConf)
 
     // NOTE: Resolution for previous concurrency issue
     // This code addresses a previously identified concurrency issue with recoverIndex

diff --git a/.../src/main/scala/org/opensearch/flint/spark/scheduler/FlintSparkJobSchedulingService.scala b/.../src/main/scala/org/opensearch/flint/spark/scheduler/FlintSparkJobSchedulingService.scala
@@ -70,7 +70,7 @@ object FlintSparkJobSchedulingService {
     if (isExternalSchedulerEnabled(index)) {
       new FlintSparkJobExternalSchedulingService(flintAsyncQueryScheduler, flintSparkConf)
     } else {
-      new FlintSparkJobInternalSchedulingService(spark, flintIndexMonitor)
+      new FlintSparkJobInternalSchedulingService(spark, flintSparkConf, flintIndexMonitor)
     }
   }