kestra-io · mgabelle · Jan 14, 2025 · Jan 9, 2025
diff --git a/core/src/main/java/io/kestra/plugin/core/debug/Echo.java b/core/src/main/java/io/kestra/plugin/core/debug/Echo.java
@@ -1,8 +1,7 @@
 package io.kestra.plugin.core.debug;
 
-import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.property.Property;
 import io.kestra.plugin.core.log.Log;
-import io.micronaut.core.annotation.NonNull;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
@@ -14,8 +13,9 @@
 import io.kestra.core.runners.RunContext;
 import org.slf4j.event.Level;
 
-import jakarta.validation.constraints.NotBlank;
-
+/**
+ * @deprecated
+ */
 @SuperBuilder
 @ToString
 @EqualsAndHashCode
@@ -46,20 +46,16 @@
 )
 @Deprecated
 public class Echo extends Task implements RunnableTask<VoidOutput> {
-    @NonNull
-    @NotBlank
-    @PluginProperty(dynamic = true)
-    private String format;
+    private Property<String> format;
 
     @Builder.Default
-    @PluginProperty
-    private Level level = Level.INFO;
+    private Property<Level> level = Property.of(Level.INFO);
 
     @Override
     public VoidOutput run(RunContext runContext) throws Exception {
         Log log = Log.builder()
-            .level(this.level)
-            .message(this.format)
+            .level(runContext.render(this.level).as(Level.class).orElseThrow())
+            .message(runContext.render(this.format).as(String.class).orElse(null))
             .build();
         log.run(runContext);
         return null;

diff --git a/core/src/main/java/io/kestra/plugin/core/debug/Return.java b/core/src/main/java/io/kestra/plugin/core/debug/Return.java
@@ -1,12 +1,12 @@
 package io.kestra.plugin.core.debug;
 
 import io.kestra.core.models.annotations.Metric;
+import io.kestra.core.models.property.Property;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 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.executions.metrics.Counter;
 import io.kestra.core.models.executions.metrics.Timer;
 import io.kestra.core.models.tasks.RunnableTask;
@@ -24,8 +24,10 @@
 @NoArgsConstructor
 @Schema(
     title = "Return a value for debugging purposes.",
-    description = "This task is mostly useful for troubleshooting.\n\n" +
-        "It allows you to return some templated functions, inputs or outputs."
+    description = """
+        This task is mostly useful for troubleshooting.
+
+        It allows you to return some templated functions, inputs or outputs."""
 )
 @Plugin(
     examples = {
@@ -51,16 +53,15 @@ public class Return extends Task implements RunnableTask<Return.Output> {
     @Schema(
         title = "The templated string to render."
     )
-    @PluginProperty(dynamic = true)
-    private String format;
+    private Property<String> format;
 
     @Override
     public Return.Output run(RunContext runContext) throws Exception {
         long start = System.nanoTime();
 
         Logger logger = runContext.logger();
 
-        String render = runContext.render(format);
+        String render = runContext.render(format).as(String.class).orElse(null);
         logger.debug(render);
 
         long end = System.nanoTime();

diff --git a/core/src/test/java/io/kestra/core/models/flows/FlowTest.java b/core/src/test/java/io/kestra/core/models/flows/FlowTest.java
@@ -1,6 +1,7 @@
 package io.kestra.core.models.flows;
 
 import io.kestra.core.exceptions.InternalException;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.validations.ModelValidator;
 import io.kestra.core.serializers.YamlParser;
@@ -114,13 +115,13 @@ void updateTask() throws InternalException {
         Flow updated = flow.updateTask("1-2-2_return", Return.builder()
             .id("1-2-2_return")
             .type(Return.class.getName())
-            .format("{{task.id}}")
+            .format(new Property<>("{{task.id}}"))
             .build()
         );
 
         Task findUpdated = updated.findTaskByTaskId("1-2-2_return");
 
-        assertThat(((Return) findUpdated).getFormat(), is("{{task.id}}"));
+        assertThat(((Return) findUpdated).getFormat().toString(), is("{{task.id}}"));
     }
 
     @Test

diff --git a/core/src/test/java/io/kestra/core/models/flows/FlowWithSourceTest.java b/core/src/test/java/io/kestra/core/models/flows/FlowWithSourceTest.java
@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import io.kestra.core.models.Label;
+import io.kestra.core.models.property.Property;
 import io.kestra.plugin.core.condition.Expression;
 import io.kestra.core.models.flows.input.StringInput;
 import io.kestra.core.models.listeners.Listener;
@@ -28,10 +29,10 @@ void source() throws JsonProcessingException {
                 Return.builder()
                     .id(IdUtils.create())
                     .type(Return.class.getName())
-                    .format("123456789 \n123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789\n" +
+                    .format(Property.of("123456789 \n123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789\n" +
                         "123456789 \n" +
                         "123456789 \n" +
-                        "123456789     \n")
+                        "123456789     \n"))
                     .build()
             ))
             .build();

diff --git a/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java b/core/src/test/java/io/kestra/core/repositories/AbstractExecutionRepositoryTest.java
@@ -14,6 +14,7 @@
 import io.kestra.core.models.executions.statistics.Flow;
 import io.kestra.core.models.flows.FlowScope;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.NamespaceUtils;
@@ -63,17 +64,17 @@ public static Execution.ExecutionBuilder builder(State.Type state, String flowId
 
         List<TaskRun> taskRuns = Arrays.asList(
             TaskRun.of(execution.build(), ResolvedTask.of(
-                    Return.builder().id("first").type(Return.class.getName()).format("test").build())
+                    Return.builder().id("first").type(Return.class.getName()).format(Property.of("test")).build())
                 )
                 .withState(State.Type.SUCCESS),
             spyTaskRun(TaskRun.of(execution.build(), ResolvedTask.of(
-                        Return.builder().id("second").type(Return.class.getName()).format("test").build())
+                        Return.builder().id("second").type(Return.class.getName()).format(Property.of("test")).build())
                     )
                     .withState(state),
                 state
             ),
             TaskRun.of(execution.build(), ResolvedTask.of(
-                Return.builder().id("third").type(Return.class.getName()).format("test").build())).withState(state)
+                Return.builder().id("third").type(Return.class.getName()).format(Property.of("test")).build())).withState(state)
         );
 
         if (flowId == null) {

diff --git a/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java b/core/src/test/java/io/kestra/core/repositories/AbstractFlowRepositoryTest.java
@@ -9,6 +9,7 @@
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.*;
 import io.kestra.core.models.flows.input.StringInput;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.schedulers.AbstractSchedulerTest;
 import io.kestra.core.serializers.JacksonMapper;
@@ -74,7 +75,7 @@ protected void init() throws IOException, URISyntaxException {
         return Flow.builder()
             .id(flowId)
             .namespace("io.kestra.unittest")
-            .tasks(Collections.singletonList(Return.builder().id(taskId).type(Return.class.getName()).format("test").build()));
+            .tasks(Collections.singletonList(Return.builder().id(taskId).type(Return.class.getName()).format(Property.of("test")).build()));
     }
 
     @Test
@@ -141,7 +142,7 @@ protected void revision() throws JsonProcessingException {
         Flow first = Flow.builder()
             .id(flowId)
             .namespace("io.kestra.unittest")
-            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()))
+            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format(Property.of("test")).build()))
             .inputs(ImmutableList.of(StringInput.builder().type(Type.STRING).id("a").build()))
             .build();
         // create with repository
