[Fix-18389][DataX] Read job definition from attached resource file when custom json is empty

dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParameters.java

@@ -25,13 +25,17 @@
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
 import org.apache.dolphinscheduler.spi.enums.Flag;
 
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.List;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 import lombok.Data;
 
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
 /**
  * DataX parameter
  */
@@ -116,7 +120,50 @@ public boolean checkParameters() {
                     && StringUtils.isNotEmpty(sql)
                     && StringUtils.isNotEmpty(targetTable);
         } else {
-            return StringUtils.isNotEmpty(json);
+            // Custom config is valid with either inline json or an attached resource file that
+            // unambiguously carries the job definition, identified as the single .json resource
+            // (issue #18389). resourceList is multi-select and also holds auxiliary files, so a
+            // non-empty list on its own is not enough.
+            return !isInlineJsonAbsent() || getJobDefinitionResource() != null;
+        }
+    }
+
+    /**
+     * When the inline json is absent the job definition must come from an attached resource file.
+     * resourceList is multi-select and also carries auxiliary files such as Kerberos keytabs and
+     * xml configs, so the job definition is identified as the single resource whose name ends with
+     * {@code .json} rather than the first entry in the list (issue #18389). Returns that resource,
+     * or {@code null} when there is not exactly one json resource, which the caller treats as a
+     * missing or ambiguous job definition.
+     */
+    public ResourceInfo getJobDefinitionResource() {
+        if (CollectionUtils.isEmpty(resourceList)) {
+            return null;
+        }
+        List<ResourceInfo> jsonResources = resourceList.stream()
+                .filter(Objects::nonNull)
+                .filter(resource -> StringUtils.endsWithIgnoreCase(resource.getResourceName(), ".json"))
+                .collect(Collectors.toList());
+        return jsonResources.size() == 1 ? jsonResources.get(0) : null;
+    }
+
+    /**
+     * Returns true when the json field carries no usable inline job definition. The UI
+     * historically stored an empty object placeholder in the json field, so a blank value
+     * and any semantically empty JSON object (for example {@code {}}, {@code { }} or a
+     * formatted multi-line empty object) are all treated as absent (issue #18389).
+     */
+    public boolean isInlineJsonAbsent() {
+        if (StringUtils.isBlank(json)) {
+            return true;
+        }
+        try {
+            ObjectNode node = JSONUtils.parseObject(json);
+            return node == null || node.isEmpty();
+        } catch (Exception e) {
+            // not parseable as a JSON object, so there is inline content: downstream
+            // validation reports the malformed definition
+            return false;
         }
     }
 

dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/main/java/org/apache/dolphinscheduler/plugin/task/datax/DataxTask.java

@@ -30,8 +30,10 @@
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.log.SensitiveDataConverter;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
 import org.apache.dolphinscheduler.plugin.task.api.model.TaskResponse;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
+import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext;
 import org.apache.dolphinscheduler.plugin.task.api.shell.IShellInterceptorBuilder;
 import org.apache.dolphinscheduler.plugin.task.api.shell.ShellInterceptorBuilderFactory;
 import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
@@ -168,6 +170,18 @@ public void cancel() throws TaskException {
         }
     }
 
