Repository: kestra-io/kestra (java)

Issue:
Flow level pluginDefaults should take precendence over global or namespace-level pluginDefaults (unless `forced: true`)
### Describe the issue

The code treats the global ones before the flow ones: https://github.com/kestra-io/kestra/blob/302210da3cfb0503435caf7ba0fdac104ecea897/core/src/main/java/io/kestra/core/services/PluginDefaultService.java#L64

This can be verified:

```bash
diff --git a/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java b/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java
index 995b85ffb..d1ba40a0b 100644
--- a/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java
+++ b/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java
@@ -99,6 +99,34 @@ class PluginDefaultServiceTest {
         ), result);
     }
 
+    @Test
+    void flowDefaultsOverrideGlobalDefaults() {
+        final DefaultTester task = DefaultTester.builder()
+            .id("test")
+            .type(DefaultTester.class.getName())
+            .build();
+        final PluginDefault flowDefault = new PluginDefault(DefaultTester.class.getName(), false, ImmutableMap.of(
+            "set", 1
+        ));
+        final PluginDefault globalDefault = new PluginDefault(DefaultTester.class.getName(), false, ImmutableMap.of(
+            "set", 2
+        ));
+
+        final Flow flowWithPluginDefault = Flow.builder()
+            .tasks(Collections.singletonList(task))
+            .pluginDefaults(List.of(flowDefault))
+            .build();
+
+        final PluginGlobalDefaultConfiguration pluginGlobalDefaultConfiguration = new PluginGlobalDefaultConfiguration();
+        pluginGlobalDefaultConfiguration.defaults = List.of(globalDefault);
+
+        pluginDefaultService.pluginGlobalDefault = pluginGlobalDefaultConfiguration;
+
+        final Flow injected = pluginDefaultService.injectDefaults(flowWithPluginDefault);
+
+        assertThat(((DefaultTester) injected.getTasks().getFirst()).getSet(), is(1));
+    }
+
     @Test
     public void injectFlowAndGlobals() {
         DefaultTester task = DefaultTester.builder()

```


The expected order:
1. Global defaults are evaluated
2. Namespace-level `pluginDefaults` are evaluated and take precedence over global once 
3. Flow-level `pluginDefaults` are evaluated and take precedence over those at namespace level 
4. Task-level properties are evaluated and they take precedence and can override values set in `pluginDefaults` at all levels

Only with `forced: true`, namespace-level defaults would take precedence over those defined at flow or task level

### Environment

- Kestra Version: develop


Pull request:
fix(core): make flow plugin defaults override global ones
### What changes are being made and why?

This handles the OSS precedence. Sadly, EE namespaces remain to be verified.

closes #4897

