Fix the bug that the lower DolphinScheduler versions will block the thread until the yarn application is completed.

Author: wushengyeyouya

dss-appconn/appconns/dss-dolphinscheduler-appconn/src/main/java/com/webank/wedatasphere/dss/appconn/dolphinscheduler/conversion/NodeConverter.java

@@ -73,6 +73,7 @@ public DolphinSchedulerTask convertNode(DolphinSchedulerConvertedRel dolphinSche
             addObjectLine.accept("JOB_SOURCE", sourceMap);
             addLine.accept("CONTEXT_ID", workflow.getContextID());
             addLine.accept("LINKIS_GATEWAY_URL", Configuration.getGateWayURL());
+            addLine.accept("DS_VERSION", DolphinSchedulerConf.DS_VERSION.getValue());
             //todo
             addLine.accept("RUN_DATE", "${global_run_date}");
             addObjectLine.accept("JOB_LABELS", new EnvDSSLabel(SchedulerConf.JOB_LABEL.getValue()).getValue());

dss-commons/dss-common/src/main/java/com/webank/wedatasphere/dss/common/conf/DSSCommonConf.java

@@ -24,11 +24,11 @@ public class DSSCommonConf {
 
     public static final CommonVars<String> DSS_EXPORT_URL = CommonVars.apply("wds.dss.server.export.url", "/appcom/tmp/dss");
 
-    public static final CommonVars<String> DSS_ORCHESTRATOR_FRAMEWORK_APP_CONN_NAME = CommonVars.apply("wds.dss.orchestrator.framework.appconn", "orchestrator-framework");
-
     public static final CommonVars<Integer> DSS_DOMAIN_LEVEL = CommonVars.apply("wds.linkis.gateway.domain.level", 3);
 
-//    public static final CommonVars<String> DSS_TOKEN_TICKET_KEY = CommonVars.apply("wds.dss.user.ticket.key", "bdp-user-ticket-id");
+    /**
+     * The old value is bdp-user-ticket-id
+      */
     public static final CommonVars<String> DSS_TOKEN_TICKET_KEY = CommonVars.apply("wds.dss.user.ticket.key", "linkis_user_session_ticket_id_v1");
 
 }

plugins/dolphinscheduler/dss-dolphinscheduler-client/src/main/scala/com/webank/wedatasphere/dss/plugins/dolphinscheduler/linkis/client/DSSDolphinSchedulerClient.scala

@@ -14,21 +14,39 @@ import org.apache.linkis.common.conf.CommonVars
 import org.apache.linkis.common.utils.{Logging, Utils}
 
 object DSSDolphinSchedulerClient extends Logging {
-  val DEFAULT_PROPERTY_FILE_NAME = "linkis.properties"
-  val DEFAULT_CONFIG_DIR = "conf"
-  val CHARSET_NAME = "utf-8"
-
-  val logObj: LinkisJobExecutionLog = new LinkisJobExecutionLog {
-    override def info(message: Object, t: Throwable): Unit =
-      if(message != null) logger.info(message.toString, t) else if(t != null) logger.info(null, t)
-    override def warn(message: Object, t: Throwable): Unit =
-      if(message != null) logger.warn(message.toString, t) else if(t != null) logger.warn(null, t)
-    override def error(message: Object, t: Throwable): Unit =
-      if(message != null) logger.error(message.toString, t) else if(t != null) logger.error(null, t)
-  }
 
   def main(args: Array[String]): Unit = {
 
+    val dsVersion = CommonVars("DS_VERSION", "").getValue
+    info(s"The DolphinScheduler Version => $dsVersion.")
+
+    val logObj: LinkisJobExecutionLog = if(StringUtils.isNotBlank(dsVersion) && dsVersion.startsWith("2.")) {
+      new LinkisJobExecutionLog {
+        override def info(message: Object, t: Throwable): Unit =
+          if(message != null) logger.info(message.toString, t) else if(t != null) logger.info(null, t)
+        override def warn(message: Object, t: Throwable): Unit =
+          if(message != null) logger.warn(message.toString, t) else if(t != null) logger.warn(null, t)
+        override def error(message: Object, t: Throwable): Unit =
+          if(message != null) logger.error(message.toString, t) else if(t != null) logger.error(null, t)
+      }
+    } else {
+      warn("For the lower DolphinScheduler versions(< 2.0), we will parse the yarn applicationId to application__\\d+_\\d+.")
+      warn("Why? Because the lower DolphinScheduler versions will block the thread until the yarn application is completed.")
+      // For more information, please visit the class AbstractCommandExecutor in DolphinScheduler (line 222 ~ 231).
+      new LinkisJobExecutionLog {
+        private implicit def parseYarnApplication(message: Object): String = message match {
+          case str: String if str.contains(" application_") =>
+            str.replaceAll(" application_", " application__")
+          case str: String => str
+          case obj if obj != null => obj.toString
+          case _ => ""
+        }
+        override def info(message: Object, t: Throwable): Unit = logger.info(message, t)
+        override def warn(message: Object, t: Throwable): Unit = logger.warn(message, t)
+        override def error(message: Object, t: Throwable): Unit = logger.error(message, t)
+      }
+    }
+
     val jobProps = new util.HashMap[String, String]
     val getAndSet = (fromKey: String, toKey: String) => {
       val value = CommonVars(fromKey, "").getValue