+    /**
+     * Reads the DataX job definition from the designated json resource file. The worker has
+     * already downloaded resources into the execution directory by the time the task runs.
+     */
+    private String readJsonFromResourceFile(ResourceInfo jobResource) throws Exception {
+        String resourceFileName = jobResource.getResourceName();
+        ResourceContext resourceContext = taskRequest.getResourceContext();
+        return FileUtils.readFileToString(
+                new File(resourceContext.getResourceItem(resourceFileName).getResourceAbsolutePathInLocal()),
+                StandardCharsets.UTF_8);
+    }
+
     /**
      * build datax configuration file
      *
@@ -185,7 +199,23 @@ private String buildDataxJsonFile(Map<String, Property> paramsMap) throws Except
         }
 
         if (dataXParameters.getCustomConfig() == Flag.YES.ordinal()) {
-            json = dataXParameters.getJson().replaceAll("\\r\\n", System.lineSeparator());
+            // An attached resource file is a valid way to supply the job definition. Without
+            // this branch the worker downloads the resource but the plugin runs with the empty
+            // inline json and the job fails (issue #18389). Existing tasks created through the
+            // UI carry an empty object placeholder, treat it the same as no inline json.
+            if (dataXParameters.isInlineJsonAbsent()) {
+                // the job definition is the single attached .json resource, never the first
+                // entry in resourceList, which may be an auxiliary keytab or xml (issue #18389)
+                ResourceInfo jobResource = dataXParameters.getJobDefinitionResource();
+                if (jobResource == null) {
+                    throw new TaskException(
+                            "DataX job definition is missing, provide inline json or attach exactly one .json resource file");
+                }
+                json = readJsonFromResourceFile(jobResource);
+            } else {
+                json = dataXParameters.getJson();
+            }
+            json = json.replaceAll("\\r\\n", System.lineSeparator());
         } else {
             ObjectNode job = JSONUtils.createObjectNode();
             job.putArray("content").addAll(buildDataxJobContentJson());

dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/test/java/org/apache/dolphinscheduler/plugin/task/datax/DataxParametersTest.java

@@ -32,6 +32,108 @@ public class DataxParametersTest {
      */
     public static final String JVM_PARAM = " --jvm=\"-Xms%sG -Xmx%sG\" ";
 
