add untracked

ytsaurus · Jul 3, 2024 · 4a3f98b · 4a3f98b
1 parent 6351e54
commit 4a3f98b
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 13 deletions.
diff --git a/resource-manager/src/main/scala/org/apache/spark/deploy/ytsaurus/Config.scala b/resource-manager/src/main/scala/org/apache/spark/deploy/ytsaurus/Config.scala
@@ -112,4 +112,9 @@ object Config {
     .version("3.2.2")
     .stringConf
     .createOptional
+
+  val YTSAURUS_UNTRACKED_DRIVER_OPERATION = ConfigBuilder("spark.ytsaurus.untracked.driver.operation")
+    .version("3.2.2")
+    .booleanConf
+    .createWithDefault(false)
 }
diff --git a/...-manager/src/main/scala/org/apache/spark/deploy/ytsaurus/YTsaurusClusterApplication.scala b/...-manager/src/main/scala/org/apache/spark/deploy/ytsaurus/YTsaurusClusterApplication.scala
@@ -4,9 +4,10 @@ package org.apache.spark.deploy.ytsaurus
 import scala.collection.mutable
 import org.apache.spark.SparkConf
 import org.apache.spark.deploy.SparkApplication
+import org.apache.spark.deploy.ytsaurus.Config.YTSAURUS_UNTRACKED_DRIVER_OPERATION
 import org.apache.spark.internal.Logging
 import org.apache.spark.scheduler.cluster.ytsaurus.YTsaurusOperationManager
-import org.apache.spark.scheduler.cluster.ytsaurus.YTsaurusOperationManager.{getWebUIAddress, getOperationState}
+import org.apache.spark.scheduler.cluster.ytsaurus.YTsaurusOperationManager.{getOperationState, getWebUIAddress}
 
 
 private[spark] class YTsaurusClusterApplication extends SparkApplication with Logging {
@@ -30,19 +31,23 @@ private[spark] class YTsaurusClusterApplication extends SparkApplication with Lo
     try {
       val driverOperation = operationManager.startDriver(conf, appArgs)
 
-      var currentState = "undefined"
-      var webUIAddress: Option[String] = None
-      while (!YTsaurusOperationManager.isFinalState(currentState)) {
-        Thread.sleep(pingInterval)
-        val opSpec = operationManager.getOperation(driverOperation)
-        currentState = getOperationState(opSpec)
-        logInfo(s"Operation: ${driverOperation.id}, State: $currentState")
-
-        val currentWebUiAddress = getWebUIAddress(opSpec)
-        if (currentWebUiAddress.isDefined && currentWebUiAddress != webUIAddress) {
-          webUIAddress = currentWebUiAddress
-          webUIAddress.foreach(addr => logInfo(s"Web UI: $addr"))
+      if (!conf.get(YTSAURUS_UNTRACKED_DRIVER_OPERATION)) {
+        var currentState = "undefined"
+        var webUIAddress: Option[String] = None
+        while (!YTsaurusOperationManager.isFinalState(currentState)) {
+          Thread.sleep(pingInterval)
+          val opSpec = operationManager.getOperation(driverOperation)
+          currentState = getOperationState(opSpec)
+          logInfo(s"Operation: ${driverOperation.id}, State: $currentState")
+
+          val currentWebUiAddress = getWebUIAddress(opSpec)
+          if (currentWebUiAddress.isDefined && currentWebUiAddress != webUIAddress) {
+            webUIAddress = currentWebUiAddress
+            webUIAddress.foreach(addr => logInfo(s"Web UI: $addr"))
+          }
         }
+      } else {
+        logInfo(s"Driver operation: ${driverOperation.id}")
       }
     } finally {
       operationManager.close()