add untracked

ytsaurus · Sep 11, 2024 · e9f7150 · e9f7150
1 parent 67b38d8
commit e9f7150
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 16 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
@@ -121,4 +121,9 @@ object Config {
   val SPYT_ANNOTATIONS = "spark.ytsaurus.annotations"
   val SPYT_DRIVER_ANNOTATIONS = "spark.ytsaurus.driver.annotations"
   val SPYT_EXECUTORS_ANNOTATIONS = "spark.ytsaurus.executors.annotations"
+
+  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
@@ -5,6 +5,7 @@ import scala.collection.mutable
 import org.apache.spark.SparkConf
 import org.apache.spark.deploy.SparkApplication
 import org.apache.spark.deploy.ytsaurus.Config.YTSAURUS_DRIVER_OPERATION_DUMP_PATH
+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.{getOperationState, getWebUIAddress, isCompletedState}
@@ -32,24 +33,27 @@ private[spark] class YTsaurusClusterApplication extends SparkApplication with Lo
 
     try {
       val driverOperation = operationManager.startDriver(conf, appArgs)
-
-      var currentState = "undefined"
-      var webUIAddress: Option[String] = None
       conf.get(YTSAURUS_DRIVER_OPERATION_DUMP_PATH).foreach { File(_).writeAll(driverOperation.id.toString) }
-      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"))
+          }
         }
-      }
-      if (!isCompletedState(currentState)) {
-        throw new IllegalStateException(s"Unsuccessful operation state: $currentState")  // For non-zero exit code
+        if (!isCompletedState(currentState)) {
+          throw new IllegalStateException(s"Unsuccessful operation state: $currentState")  // For non-zero exit code
+        }
+      } else {
+        logInfo(s"Driver operation: ${driverOperation.id}")
       }
     } finally {
       operationManager.close()