zixi0825 commented on a change in pull request #3659:
URL:
https://github.com/apache/incubator-dolphinscheduler/pull/3659#discussion_r489395609
##########
File path:
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/python/PythonTask.java
##########
@@ -38,103 +38,110 @@
*/
public class PythonTask extends AbstractTask {
- /**
- * python parameters
- */
- private PythonParameters pythonParameters;
-
- /**
- * task dir
- */
- private String taskDir;
-
- /**
- * python command executor
- */
- private PythonCommandExecutor pythonCommandExecutor;
-
- /**
- * taskExecutionContext
- */
- private TaskExecutionContext taskExecutionContext;
-
- /**
- * constructor
- * @param taskExecutionContext taskExecutionContext
- * @param logger logger
- */
- public PythonTask(TaskExecutionContext taskExecutionContext, Logger logger) {
- super(taskExecutionContext, logger);
- this.taskExecutionContext = taskExecutionContext;
-
- this.pythonCommandExecutor = new PythonCommandExecutor(this::logHandle,
- taskExecutionContext,
- logger);
- }
-
- @Override
- public void init() {
- logger.info("python task params {}", taskExecutionContext.getTaskParams());
-
- pythonParameters =
JSONUtils.parseObject(taskExecutionContext.getTaskParams(),
PythonParameters.class);
-
- if (!pythonParameters.checkParameters()) {
- throw new RuntimeException("python task params is not valid");
+ /**
+ * python parameters
+ */
+ private PythonParameters pythonParameters;
+
+ /**
+ * task dir
+ */
+ private String taskDir;
+
+ /**
+ * python command executor
+ */
+ private PythonCommandExecutor pythonCommandExecutor;
+
+ /**
+ * taskExecutionContext
+ */
+ private TaskExecutionContext taskExecutionContext;
+
+ /**
+ * constructor
+ * @param taskExecutionContext taskExecutionContext
+ * @param logger logger
+ */
+ public PythonTask(TaskExecutionContext taskExecutionContext, Logger
logger) {
+ super(taskExecutionContext, logger);
+ this.taskExecutionContext = taskExecutionContext;
+
+ this.pythonCommandExecutor = new PythonCommandExecutor(this::logHandle,
+ taskExecutionContext,
+ logger);
}
- }
- @Override
- public void handle() throws Exception {
- try {
- // construct process
- CommandExecuteResult commandExecuteResult =
pythonCommandExecutor.run(buildCommand());
+ @Override
+ public void init() {
+ logger.info("python task params {}",
taskExecutionContext.getTaskParams());
- setExitStatusCode(commandExecuteResult.getExitStatusCode());
- setAppIds(commandExecuteResult.getAppIds());
- setProcessId(commandExecuteResult.getProcessId());
- }
- catch (Exception e) {
- logger.error("python task failure", e);
- setExitStatusCode(Constants.EXIT_CODE_FAILURE);
- throw e;
- }
- }
-
- @Override
- public void cancelApplication(boolean cancelApplication) throws Exception {
- // cancel process
- pythonCommandExecutor.cancelApplication();
- }
-
- /**
- * build command
- * @return raw python script
- * @throws Exception exception
- */
- private String buildCommand() throws Exception {
- String rawPythonScript =
pythonParameters.getRawScript().replaceAll("\\r\\n", "\n");
-
- // replace placeholder
- Map<String, Property> paramsMap =
ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
- taskExecutionContext.getDefinedParams(),
- pythonParameters.getLocalParametersMap(),
- CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
- taskExecutionContext.getScheduleTime());
- if (paramsMap != null){
- rawPythonScript =
ParameterUtils.convertParameterPlaceholders(rawPythonScript,
ParamUtils.convert(paramsMap));
- }
+ pythonParameters =
JSONUtils.parseObject(taskExecutionContext.getTaskParams(),
PythonParameters.class);
- logger.info("raw python script : {}", pythonParameters.getRawScript());
- logger.info("task dir : {}", taskDir);
-
- return rawPythonScript;
- }
+ if (!pythonParameters.checkParameters()) {
+ throw new RuntimeException("python task params is not valid");
+ }
+ }
- @Override
- public AbstractParameters getParameters() {
- return pythonParameters;
- }
+ @Override
+ public void handle() throws Exception {
+ try {
+ // construct process
+ CommandExecuteResult commandExecuteResult =
pythonCommandExecutor.run(buildCommand());
+
+ setExitStatusCode(commandExecuteResult.getExitStatusCode());
+ setAppIds(commandExecuteResult.getAppIds());
+ setProcessId(commandExecuteResult.getProcessId());
+ setVarPool(pythonCommandExecutor.getVarPool());
+ }
+ catch (Exception e) {
+ logger.error("python task failure", e);
+ setExitStatusCode(Constants.EXIT_CODE_FAILURE);
+ throw e;
+ }
+ }
+ @Override
+ public void cancelApplication(boolean cancelApplication) throws Exception {
+ // cancel process
+ pythonCommandExecutor.cancelApplication();
+ }
+ /**
+ * build command
+ * @return raw python script
+ * @throws Exception exception
+ */
+ private String buildCommand() throws Exception {
+ String rawPythonScript =
pythonParameters.getRawScript().replaceAll("\\r\\n", "\n");
+
+ // replace placeholder
+ Map<String, Property> paramsMap =
ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
+ taskExecutionContext.getDefinedParams(),
+ pythonParameters.getLocalParametersMap(),
+
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
+ taskExecutionContext.getScheduleTime());
+
+ try {
+ rawPythonScript =
VarPoolUtils.convertPythonScriptPlaceholders(rawPythonScript);
+ }
+ catch (StringIndexOutOfBoundsException e) {
+ // TODO: handle exception
Review comment:
If there is no other way to deal with this error, then you should print
the error message
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