@@ -384,7 +385,7 @@ void updateConflict() {
             .id(flowId)
             .namespace("io.kestra.unittest")
             .inputs(ImmutableList.of(StringInput.builder().type(Type.STRING).id("a").build()))
-            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()))
+            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format(Property.of("test")).build()))
             .build();
 
         Flow save = flowRepository.create(flow, flow.generateSource(), pluginDefaultService.injectDefaults(flow.withSource(flow.generateSource())));
@@ -396,7 +397,7 @@ void updateConflict() {
                 .id(IdUtils.create())
                 .namespace("io.kestra.unittest2")
                 .inputs(ImmutableList.of(StringInput.builder().type(Type.STRING).id("b").build()))
-                .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()))
+                .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format(Property.of("test")).build()))
                 .build();
             ;
 
@@ -422,7 +423,7 @@ void removeTrigger() throws TimeoutException, QueueException {
                 .id("sleep")
                 .type(AbstractSchedulerTest.UnitTest.class.getName())
                 .build()))
-            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()))
+            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format(Property.of("test")).build()))
             .build();
 
         flow = flowRepository.create(flow, flow.generateSource(), pluginDefaultService.injectDefaults(flow.withSource(flow.generateSource())));
@@ -432,7 +433,7 @@ void removeTrigger() throws TimeoutException, QueueException {
             Flow update = Flow.builder()
                 .id(flowId)
                 .namespace("io.kestra.unittest")
-                .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()))
+                .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format(Property.of("test")).build()))
                 .build();
             ;
 
@@ -460,7 +461,7 @@ void removeTriggerDelete() throws TimeoutException, QueueException {
                 .id("sleep")
                 .type(AbstractSchedulerTest.UnitTest.class.getName())
                 .build()))
-            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()))
+            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format(Property.of("test")).build()))
             .build();
 
         Flow save = flowRepository.create(flow, flow.generateSource(), pluginDefaultService.injectDefaults(flow.withSource(flow.generateSource())));
@@ -533,7 +534,7 @@ protected void lastRevision() {
             .tenantId(tenantId)
             .id(flowId)
             .namespace(namespace)
-            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()))
+            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format(Property.of("test")).build()))
             .inputs(ImmutableList.of(StringInput.builder().type(Type.STRING).id("a").build()))
             .build();
         // create with repository

