Repository: kestra-io/kestra (java)

Issue:
Add a new Runnable task — DynamicTask allowing to template the entire task definition even for non-dynamic properties
Sometimes, we can't make specific properties dynamic because we want to offer type validation for integers, floats, booleans, or complex objects. This issue keeps coming more and more across various user requests (Databricks, Kubernetes PodCreate, and more).

The challenge is that: 
- _some_ users want to make almost every property dynamic. They care about making all inputs dynamic more than type safety. 
- for another type of user, that type safety and syntax/type validation while typing the declarative flow definition are important.


To offer a balanced approach, we may offer a new core task that allows users to fully dynamically generate a task definition. 


Name TBD, possible options:
- TemplatedTask
- DynamicTask

Given we disable [recursive rendering](https://kestra.io/docs/migrations/recursive-rendering#how-to-keep-the-previous-behavior) in the app globally, here it's also set to false by default.

```yaml
id: template
namespace: dev

inputs:
  - name: restartPolicy
    type: STRING

  - name: taskId
    type: STRING

tasks:
  - id: dynamic
    type: io.kestra.core.tasks.templating.TemplatedTask
    spec: |
      id: "{{ inputs.taskId }}"
      type: io.kestra.plugin.kubernetes.PodCreate
      namespace: default
      metadata:
        labels:
          my-label: my-value
      spec:
        containers:
        - name: unittest
          image: debian:stable-slim
          command: 
            - 'bash' 
            - '-c'
            - 'for i in {1..10}; do echo $i; sleep 0.1; done'
          {% if inputs.restartPolicy}
          restartPolicy: "{{ inputs.restartPolicy }}"
          {% endif %}
```

Note how, in this example, even task ID or task type can be dynamically generated at runtime based on user inputs, environment variables and more. 

In this example, the flow topology view will show the `DynamicTask` as a task type, while the execution topology will dynamically render that template into the right task type. 

Pull request description:
feat(core): templated task
Fixes #2962

Patch:
diff --git a/core/src/main/java/io/kestra/core/tasks/templating/TemplatedTask.java b/core/src/main/java/io/kestra/core/tasks/templating/TemplatedTask.java
new file mode 100644
index 000000000..2a6398f26
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/tasks/templating/TemplatedTask.java
@@ -0,0 +1,64 @@
+package io.kestra.core.tasks.templating;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.annotations.Example;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.tasks.*;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.serializers.JacksonMapper;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+@Schema(
+    title = "This task's `spec` property allows you to fully template all task properties using Kestra's Pebble templating. This way, all task properties and their values can be dynamically rendered based on your custom inputs, variables, and outputs from other tasks!"
+)
+@Plugin(
+    examples = {
+        @Example(
+            code = {
+                """
+                    spec: |
+                      type: io.kestra.plugin.fs.http.Download
+                      {{ task.property }}: {{ task.value }}"""
+            }
+        )
+    }
+)
+public class TemplatedTask extends Task implements RunnableTask<Output> {
+    private static final ObjectMapper OBJECT_MAPPER = JacksonMapper.ofYaml();
+
+    @NotNull
+    @PluginProperty(dynamic = true)
+    @Schema(title = "The templated task specification")
+    private String spec;
+
+    @Override
+    public Output run(RunContext runContext) throws Exception {
+        String taskSpec = runContext.render(this.spec);
+        try {
+            Task task = OBJECT_MAPPER.readValue(taskSpec, Task.class);
+            if (task instanceof TemplatedTask) {
+                throw new IllegalArgumentException("The templated task cannot be of type 'io.kestra.core.tasks.templating.TemplatedTask'");
+            }
+            if (task instanceof RunnableTask<?> runnableTask) {
+                return runnableTask.run(runContext);
+            }
+            throw new IllegalArgumentException("The templated task must be a runnable task");
+        } catch (JsonProcessingException e) {
+            throw new IllegalVariableEvaluationException(e);
+        }
+    }
+}
diff --git a/core/src/main/resources/icons/io.kestra.core.tasks.templating.svg b/core/src/main/resources/icons/io.kestra.core.tasks.templating.svg
new file mode 100644
index 000000000..42c9dbdd3
--- /dev/null
+++ b/core/src/main/resources/icons/io.kestra.core.tasks.templating.svg
@@ -0,0 +1,2 @@
+<?xml version="1.0" encoding="utf-8"?>
+<svg fill="#000000" width="800px" height="800px" viewBox="0 0 100 100" xmlns="http://www.w3.org/2000/svg"><path d="M69.6,59.3A10.2,10.2,0,1,0,79.8,69.5,10.24,10.24,0,0,0,69.6,59.3Zm6.6,7.4-7.5,7.5a1.45,1.45,0,0,1-2,0L63,70.6a.67.67,0,0,1,0-1l1-1a.67.67,0,0,1,1,0l2.6,2.6,6.5-6.5a.67.67,0,0,1,1,0l1,1A.76.76,0,0,1,76.2,66.7Z"/><path d="M44.5,30H62.3a1.58,1.58,0,0,0,1.6-1.6V25.1A4.91,4.91,0,0,0,59,20.2H47.7a4.91,4.91,0,0,0-4.9,4.9v3.3A1.73,1.73,0,0,0,44.5,30Z"/><path d="M56.1,73.3H31.4a4.89,4.89,0,0,1-4.8-4.9v-34H25a4.89,4.89,0,0,0-4.8,4.9V74.7A4.89,4.89,0,0,0,25,79.6H59.9A12.11,12.11,0,0,1,56.1,73.3Z"/><path d="M35.6,69.5H55.7a15.14,15.14,0,0,1,.7-3.7,13.68,13.68,0,0,1,2.3-4.5H49.3a1.58,1.58,0,0,1-1.6-1.6V58a1.58,1.58,0,0,1,1.6-1.6H65.5c.2,0,.3,0,.4.1a13.61,13.61,0,0,1,3.6-.5A13.89,13.89,0,0,1,76,57.6V29.2a4.91,4.91,0,0,0-4.9-4.9H69.5a.74.74,0,0,0-.8.8v3.3A6.57,6.57,0,0,1,62.2,35H44.5A6.64,6.64,0,0,1,38,28.4V25.1a.74.74,0,0,0-.8-.8H35.6a4.91,4.91,0,0,0-4.9,4.9V64.5A5.06,5.06,0,0,0,35.6,69.5Zm12.1-28a1.58,1.58,0,0,1,1.6-1.6H65.5a1.58,1.58,0,0,1,1.6,1.6v1.6a1.58,1.58,0,0,1-1.6,1.6H49.4a1.58,1.58,0,0,1-1.6-1.6V41.5Zm0,8.2a1.58,1.58,0,0,1,1.6-1.6H65.5a1.58,1.58,0,0,1,1.6,1.6v1.6a1.58,1.58,0,0,1-1.6,1.6H49.4a1.58,1.58,0,0,1-1.6-1.6V49.7Zm-8-8.2a1.58,1.58,0,0,1,1.6-1.6h1.6a1.58,1.58,0,0,1,1.6,1.6v1.6a1.58,1.58,0,0,1-1.6,1.6H41.3a1.58,1.58,0,0,1-1.6-1.6Zm0,8.2a1.58,1.58,0,0,1,1.6-1.6h1.6a1.58,1.58,0,0,1,1.6,1.6v1.6a1.58,1.58,0,0,1-1.6,1.6H41.3a1.58,1.58,0,0,1-1.6-1.6Zm0,8.3a1.58,1.58,0,0,1,1.6-1.6h1.6A1.58,1.58,0,0,1,44.5,58v1.6a1.58,1.58,0,0,1-1.6,1.6H41.3a1.58,1.58,0,0,1-1.6-1.6Z"/></svg>
\ No newline at end of file


Test patch:
diff --git a/core/src/test/java/io/kestra/core/tasks/templating/TemplatedTaskTest.java b/core/src/test/java/io/kestra/core/tasks/templating/TemplatedTaskTest.java
new file mode 100644
index 000000000..50f651904
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/templating/TemplatedTaskTest.java
@@ -0,0 +1,73 @@
+package io.kestra.core.tasks.templating;
+
+import io.kestra.core.models.tasks.Output;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.runners.RunContextFactory;
+import io.kestra.core.tasks.debugs.Return;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.core.Is.is;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+@MicronautTest
+class TemplatedTaskTest {
+
+    @Inject
+    private RunContextFactory runContextFactory;
+
+    @Test
+    void templatedType() throws Exception {
+        RunContext runContext = runContextFactory.of(Map.of("type", "io.kestra.core.tasks.debugs.Return"));
+        TemplatedTask templatedTask = TemplatedTask.builder()
+            .id("template")
+            .type(TemplatedTask.class.getName())
+            .spec("""
+                type: {{ type }}
+                format: It's alive!""")
+            .build();
+
+        Output output = templatedTask.run(runContext);
+
+        assertThat(output, notNullValue());
+        assertThat(output, instanceOf(Return.Output.class));
+        assertThat(((Return.Output)output).getValue(), is("It's alive!"));
+    }
+
+    @Test
+    void templatedFlowable() {
+        RunContext runContext = runContextFactory.of();
+        TemplatedTask templatedTask = TemplatedTask.builder()
+            .id("template")
+            .type(TemplatedTask.class.getName())
+            .spec("""
+                type: io.kestra.core.tasks.flows.Pause
+                delay: PT10S""")
+            .build();
+
+        var exception = assertThrows(IllegalArgumentException.class, () -> templatedTask.run(runContext));
+        assertThat(exception.getMessage(), is("The templated task must be a runnable task"));
+    }
+
+    @Test
+    void templatedTemplated() {
+        RunContext runContext = runContextFactory.of();
+        TemplatedTask templatedTask = TemplatedTask.builder()
+            .id("template")
+            .type(TemplatedTask.class.getName())
+            .spec("""
+                type: io.kestra.core.tasks.templating.TemplatedTask
+                spec: whatever""")
+            .build();
+
+        var exception = assertThrows(IllegalArgumentException.class, () -> templatedTask.run(runContext));
+        assertThat(exception.getMessage(), is("The templated task cannot be of type 'io.kestra.core.tasks.templating.TemplatedTask'"));
+    }
+
+}
\ No newline at end of file