Update AbstractCommandExecutor.java

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutor.java

@@ -339,50 +339,40 @@ public abstract class AbstractCommandExecutor {
     private void parseProcessOutput(Process process) {
         String threadLoggerInfoName = String.format(LoggerUtils.TASK_LOGGER_THREAD_NAME + "-%s", taskExecutionContext.getTaskAppId());
         ExecutorService getOutputLogService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName + "-" + "getOutputLogService");
-        getOutputLogService.submit(new Runnable() {
+        getOutputLogService.submit(() -> {
-            @Override
+            BufferedReader inReader = null;
-            public void run() {
-                BufferedReader inReader = null;
 
-                try {
+            try {
-                    inReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
+                inReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
-                    String line;
+                String line;
-                    while ((line = inReader.readLine()) != null) {
+                logBuffer.add("welcome to use bigdata scheduling system...");
-                        if (line.startsWith("${setValue(")) {
+                while ((line = inReader.readLine()) != null) {
-                            varPool.append(line.substring("${setValue(".length(), line.length() - 2));
+                    logBuffer.add(line);
-                            varPool.append("$VarPool$");
-                        } else {
-                            logBuffer.add(line);
-                        }
-                    }
-                } catch (Exception e) {
-                    logger.error(e.getMessage(), e);
-                } finally {
-                    logOutputIsScuccess = true;
-                    close(inReader);
                 }
+            } catch (Exception e) {
+                logger.error(e.getMessage(), e);
+            } finally {
+                logOutputIsScuccess = true;
+                close(inReader);
             }
         });
         getOutputLogService.shutdown();
 
         ExecutorService parseProcessOutputExecutorService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
-        parseProcessOutputExecutorService.submit(new Runnable() {
+        parseProcessOutputExecutorService.submit(() -> {
-            @Override
+            try {
-            public void run() {
+                long lastFlushTime = System.currentTimeMillis();
-                try {
+                while (logBuffer.size() > 0 || !logOutputIsScuccess) {
-                    long lastFlushTime = System.currentTimeMillis();
+                    if (logBuffer.size() > 0) {
-                    while (logBuffer.size() > 0 || !logOutputIsScuccess) {
+                        lastFlushTime = flush(lastFlushTime);
-                        if (logBuffer.size() > 0) {
+                    } else {
-                            lastFlushTime = flush(lastFlushTime);
+                        Thread.sleep(Constants.DEFAULT_LOG_FLUSH_INTERVAL);
-                        } else {
-                            Thread.sleep(Constants.DEFAULT_LOG_FLUSH_INTERVAL);
-                        }
                     }
-                } catch (Exception e) {
-                    logger.error(e.getMessage(), e);
-                } finally {
-                    clear();
                 }
+            } catch (Exception e) {
+                logger.error(e.getMessage(), e);
+            } finally {
+                clear();
             }
         });
         parseProcessOutputExecutorService.shutdown();