diff --git a/core/src/test/java/io/kestra/core/repositories/AbstractTemplateRepositoryTest.java b/core/src/test/java/io/kestra/core/repositories/AbstractTemplateRepositoryTest.java
@@ -2,6 +2,7 @@
 
 import io.kestra.core.events.CrudEvent;
 import io.kestra.core.events.CrudEventType;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.templates.Template;
 import io.kestra.plugin.core.debug.Return;
 import io.kestra.core.utils.IdUtils;
@@ -41,7 +42,7 @@ protected void init() throws IOException, URISyntaxException {
         return Template.builder()
             .id(IdUtils.create())
             .namespace(namespace == null ? "kestra.test" : namespace)
-            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format("test").build()));
+            .tasks(Collections.singletonList(Return.builder().id("test").type(Return.class.getName()).format(Property.of("test")).build()));
     }
 
     @Test

diff --git a/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java b/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
@@ -9,6 +9,7 @@
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.repositories.ExecutionRepositoryInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.repositories.LogRepositoryInterface;
@@ -83,7 +84,7 @@ void restartSimpleRevision() throws Exception {
                 Return.builder()
                     .id("a")
                     .type(Return.class.getName())
-                    .format("replace")
+                    .format(Property.of("replace"))
                     .build()
             ),
             JacksonMapper.ofYaml().writeValueAsString(flow),

diff --git a/core/src/test/java/io/kestra/core/runners/FlowListenersTest.java b/core/src/test/java/io/kestra/core/runners/FlowListenersTest.java
@@ -1,6 +1,7 @@
 package io.kestra.core.runners;
 
 import io.kestra.core.models.flows.FlowWithSource;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.services.PluginDefaultService;
 import io.kestra.core.junit.annotations.KestraTest;
 import lombok.SneakyThrows;
@@ -35,7 +36,7 @@ protected static FlowWithSource create(String flowId, String taskId) {
             .tasks(Collections.singletonList(Return.builder()
                 .id(taskId)
                 .type(Return.class.getName())
-                .format("test")
+                .format(Property.of("test"))
                 .build()))
             .build();
         return flow.withSource(flow.generateSource());

diff --git a/core/src/test/java/io/kestra/core/runners/SkipExecutionCaseTest.java b/core/src/test/java/io/kestra/core/runners/SkipExecutionCaseTest.java
@@ -3,6 +3,7 @@
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
@@ -62,7 +63,7 @@ private Flow createFlow() {
             .tasks(Collections.singletonList(Return.builder()
                 .id("test")
                 .type(Return.class.getName())
-                .format("{{ inputs.testInputs }}")
+                .format(new Property<>("{{ inputs.testInputs }}"))
                 .build()))
             .build();
     }

diff --git a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
@@ -7,6 +7,7 @@
 import io.kestra.core.models.executions.ExecutionTrigger;
 import io.kestra.core.models.flows.*;
 import io.kestra.core.models.flows.input.StringInput;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.PollingTriggerInterface;
@@ -89,7 +90,7 @@ protected static FlowWithSource createFlow(List<AbstractTrigger> triggers, List<
             .tasks(Collections.singletonList(Return.builder()
                 .id("test")
                 .type(Return.class.getName())
-                .format("{{ inputs.testInputs }}")
+                .format(new Property<>("{{ inputs.testInputs }}"))
                 .build()));
 
         if (list != null) {

diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerTriggerChangeTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerTriggerChangeTest.java
@@ -5,8 +5,8 @@
 import io.kestra.core.models.executions.ExecutionKilled;
 import io.kestra.core.models.executions.ExecutionKilledTrigger;
 import io.kestra.core.models.executions.LogEntry;
-import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithSource;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.PollingTriggerInterface;
 import io.kestra.core.models.triggers.TriggerContext;
@@ -34,6 +34,7 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
 
 public class SchedulerTriggerChangeTest extends AbstractSchedulerTest {
     @Inject
@@ -66,7 +67,7 @@ public static FlowWithSource createFlow(Duration sleep) {
             .tasks(Collections.singletonList(Return.builder()
                 .id("test")
                 .type(Return.class.getName())
-                .format("{{ inputs.testInputs }}")
+                .format(new Property<>("{{ inputs.testInputs }}"))
                 .build())
             )
             .build();
@@ -113,7 +114,7 @@ void run() throws Exception {
             WorkerTrigger workerTrigger = worker.getWorkerThreadTasks()
                 .stream()
                 .filter(workerJob -> workerJob instanceof WorkerTrigger)
-                .map(workerJob -> (WorkerTrigger) workerJob)
+                .map(WorkerTrigger.class::cast)
                 .findFirst()
                 .orElseThrow();