+    @Test
+    public void testCheckParametersWithCustomConfig() {
+        DataxParameters withInlineJson = new DataxParameters();
+        withInlineJson.setCustomConfig(1);
+        withInlineJson.setJson("{\"job\":{}}");
+        Assertions.assertTrue(withInlineJson.checkParameters());
+
+        // a blank json field or any semantically empty JSON object is no inline
+        // definition: invalid without a resource file, valid with one because the
+        // resource then carries the job definition (issue #18389)
+        String[] absentJsonVariants = {null, "", "   ", "{}", "{ }", "{\n\n}", " { } "};
+        for (String variant : absentJsonVariants) {
+            DataxParameters withoutResource = new DataxParameters();
+            withoutResource.setCustomConfig(1);
+            withoutResource.setJson(variant);
+            Assertions.assertTrue(withoutResource.isInlineJsonAbsent(),
+                    "expected inline json to be absent for: [" + variant + "]");
+            Assertions.assertFalse(withoutResource.checkParameters(),
+                    "expected invalid without resource for json: [" + variant + "]");
+
+            DataxParameters withResource = new DataxParameters();
+            withResource.setCustomConfig(1);
+            withResource.setJson(variant);
+            withResource.setResourceList(buildResourceList());
+            Assertions.assertTrue(withResource.checkParameters(),
+                    "expected valid with resource for json: [" + variant + "]");
+        }
+
+        // a non-empty inline definition stays inline even when a resource is attached
+        DataxParameters inlineWithResource = new DataxParameters();
+        inlineWithResource.setCustomConfig(1);
+        inlineWithResource.setJson("{\"job\":{}}");
+        inlineWithResource.setResourceList(buildResourceList());
+        Assertions.assertFalse(inlineWithResource.isInlineJsonAbsent());
+        Assertions.assertTrue(inlineWithResource.checkParameters());
+
+        // malformed json is not treated as absent, downstream validation reports it
+        DataxParameters malformed = new DataxParameters();
+        malformed.setCustomConfig(1);
+        malformed.setJson("{invalid");
+        Assertions.assertFalse(malformed.isInlineJsonAbsent());
+
+        DataxParameters withNeither = new DataxParameters();
+        withNeither.setCustomConfig(1);
+        Assertions.assertFalse(withNeither.checkParameters());
+    }
+
+    private List<ResourceInfo> buildResourceList() {
+        ResourceInfo resource = new ResourceInfo();
+        resource.setResourceName("/datax/job.json");
+        List<ResourceInfo> resources = new ArrayList<>();
+        resources.add(resource);
+        return resources;
+    }
+
+    @Test
+    public void testJobDefinitionResourceIsTheSingleJsonResource() {
+        // resourceList is multi-select and also carries auxiliary files, so the job definition
+        // is identified as the single .json resource, not resourceList.get(0) (issue #18389)
+
+        // only an auxiliary keytab and no json: no job definition, invalid
+        DataxParameters onlyAuxiliary = new DataxParameters();
+        onlyAuxiliary.setCustomConfig(1);
+        onlyAuxiliary.setResourceList(resources("/datax/hdfs.keytab"));
+        Assertions.assertNull(onlyAuxiliary.getJobDefinitionResource());
+        Assertions.assertFalse(onlyAuxiliary.checkParameters(),
+                "a task carrying only auxiliary resources has no job definition");
+
+        // a keytab listed before the job file: the .json is chosen, not the first entry
+        DataxParameters auxiliaryBeforeJob = new DataxParameters();
+        auxiliaryBeforeJob.setCustomConfig(1);
+        auxiliaryBeforeJob.setResourceList(resources("/datax/hdfs.keytab", "/datax/job.json"));
+        Assertions.assertEquals("/datax/job.json",
+                auxiliaryBeforeJob.getJobDefinitionResource().getResourceName());
+        Assertions.assertTrue(auxiliaryBeforeJob.checkParameters());
+
+        // two json resources are ambiguous: no single job definition, invalid
+        DataxParameters twoJson = new DataxParameters();
+        twoJson.setCustomConfig(1);
+        twoJson.setResourceList(resources("/datax/a.json", "/datax/b.json"));
+        Assertions.assertNull(twoJson.getJobDefinitionResource());
+        Assertions.assertFalse(twoJson.checkParameters());
+
+        // exactly one json resource is the job definition
+        DataxParameters singleJson = new DataxParameters();
+        singleJson.setCustomConfig(1);
+        singleJson.setResourceList(resources("/datax/job.json"));
+        Assertions.assertEquals("/datax/job.json",
+                singleJson.getJobDefinitionResource().getResourceName());
+        Assertions.assertTrue(singleJson.checkParameters());
+    }
+
+    private List<ResourceInfo> resources(String... names) {
+        List<ResourceInfo> list = new ArrayList<>();
+        for (String name : names) {
+            ResourceInfo resource = new ResourceInfo();
+            resource.setResourceName(name);
+            list.add(resource);
+        }
+        return list;
+    }
+
     @Test
     public void testLoadJvmEnv() {
 

dolphinscheduler-task-plugin/dolphinscheduler-task-datax/src/test/java/org/apache/dolphinscheduler/plugin/task/datax/DataxTaskTest.java

@@ -37,8 +37,10 @@
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskRunStatus;
 import org.apache.dolphinscheduler.plugin.task.api.model.ApplicationInfo;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
 import org.apache.dolphinscheduler.plugin.task.api.model.TaskResponse;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
+import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext;
 import org.apache.dolphinscheduler.spi.datasource.BaseConnectionParam;
 import org.apache.dolphinscheduler.spi.enums.DbType;
 
@@ -271,6 +273,128 @@ private Map<String, Property> createPrepareParamsMap() {
         return paramsMap;
     }
 
+    @Test
+    public void testCustomConfigReadsJobDefinitionFromResourceFile() throws Exception {
+        // a real resource file carrying the job definition, with a formatted empty object
+        // placeholder inline (the semantic-absence rule, not a literal "{}" compare)
+        String resourceJson = "{\"job\":{\"content\":[{\"reader\":{\"name\":\"mysqlreader\"}}]}}";
+        File resourceFile = File.createTempFile("datax-job", ".json");
+        resourceFile.deleteOnExit();
+        java.nio.file.Files.write(resourceFile.toPath(),
+                resourceJson.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+
+        DataxParameters parameters = new DataxParameters();
+        parameters.setCustomConfig(1);
+        parameters.setJson("{\n  \n}");
+        parameters.setXms(1);
+        parameters.setXmx(1);
+        ResourceInfo resourceInfo = new ResourceInfo();
+        resourceInfo.setResourceName("/datax/job.json");
+        parameters.setResourceList(java.util.Collections.singletonList(resourceInfo));
+
+        TaskExecutionContext taskExecutionContext = buildTestTaskExecutionContext();
+        // own app id so the generated file cannot collide with other tests' job files
+        taskExecutionContext.setTaskAppId("app-id-resource");
+        taskExecutionContext.setPrepareParamsMap(null);
+        taskExecutionContext.setTaskParams(JSONUtils.toJsonString(parameters));
+        ResourceContext resourceContext = new ResourceContext();
+        resourceContext.addResourceItem(ResourceContext.ResourceItem.builder()
+                .resourceAbsolutePathInStorage("/datax/job.json")
+                .resourceAbsolutePathInLocal(resourceFile.getAbsolutePath())
+                .build());
+        taskExecutionContext.setResourceContext(resourceContext);
+
+        DataxTask dataxTask = new DataxTask(taskExecutionContext);
+        dataxTask.init();
+
+        ShellCommandExecutor shellCommandExecutor = mock(ShellCommandExecutor.class);
+        Field shellCommandExecutorFiled = DataxTask.class.getDeclaredField("shellCommandExecutor");
+        shellCommandExecutorFiled.setAccessible(true);
+        shellCommandExecutorFiled.set(dataxTask, shellCommandExecutor);
+
+        TaskResponse taskResponse = new TaskResponse();
+        taskResponse.setStatus(TaskRunStatus.SUCCESS);
+        taskResponse.setExitStatusCode(0);
+        taskResponse.setProcessId(1);
+        when(shellCommandExecutor.run(any(), eq(taskCallBack))).thenReturn(taskResponse);
+
+        dataxTask.handle(taskCallBack);
+        Assertions.assertEquals(0, dataxTask.getExitStatusCode());
+
+        // the generated job file must carry the resource content, not the "{}" placeholder
+        File jsonFile = new File("/tmp/execution/app-id-resource_job.json");
+        String generated = FileUtils.readFile2Str(Files.newInputStream(jsonFile.toPath()));
+        Assertions.assertTrue(generated.contains("mysqlreader"),
+                "generated job file should contain the resource file definition, was: " + generated);
+        Assertions.assertTrue(jsonFile.delete());
+    }
+
+    @Test
+    public void testCustomConfigReadsJobFromJsonResourceNotFirstAuxiliaryResource() throws Exception {
+        // resourceList carries a keytab BEFORE the job file. The worker must read the .json job
+        // definition, not resourceList.get(0) which is the keytab (issue #18389, review by SbloodyS)
+        String jobJson = "{\"job\":{\"content\":[{\"reader\":{\"name\":\"mysqlreader\"}}]}}";
+        File jobFile = File.createTempFile("datax-job", ".json");
+        jobFile.deleteOnExit();
+        java.nio.file.Files.write(jobFile.toPath(),
+                jobJson.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+        File keytabFile = File.createTempFile("hdfs", ".keytab");
+        keytabFile.deleteOnExit();
+        java.nio.file.Files.write(keytabFile.toPath(),
+                "keytab-binary-not-json".getBytes(java.nio.charset.StandardCharsets.UTF_8));
+
+        DataxParameters parameters = new DataxParameters();
+        parameters.setCustomConfig(1);
+        parameters.setJson("{}");
+        parameters.setXms(1);
+        parameters.setXmx(1);
+        ResourceInfo keytab = new ResourceInfo();
+        keytab.setResourceName("/datax/hdfs.keytab");
+        ResourceInfo job = new ResourceInfo();
+        job.setResourceName("/datax/job.json");
+        parameters.setResourceList(java.util.Arrays.asList(keytab, job));
+
+        TaskExecutionContext taskExecutionContext = buildTestTaskExecutionContext();
+        taskExecutionContext.setTaskAppId("app-id-multi-resource");
+        taskExecutionContext.setPrepareParamsMap(null);
+        taskExecutionContext.setTaskParams(JSONUtils.toJsonString(parameters));
+        ResourceContext resourceContext = new ResourceContext();
+        resourceContext.addResourceItem(ResourceContext.ResourceItem.builder()
+                .resourceAbsolutePathInStorage("/datax/hdfs.keytab")
+                .resourceAbsolutePathInLocal(keytabFile.getAbsolutePath())
+                .build());
+        resourceContext.addResourceItem(ResourceContext.ResourceItem.builder()
+                .resourceAbsolutePathInStorage("/datax/job.json")
+                .resourceAbsolutePathInLocal(jobFile.getAbsolutePath())
+                .build());
+        taskExecutionContext.setResourceContext(resourceContext);
+
+        DataxTask dataxTask = new DataxTask(taskExecutionContext);
+        dataxTask.init();
+
+        ShellCommandExecutor shellCommandExecutor = mock(ShellCommandExecutor.class);
+        Field shellCommandExecutorFiled = DataxTask.class.getDeclaredField("shellCommandExecutor");
+        shellCommandExecutorFiled.setAccessible(true);
+        shellCommandExecutorFiled.set(dataxTask, shellCommandExecutor);
+
+        TaskResponse taskResponse = new TaskResponse();
+        taskResponse.setStatus(TaskRunStatus.SUCCESS);
+        taskResponse.setExitStatusCode(0);
+        taskResponse.setProcessId(1);
+        when(shellCommandExecutor.run(any(), eq(taskCallBack))).thenReturn(taskResponse);
+
+        dataxTask.handle(taskCallBack);
+        Assertions.assertEquals(0, dataxTask.getExitStatusCode());
+
+        File jsonFile = new File("/tmp/execution/app-id-multi-resource_job.json");
+        String generated = FileUtils.readFile2Str(Files.newInputStream(jsonFile.toPath()));
+        Assertions.assertTrue(generated.contains("mysqlreader"),
+                "generated job file should carry the .json resource content, was: " + generated);
+        Assertions.assertFalse(generated.contains("keytab-binary-not-json"),
+                "generated job file must not read the auxiliary keytab as the job definition");
+        Assertions.assertTrue(jsonFile.delete());
+    }
+
     private TaskExecutionContext buildTestTaskExecutionContext() {
         TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
         taskExecutionContext.setTaskAppId("app-id");

dolphinscheduler-ui/src/locales/en_US/project.ts

@@ -645,6 +645,8 @@ export default {
     or: 'or',
     datax_custom_template: 'Custom Template',
     datax_json_template: 'JSON',
+    datax_custom_json_resource_tips:
+      'When the custom JSON is empty, attach exactly one .json resource file that carries the DataX job definition.',
     datax_target_datasource_type: 'Target Datasource Types',
     datax_target_database: 'Target Database',
     datax_target_table: 'Target Table',

dolphinscheduler-ui/src/locales/zh_CN/project.ts

@@ -626,6 +626,8 @@ export default {
     or: '或',
     datax_custom_template: '自定义模板',
     datax_json_template: 'JSON',
+    datax_custom_json_resource_tips:
+      '当自定义 JSON 为空时, 需要且仅需要附加一个携带 DataX 任务定义的 .json 资源文件。',
     datax_target_datasource_type: '目标源类型',
     datax_target_database: '目标源实例',
     datax_target_table: '目标表',