Patch:
diff --git a/core/src/main/java/io/kestra/core/services/PluginDefaultService.java b/core/src/main/java/io/kestra/core/services/PluginDefaultService.java
index cfd2657de..4bbe85afc 100644
--- a/core/src/main/java/io/kestra/core/services/PluginDefaultService.java
+++ b/core/src/main/java/io/kestra/core/services/PluginDefaultService.java
@@ -68,6 +68,10 @@ public class PluginDefaultService {
     protected List<PluginDefault> mergeAllDefaults(Flow flow) {
         List<PluginDefault> list = new ArrayList<>();
 
+        if (flow.getPluginDefaults() != null) {
+            list.addAll(flow.getPluginDefaults());
+        }
+
         if (taskGlobalDefault != null && taskGlobalDefault.getDefaults() != null) {
             if (warnOnce.compareAndSet(false, true)) {
                 log.warn("Global Task Defaults are deprecated, please use Global Plugin Defaults instead via the 'kestra.plugins.defaults' property.");
@@ -79,10 +83,6 @@ public class PluginDefaultService {
             list.addAll(pluginGlobalDefault.getDefaults());
         }
 
-        if (flow.getPluginDefaults() != null) {
-            list.addAll(flow.getPluginDefaults());
-        }
-
         return list;
     }
 
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
index f40464a13..4c6e59819 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
@@ -664,7 +664,7 @@ public class ExecutionController {
         @Parameter(description = "The execution id") @PathVariable String executionId,
         @Parameter(description = "The internal storage uri") @QueryValue URI path
     ) throws IOException, URISyntaxException {
-        HttpResponse<StreamedFile> httpResponse = this.validateFile(executionId, path, "/api/v1/executions/{executionId}/file?path=" + path);
+        HttpResponse<StreamedFile> httpResponse = this.validateFile(executionId, path, "/api/v1/" + this.getTenant() + "executions/{executionId}/file?path=" + path);
         if (httpResponse != null) {
             return httpResponse;
         }
@@ -682,7 +682,7 @@ public class ExecutionController {
         @Parameter(description = "The execution id") @PathVariable String executionId,
         @Parameter(description = "The internal storage uri") @QueryValue URI path
     ) throws IOException {
-        HttpResponse<FileMetas> httpResponse = this.validateFile(executionId, path, "/api/v1/executions/{executionId}/file/metas?path=" + path);
+        HttpResponse<FileMetas> httpResponse = this.validateFile(executionId, path, "/api/v1/" + this.getTenant() + "executions/{executionId}/file/metas?path=" + path);
         if (httpResponse != null) {
             return httpResponse;
         }
@@ -1445,7 +1445,7 @@ public class ExecutionController {
         @Parameter(description = "The max row returns") @QueryValue @Nullable Integer maxRows,
         @Parameter(description = "The file encoding as Java charset name. Defaults to UTF-8", example = "ISO-8859-1") @QueryValue(defaultValue = "UTF-8") String encoding
     ) throws IOException {
-        this.validateFile(executionId, path, "/api/v1/executions/{executionId}/file?path=" + path);
+        this.validateFile(executionId, path, "/api/v1/" + this.getTenant() + "executions/{executionId}/file?path=" + path);
 
         String extension = FilenameUtils.getExtension(path.toString());
         Optional<Charset> charset;
@@ -1646,4 +1646,8 @@ public class ExecutionController {
         return flowRepository.findByNamespaceExecutable(tenantService.resolveTenant(), namespace);
     }
 
+    public String getTenant() {
+        return tenantService.resolveTenant() != null ? tenantService.resolveTenant() + "/" : "";
+    }
+
 }


Test patch:
diff --git a/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java b/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java
index 995b85ffb..3060f4ec3 100644
--- a/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java
+++ b/core/src/test/java/io/kestra/core/services/PluginDefaultServiceTest.java
@@ -25,12 +25,16 @@ import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -99,6 +103,51 @@ class PluginDefaultServiceTest {
         ), result);
     }
 
+    @ParameterizedTest
+    @MethodSource
+    void flowDefaultsOverrideGlobalDefaults(boolean flowDefaultForced, boolean globalDefaultForced, String fooValue, String barValue, String bazValue) {
+        final DefaultPrecedenceTester task = DefaultPrecedenceTester.builder()
+            .id("test")
+            .type(DefaultPrecedenceTester.class.getName())
+            .propBaz("taskValue")
+            .build();
+
+        final PluginDefault flowDefault = new PluginDefault(DefaultPrecedenceTester.class.getName(), flowDefaultForced, ImmutableMap.of(
+            "propBar", "flowValue",
+            "propBaz", "flowValue"
+        ));
+        final PluginDefault globalDefault = new PluginDefault(DefaultPrecedenceTester.class.getName(), globalDefaultForced, ImmutableMap.of(
+            "propFoo", "globalValue",
+            "propBar", "globalValue",
+            "propBaz", "globalValue"
+        ));
+
+        final Flow flowWithPluginDefault = Flow.builder()
+            .tasks(Collections.singletonList(task))
+            .pluginDefaults(List.of(flowDefault))
+            .build();
+
+        final PluginGlobalDefaultConfiguration pluginGlobalDefaultConfiguration = new PluginGlobalDefaultConfiguration();
+        pluginGlobalDefaultConfiguration.defaults = List.of(globalDefault);
+
+        pluginDefaultService.pluginGlobalDefault = pluginGlobalDefaultConfiguration;
+
+        final Flow injected = pluginDefaultService.injectDefaults(flowWithPluginDefault);
+
+        assertThat(((DefaultPrecedenceTester) injected.getTasks().getFirst()).getPropFoo(), is(fooValue));
+        assertThat(((DefaultPrecedenceTester) injected.getTasks().getFirst()).getPropBar(), is(barValue));
+        assertThat(((DefaultPrecedenceTester) injected.getTasks().getFirst()).getPropBaz(), is(bazValue));
+    }
+
+    private static Stream<Arguments> flowDefaultsOverrideGlobalDefaults() {
+        return Stream.of(
+            Arguments.of(false, false, "globalValue", "flowValue", "taskValue"),
+            Arguments.of(false, true, "globalValue", "globalValue", "globalValue"),
+            Arguments.of(true, false, "globalValue", "flowValue", "flowValue"),
+            Arguments.of(true, true, "globalValue", "flowValue", "flowValue")
+        );
+    }
+
     @Test
     public void injectFlowAndGlobals() {
         DefaultTester task = DefaultTester.builder()
@@ -297,4 +346,23 @@ class PluginDefaultServiceTest {
             private Map<String, String> val;
         }
     }
+
+    @SuperBuilder
+    @ToString
+    @EqualsAndHashCode
+    @Getter
+    @NoArgsConstructor
+    @Plugin(aliases = "io.kestra.core.services.DefaultPrecedenceTesterAlias")
+    public static class DefaultPrecedenceTester extends Task implements RunnableTask<VoidOutput> {
+        private String propFoo;
+
+        private String propBar;
+
+        private String propBaz;
+
+        @Override
+        public VoidOutput run(RunContext runContext) throws Exception {
+            return null;
+        }
+    }
 }
\ No newline at end of file