Repository: kestra-io/kestra (java)

Issue:
Hide flow without source from the trigger page 
![image](https://github.com/user-attachments/assets/f2905c3e-3fbd-4f71-a82d-b2ab14c45a58)

Since there is no source, nothing will happen, 
We should delete the trigger when the flow is deleted. 

Pull request description:
fix(scheduler): delete trigger when flow is not found
close #7312

Patch:
diff --git a/.plugins b/.plugins
index b355b1de2..6247da3f5 100644
--- a/.plugins
+++ b/.plugins
@@ -40,6 +40,7 @@
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-db2:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-duckdb:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-druid:LATEST
+#plugin-jdbc:io.kestra.plugin:plugin-jdbc-mariadb:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-mysql:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-oracle:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-pinot:LATEST
diff --git a/cli/src/main/resources/application.yml b/cli/src/main/resources/application.yml
index 37d7ea34f..60836746f 100644
--- a/cli/src/main/resources/application.yml
+++ b/cli/src/main/resources/application.yml
@@ -135,6 +135,11 @@ kestra:
       initial-delay: 1h
       fixed-delay: 1h
       retention: 7d
+      types:
+        - type : io.kestra.core.models.executions.LogEntry
+          retention: 1h
+        - type: io.kestra.core.models.executions.MetricEntry
+          retention: 1h
 
   plugins:
     repositories:
diff --git a/core/src/main/java/io/kestra/core/http/HttpRequest.java b/core/src/main/java/io/kestra/core/http/HttpRequest.java
index 0518e0dec..8bc902ab4 100644
--- a/core/src/main/java/io/kestra/core/http/HttpRequest.java
+++ b/core/src/main/java/io/kestra/core/http/HttpRequest.java
@@ -142,12 +142,18 @@ public class HttpRequest {
     public abstract static class RequestBody {
         public abstract HttpEntity to() throws IOException;
 
+        public abstract Object getContent() throws IOException;
+
+        public abstract Charset getCharset() throws IOException;
+
+        public abstract String getContentType() throws IOException;
+
         public static RequestBody from(HttpEntity entity) throws IOException {
             if (entity == null) {
                 return null;
             }
 
-            Charset charset = Charset.forName(entity.getContentEncoding());
+            Charset charset = entity.getContentEncoding() != null ? Charset.forName(entity.getContentEncoding()) : StandardCharsets.UTF_8;
 
             if (entity.getContentType().equals(ContentType.APPLICATION_OCTET_STREAM.getMimeType())) {
                 return ByteArrayRequestBody.builder()
@@ -172,10 +178,15 @@ public class HttpRequest {
                     .build();
             }
 
-            throw new IllegalArgumentException("Unsupported Content-Type: " + entity.getContentType());
+            return ByteArrayRequestBody.builder()
+                .charset(charset)
+                .contentType(entity.getContentType())
+                .content(entity.getContent().readAllBytes())
+                .build();
         }
     }
 
+    @Getter
     @AllArgsConstructor
     @SuperBuilder
     public static class InputStreamRequestBody extends RequestBody {
@@ -192,6 +203,7 @@ public class HttpRequest {
         }
     }
 
+    @Getter
     @AllArgsConstructor
     @SuperBuilder
     public static class StringRequestBody extends RequestBody {
@@ -208,6 +220,7 @@ public class HttpRequest {
         }
     }
 
+    @Getter
     @AllArgsConstructor
     @SuperBuilder
     public static class ByteArrayRequestBody extends RequestBody {
@@ -224,6 +237,7 @@ public class HttpRequest {
         }
     }
 
+    @Getter
     @AllArgsConstructor
     @SuperBuilder
     public static class JsonRequestBody extends RequestBody {
@@ -232,6 +246,11 @@ public class HttpRequest {
 
         private Object content;
 
+        @Override
+        public String getContentType() throws IOException {
+            return ContentType.APPLICATION_JSON.getMimeType();
+        }
+
         public HttpEntity to() throws IOException {
             try {
                 return new StringEntity(
@@ -244,6 +263,7 @@ public class HttpRequest {
         }
     }
 
+    @Getter
     @AllArgsConstructor
     @SuperBuilder
     public static class UrlEncodedRequestBody extends RequestBody {
@@ -252,6 +272,11 @@ public class HttpRequest {
 
         private Map<String, Object> content;
 
+        @Override
+        public String getContentType() throws IOException {
+            return ContentType.APPLICATION_FORM_URLENCODED.getMimeType();
+        }
+
         public HttpEntity to() throws IOException {
             return new UrlEncodedFormEntity(
                 this.content .entrySet()
@@ -263,6 +288,7 @@ public class HttpRequest {
         }
     }
 
+    @Getter
     @AllArgsConstructor
     @SuperBuilder
     public static class MultipartRequestBody extends RequestBody {
@@ -271,6 +297,11 @@ public class HttpRequest {
 
         private Map<String, Object> content;
 
+        @Override
+        public String getContentType() throws IOException {
+            return ContentType.MULTIPART_MIXED.getMimeType();
+        }
+
         public HttpEntity to() throws IOException {
             MultipartEntityBuilder builder = MultipartEntityBuilder
                 .create()
diff --git a/core/src/main/java/io/kestra/core/models/flows/Flow.java b/core/src/main/java/io/kestra/core/models/flows/Flow.java
index 4db020e5a..d1172ecaf 100644
--- a/core/src/main/java/io/kestra/core/models/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/models/flows/Flow.java
@@ -21,6 +21,7 @@ import io.kestra.core.models.tasks.FlowableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
@@ -176,6 +177,14 @@ public class Flow extends AbstractFlow implements HasUID {
         );
     }
 
+    public static String uid(Trigger trigger) {
+        return IdUtils.fromParts(
+            trigger.getTenantId(),
+            trigger.getNamespace(),
+            trigger.getFlowId()
+        );
+    }
+
     public static String uidWithoutRevision(Execution execution) {
         return IdUtils.fromParts(
             execution.getTenantId(),
diff --git a/core/src/main/java/io/kestra/core/models/property/Property.java b/core/src/main/java/io/kestra/core/models/property/Property.java
index 9677f6336..328ae2c94 100644
--- a/core/src/main/java/io/kestra/core/models/property/Property.java
+++ b/core/src/main/java/io/kestra/core/models/property/Property.java
@@ -22,6 +22,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
+import static io.kestra.core.utils.Rethrow.throwFunction;
+
 /**
  * Define a plugin properties that will be rendered and converted to a target type at use time.
  *
@@ -136,12 +138,29 @@ public class Property<T> {
      *
      * @see io.kestra.core.runners.RunContextProperty#asList(Class, Map)
      */
+    @SuppressWarnings("unchecked")
     public static <T, I> T asList(Property<T> property, RunContext runContext, Class<I> itemClazz, Map<String, Object> variables) throws IllegalVariableEvaluationException {
         if (property.value == null) {
-            String rendered =  runContext.render(property.expression, variables);
             JavaType type = MAPPER.getTypeFactory().constructCollectionLikeType(List.class, itemClazz);
             try {
-                property.value = MAPPER.readValue(rendered, type);
+                String trimmedExpression = property.expression.trim();
+                // We need to detect if the expression is already a list or if it's a pebble expression (for eg. referencing a variable containing a list).
+                // Doing that allows us to, if it's an expression, first render then read it as a list.
+                if (trimmedExpression.startsWith("{{") && trimmedExpression.endsWith("}}")) {
+                    property.value = MAPPER.readValue(runContext.render(property.expression, variables), type);
+                }
+                // Otherwise if it's already a list, we read it as a list first then render it from run context which handle list rendering by rendering each item of the list
+                else {
+                    List<?> asRawList = MAPPER.readValue(runContext.render(property.expression, variables), List.class);
+                    property.value = (T) asRawList.stream()
+                        .map(throwFunction(item -> {
+                            if (item instanceof String str) {
+                                return MAPPER.convertValue(str, itemClazz);
+                            }
+                            return item;
+                        }))
+                        .toList();
+                }
             } catch (JsonProcessingException e) {
                 throw new IllegalVariableEvaluationException(e);
             }
diff --git a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
index a2cb55160..a9de32a3f 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -30,7 +30,7 @@ import io.kestra.core.utils.Await;
 import io.kestra.core.utils.Either;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.ListUtils;
-import io.kestra.core.models.triggers.RecoverMissedSchedules;
+import io.kestra.core.models.flows.Flow;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.context.event.ApplicationEventPublisher;
 import io.micronaut.inject.qualifiers.Qualifiers;
@@ -70,7 +70,8 @@ public abstract class AbstractScheduler implements Scheduler, Service {
     private final QueueInterface<WorkerJob> workerTaskQueue;
     private final WorkerTriggerResultQueueInterface workerTriggerResultQueue;
     private final QueueInterface<ExecutionKilled> executionKilledQueue;
-    @SuppressWarnings("rawtypes") private final Optional<QueueInterface> clusterEventQueue;
+    @SuppressWarnings("rawtypes")
+    private final Optional<QueueInterface> clusterEventQueue;
     protected final FlowListenersInterface flowListeners;
     private final RunContextFactory runContextFactory;
     private final RunContextInitializer runContextInitializer;
@@ -408,6 +409,16 @@ public abstract class AbstractScheduler implements Scheduler, Service {
     private List<FlowWithTriggers> computeSchedulable(List<FlowWithSource> flows, List<Trigger> triggerContextsToEvaluate, ScheduleContextInterface scheduleContext) {
         List<String> flowToKeep = triggerContextsToEvaluate.stream().map(Trigger::getFlowId).toList();
 
+        triggerContextsToEvaluate.stream()
+            .filter(trigger -> !flows.stream().map(FlowWithSource::uidWithoutRevision).toList().contains(Flow.uid(trigger)))
+            .forEach(trigger -> {
+                try {
+                    this.triggerState.delete(trigger);
+                } catch (QueueException e) {
+                    log.error("Unable to delete the trigger: {}.{}.{}", trigger.getNamespace(), trigger.getFlowId(), trigger.getTriggerId(), e);
+                }
+            });
+
         return flows
             .stream()
             .filter(flow -> flowToKeep.contains(flow.getId()))
diff --git a/core/src/main/java/io/kestra/core/schedulers/SchedulerTriggerStateInterface.java b/core/src/main/java/io/kestra/core/schedulers/SchedulerTriggerStateInterface.java
index 54a85e2c4..c3db9099b 100644
--- a/core/src/main/java/io/kestra/core/schedulers/SchedulerTriggerStateInterface.java
+++ b/core/src/main/java/io/kestra/core/schedulers/SchedulerTriggerStateInterface.java
@@ -6,6 +6,7 @@ import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.TriggerContext;
+import io.kestra.core.queues.QueueException;
 import jakarta.validation.ConstraintViolationException;
 
 import java.time.ZonedDateTime;
@@ -25,7 +26,10 @@ public interface SchedulerTriggerStateInterface {
 
     Trigger update(Flow flow, AbstractTrigger abstractTrigger, ConditionContext conditionContext) throws Exception;
 
-
+    /**
+     * QueueException required for Kafka implementation
+     */
+    void delete(Trigger trigger) throws QueueException;
     /**
      * Used by the JDBC implementation: find triggers in all tenants.
      */
diff --git a/jdbc-h2/src/main/java/io/kestra/runner/h2/H2JdbcCleanerService.java b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2JdbcCleanerService.java
new file mode 100644
index 000000000..5d3633576
--- /dev/null
+++ b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2JdbcCleanerService.java
@@ -0,0 +1,15 @@
+package io.kestra.runner.h2;
+
+import io.kestra.jdbc.repository.AbstractJdbcRepository;
+import io.kestra.jdbc.runner.JdbcCleanerService;
+import jakarta.inject.Singleton;
+import org.jooq.Condition;
+
+@Singleton
+@H2QueueEnabled
+public class H2JdbcCleanerService implements JdbcCleanerService {
+    @Override
+    public Condition buildTypeCondition(String type) {
+        return AbstractJdbcRepository.field("type").eq(type);
+    }
+}
diff --git a/jdbc-h2/src/main/java/io/kestra/runner/h2/H2Queue.java b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2Queue.java
index e0a271fe5..0a71bfb60 100644
--- a/jdbc-h2/src/main/java/io/kestra/runner/h2/H2Queue.java
+++ b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2Queue.java
@@ -8,6 +8,7 @@ import org.jooq.Record;
 import org.jooq.impl.DSL;
 import org.jooq.impl.SQLDataType;
 
+import java.time.LocalDateTime;
 import java.util.List;
 
 public class H2Queue<T> extends JdbcQueue<T> {
@@ -53,7 +54,6 @@ public class H2Queue<T> extends JdbcQueue<T> {
             .getFirst();
     }
 
-    @SuppressWarnings("RedundantCast")
     @Override
     protected void updateGroupOffsets(DSLContext ctx, String consumerGroup, String queueType, List<Integer> offsets) {
         var update = ctx.update(DSL.table(table.getName()))
@@ -65,7 +65,8 @@ public class H2Queue<T> extends JdbcQueue<T> {
                     (Object) new String[]{queueType}
                 )
             )
-            .where(AbstractJdbcRepository.field("offset").in((Object[]) offsets.toArray(Integer[]::new)));
+            .set(AbstractJdbcRepository.field("updated"), LocalDateTime.now())
+            .where(AbstractJdbcRepository.field("offset").in(offsets.toArray(Integer[]::new)));
 
         if (consumerGroup != null) {
             update = update.and(AbstractJdbcRepository.field("consumer_group").eq(consumerGroup));
diff --git a/jdbc-h2/src/main/resources/migrations/h2/V1_31__queues_updated_date.sql b/jdbc-h2/src/main/resources/migrations/h2/V1_31__queues_updated_date.sql
new file mode 100644
index 000000000..a6d19dcb1
--- /dev/null
+++ b/jdbc-h2/src/main/resources/migrations/h2/V1_31__queues_updated_date.sql
@@ -0,0 +1,2 @@
+ALTER TABLE queues ALTER COLUMN "updated" DROP NOT NULL;
+ALTER TABLE queues ALTER COLUMN "updated" DROP DEFAULT;
\ No newline at end of file
diff --git a/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlJdbcCleanerService.java b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlJdbcCleanerService.java
new file mode 100644
index 000000000..5f41e1bbd
--- /dev/null
+++ b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlJdbcCleanerService.java
@@ -0,0 +1,15 @@
+package io.kestra.runner.mysql;
+
+import io.kestra.jdbc.repository.AbstractJdbcRepository;
+import io.kestra.jdbc.runner.JdbcCleanerService;
+import jakarta.inject.Singleton;
+import org.jooq.Condition;
+
+@Singleton
+@MysqlQueueEnabled
+public class MysqlJdbcCleanerService implements JdbcCleanerService {
+    @Override
+    public Condition buildTypeCondition(String type) {
+        return AbstractJdbcRepository.field("type").eq(type);
+    }
+}
diff --git a/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlQueue.java b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlQueue.java
index 8dabb6359..530ca638e 100644
--- a/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlQueue.java
+++ b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlQueue.java
@@ -7,6 +7,7 @@ import org.jooq.*;
 import org.jooq.Record;
 import org.jooq.impl.DSL;
 
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -62,7 +63,6 @@ public class MysqlQueue<T> extends JdbcQueue<T> {
             .getFirst();
     }
 
-    @SuppressWarnings("RedundantCast")
     @Override
     protected void updateGroupOffsets(DSLContext ctx, String consumerGroup, String queueType, List<Integer> offsets) {
         var update = ctx
@@ -71,7 +71,8 @@ public class MysqlQueue<T> extends JdbcQueue<T> {
                 AbstractJdbcRepository.field("consumers"),
                 DSL.field("CONCAT_WS(',', consumers, ?)", String.class, queueType)
             )
-            .where(AbstractJdbcRepository.field("offset").in((Object[]) offsets.toArray(Integer[]::new)));
+            .set(AbstractJdbcRepository.field("updated"), LocalDateTime.now())
+            .where(AbstractJdbcRepository.field("offset").in(offsets.toArray(Integer[]::new)));
 
         if (consumerGroup != null) {
             update = update.and(AbstractJdbcRepository.field("consumer_group").eq(consumerGroup));
diff --git a/jdbc-mysql/src/main/resources/migrations/mysql/V1_31__queues_updated_date.sql b/jdbc-mysql/src/main/resources/migrations/mysql/V1_31__queues_updated_date.sql
new file mode 100644
index 000000000..f67e1fbf1
--- /dev/null
+++ b/jdbc-mysql/src/main/resources/migrations/mysql/V1_31__queues_updated_date.sql
@@ -0,0 +1,1 @@
+ALTER TABLE queues MODIFY COLUMN `updated` TIMESTAMP NULL DEFAULT NULL;
\ No newline at end of file
diff --git a/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresJdbcCleanerService.java b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresJdbcCleanerService.java
new file mode 100644
index 000000000..6918a7ba1
--- /dev/null
+++ b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresJdbcCleanerService.java
@@ -0,0 +1,15 @@
+package io.kestra.runner.postgres;
+
+import io.kestra.jdbc.runner.JdbcCleanerService;
+import jakarta.inject.Singleton;
+import org.jooq.Condition;
+import org.jooq.impl.DSL;
+
+@Singleton
+@PostgresQueueEnabled
+public class PostgresJdbcCleanerService implements JdbcCleanerService {
+    @Override
+    public Condition buildTypeCondition(String type) {
+        return DSL.condition("type = CAST(? AS queue_type)", type);
+    }
+}
diff --git a/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresQueue.java b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresQueue.java
index f592ddca2..bc2435d21 100644
--- a/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresQueue.java
+++ b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresQueue.java
@@ -12,6 +12,8 @@ import org.jooq.*;
 import org.jooq.Record;
 import org.jooq.impl.DSL;
 
+import java.time.LocalDateTime;
+import java.time.ZonedDateTime;
 import java.util.List;
 import java.util.Map;
 
@@ -80,15 +82,12 @@ public class PostgresQueue<T> extends JdbcQueue<T> {
             .getFirst();
     }
 
-    @SuppressWarnings("RedundantCast")
     @Override
     protected void updateGroupOffsets(DSLContext ctx, String consumerGroup, String queueType, List<Integer> offsets) {
         var update = ctx.update(DSL.table(table.getName()))
-            .set(
-                AbstractJdbcRepository.field("consumer_" + queueType),
-                true
-            )
-            .where(AbstractJdbcRepository.field("offset").in((Object[]) offsets.toArray(Integer[]::new)));
+            .set(AbstractJdbcRepository.field("consumer_" + queueType), true)
+            .set(AbstractJdbcRepository.field("updated"), LocalDateTime.now())
+            .where(AbstractJdbcRepository.field("offset").in(offsets.toArray(Integer[]::new)));
 
         if (consumerGroup != null) {
             update = update.and(AbstractJdbcRepository.field("consumer_group").eq(consumerGroup));
diff --git a/jdbc-postgres/src/main/resources/migrations/postgres/V1_31__queues_updated_date.sql b/jdbc-postgres/src/main/resources/migrations/postgres/V1_31__queues_updated_date.sql
new file mode 100644
index 000000000..d1488c3b9
--- /dev/null
+++ b/jdbc-postgres/src/main/resources/migrations/postgres/V1_31__queues_updated_date.sql
@@ -0,0 +1,2 @@
+ALTER TABLE queues ALTER COLUMN updated DROP NOT NULL;
+ALTER TABLE queues ALTER COLUMN updated DROP DEFAULT;
\ No newline at end of file
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcCleaner.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcCleaner.java
index 046f58b0a..bd6cd3304 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcCleaner.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcCleaner.java
@@ -1,9 +1,11 @@
 package io.kestra.jdbc.runner;
 
+import io.kestra.core.utils.ListUtils;
 import io.kestra.jdbc.JdbcTableConfig;
 import io.kestra.jdbc.JooqDSLContextWrapper;
 import io.kestra.jdbc.repository.AbstractJdbcRepository;
 import io.micronaut.context.annotation.ConfigurationProperties;
+import io.micronaut.context.annotation.EachProperty;
 import io.micronaut.context.annotation.Requires;
 import io.micronaut.scheduling.annotation.Scheduled;
 import jakarta.inject.Inject;
@@ -17,6 +19,7 @@ import org.jooq.impl.DSL;
 
 import java.time.Duration;
 import java.time.ZonedDateTime;
+import java.util.List;
 
 @Singleton
 @JdbcRunnerEnabled
@@ -25,20 +28,37 @@ import java.time.ZonedDateTime;
 public class JdbcCleaner {
     private final JooqDSLContextWrapper dslContextWrapper;
     private final Configuration configuration;
-    protected final Table<Record> queueTable;
+    private final JdbcCleanerService jdbcCleanerService;
+    private final Table<Record> queueTable;
 
     @Inject
     public JdbcCleaner(@Named("queues") JdbcTableConfig jdbcTableConfig,
                        JooqDSLContextWrapper dslContextWrapper,
-                       Configuration configuration
+                       Configuration configuration,
+                       JdbcCleanerService jdbcCleanerService
     ) {
         this.dslContextWrapper = dslContextWrapper;
         this.configuration = configuration;
+        this.jdbcCleanerService = jdbcCleanerService;
 
         this.queueTable = DSL.table(jdbcTableConfig.table());
     }
 
     public void deleteQueue() {
+        // first, delete types that are configured more specifically
+        ListUtils.emptyOnNull(configuration.getTypes()).forEach(type -> {
+            dslContextWrapper.transaction(configuration -> {
+                int deleted = DSL
+                    .using(configuration)
+                    .delete(this.queueTable)
+                    .where(AbstractJdbcRepository.field("updated").lessOrEqual(ZonedDateTime.now().minus(type.getRetention()).toOffsetDateTime()))
+                    .and(jdbcCleanerService.buildTypeCondition(type.getType()))
+                    .execute();
+                log.info("Cleaned {} records from {} for type {}", deleted, this.queueTable.getName(), type.getType());
+            });
+        });
+
+        // then, delete all other records
         dslContextWrapper.transaction(configuration -> {
             int deleted = DSL
                 .using(configuration)
@@ -61,5 +81,13 @@ public class JdbcCleaner {
     @Getter
     public static class Configuration {
         Duration retention;
+        List<TypeConfiguration> types;
+
+        @Getter
+        @EachProperty(value = "types", list = true)
+        public static class TypeConfiguration {
+            String type;
+            Duration retention;
+        }
     }
 }
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcCleanerService.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcCleanerService.java
new file mode 100644
index 000000000..4b10ec8a8
--- /dev/null
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcCleanerService.java
@@ -0,0 +1,13 @@
+package io.kestra.jdbc.runner;
+
+import org.jooq.Condition;
+
+/**
+ * This service is used solely by the {@link JdbcCleaner} to handle database-specific queries.
+ */
+public interface JdbcCleanerService {
+    /**
+     * Build the condition for the <code>types</code> column of the <code>queues</code> table.
+     */
+    Condition buildTypeCondition(String type);
+}
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
index e9f03d8de..20934df53 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -933,6 +933,13 @@ public class JdbcExecutor implements ExecutorInterface, Service {
             }
 
             if (!shouldSend) {
+                // delete the execution from the state storage if ended
+                // IMPORTANT: it must be done here as it's when the execution arrives 'again' with a terminated state,
+                // so we are sure at this point that no new executions will be created otherwise the tate storage would be re-created by the execution queue.
+                if (executorService.canBePurged(executor)) {
+                    executorStateStorage.delete(executor.getExecution());
+                }
+
                 return;
             }
 
@@ -957,11 +964,6 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                 ((JdbcQueue<Execution>) this.executionQueue).emitOnly(null, executor.getExecution());
             }
 
-            // delete the execution from the state storage if ended
-            if (executorService.canBePurged(executor)) {
-                executorStateStorage.delete(executor.getExecution());
-            }
-
             Execution execution = executor.getExecution();
             // handle flow triggers on state change
             if (!execution.getState().getCurrent().equals(executor.getOriginalState())) {
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java
index 3193bf186..d4cb5f8be 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java
@@ -185,7 +185,7 @@ public abstract class JdbcQueue<T> implements QueueInterface<T> {
      * This is used to purge a queue for specific keys.
      */
     public void deleteByKeys(List<String> keys) throws QueueException {
-        // process in batches of 100 items to avoid too big IN clause
+        // process in batches of 100 items to avoid too big IN clausecQueue
         Iterables.partition(keys, 100).forEach(batch -> {
             dslContextWrapper.transaction(configuration -> {
                 int deleted = DSL
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
index 1b4c4024e..086aca64b 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
@@ -6,6 +6,7 @@ import io.kestra.core.models.flows.FlowWithSource;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.TriggerContext;
+import io.kestra.core.queues.QueueException;
 import io.kestra.core.schedulers.ScheduleContextInterface;
 import io.kestra.core.schedulers.SchedulerTriggerStateInterface;
 import io.kestra.jdbc.repository.AbstractJdbcTriggerRepository;
@@ -77,6 +78,10 @@ public class JdbcSchedulerTriggerState implements SchedulerTriggerStateInterface
         return this.triggerRepository.update(flow, abstractTrigger, conditionContext);
     }
 
+    public void delete(Trigger trigger) throws QueueException {
+        this.triggerRepository.delete(trigger);
+    }
+
     @Override
     public List<Trigger> findByNextExecutionDateReadyForAllTenants(ZonedDateTime now, ScheduleContextInterface scheduleContext) {
         return this.triggerRepository.findByNextExecutionDateReadyForAllTenants(now, scheduleContext);
diff --git a/ui/src/components/filter/segments/Dashboards.vue b/ui/src/components/filter/segments/Dashboards.vue
index ad72446b0..afd847960 100644
--- a/ui/src/components/filter/segments/Dashboards.vue
+++ b/ui/src/components/filter/segments/Dashboards.vue
@@ -71,7 +71,7 @@
 </template>
 
 <script setup lang="ts">
-    import {onBeforeMount, ref, computed, getCurrentInstance} from "vue";
+    import {onBeforeMount, ref, computed, getCurrentInstance, watch} from "vue";
     import KestraIcon from "../../Kicon.vue";
     import {Menu, Plus, DeleteOutline, Magnify, Pencil} from "../utils/icons";
     import {useI18n} from "vue-i18n";
@@ -81,6 +81,7 @@
     const {t} = useI18n({useScope: "global"});
     const store = useStore();
     const route = useRoute();
+    const routeTenant = ref(route.params.tenant);
     const router = useRouter();
     const emits = defineEmits(["dashboard"]);
     const toast = getCurrentInstance().appContext.config.globalProperties.$toast();
@@ -116,7 +117,7 @@
         router.push({name: "dashboards/update", params: {id: dashboard.id}});
     }
 
-    onBeforeMount(() => {
+    const fetchDashboards = () => {
         store
             .dispatch("dashboard/list", {})
             .then((response: { results: { id: string; title: string }[] }) => {
@@ -125,28 +126,43 @@
                     const dashboard = dashboards.value.find(d => d.id === route.params.id);
                     if (dashboard) {
                         selectedDashboard.value = dashboard.title;
+                    } else {
+                        selectedDashboard.value = null;
                     }
                 }
             });
+    }
+
+    onBeforeMount(() => {
+        fetchDashboards()
     });
+
+    watch(
+        route,
+        (newRoute) => {
+            if (routeTenant.value !== newRoute.params.tenant) {
+                fetchDashboards();
+                routeTenant.value = newRoute.params.tenant;
+            }
+        }, {deep: true});
 </script>
 
 <style scoped lang="scss">
-@import "../styles/filter";
+    @import "../styles/filter";
 
-.dropdown {
-    width: 300px;
-}
+    .dropdown {
+        width: 300px;
+    }
 
-.items {
-    max-height: 160px !important; // 5 visible items
-}
+    .items {
+        max-height: 160px !important; // 5 visible items
+    }
 
-.main-button {
-    max-width: 300px;
+    .main-button {
+        max-width: 300px;
 
-    span {
-        max-width: 250px;
+        span {
+            max-width: 250px;
+        }
     }
-}
 </style>
\ No newline at end of file


Test patch:
diff --git a/core/src/test/java/io/kestra/core/models/property/PropertyTest.java b/core/src/test/java/io/kestra/core/models/property/PropertyTest.java
index 8cb51ab6b..72e9fc16a 100644
--- a/core/src/test/java/io/kestra/core/models/property/PropertyTest.java
+++ b/core/src/test/java/io/kestra/core/models/property/PropertyTest.java
@@ -298,6 +298,30 @@ class PropertyTest {
         assertThat(output.getMap().get("mapKey2"), is("mapValue2"));
     }
 
+    @Test
+    void aListToRender() throws Exception {
+        var task = DynamicPropertyExampleTask.builder()
+            .items(new Property<>("""
+                ["python test.py --input1 \\"{{ item1 }}\\" --input2 \\"{{ item2 }}\\""]"""))
+            .properties(new Property<>("""
+                {
+                  "key1": "{{value1}}",
+                  "key2": "{{value2}}"
+                }"""))
+            .build();
+        var runContext = runContextFactory.of(Map.ofEntries(
+            entry("item1", "item1"),
+            entry("item2", "item2"),
+            entry("value1", "value1"),
+            entry("value2", "value2")
+        ));
+
+        var output = task.run(runContext);
+
+        assertThat(output, notNullValue());
+        assertThat(output.getList(), containsInAnyOrder("python test.py --input1 \"item1\" --input2 \"item2\""));
+    }
+
     @Builder
     @Getter
     private static class TestObj {
diff --git a/core/src/test/java/io/kestra/plugin/core/http/RequestTest.java b/core/src/test/java/io/kestra/plugin/core/http/RequestTest.java
index 718aa48e1..b84566376 100644
--- a/core/src/test/java/io/kestra/plugin/core/http/RequestTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/http/RequestTest.java
@@ -209,6 +209,37 @@ class RequestTest {
         }
     }
 
+    @Test
+    void failedPost() throws Exception {
+        try (
+            ApplicationContext applicationContext = ApplicationContext.run();
+            EmbeddedServer server = applicationContext.getBean(EmbeddedServer.class).start();
+
+        ) {
+            Request task = Request.builder()
+                .id(RequestTest.class.getSimpleName())
+                .type(RequestTest.class.getName())
+                .uri(Property.of(server.getURL().toString() + "/markdown"))
+                .method(Property.of("POST"))
+                .body(Property.of("# hello web!"))
+                .contentType(Property.of("text/markdown"))
+                .options(HttpConfiguration.builder().defaultCharset(Property.of(null)).build())
+                .build();
+
+            RunContext runContext = TestsUtils.mockRunContext(this.runContextFactory, task, ImmutableMap.of());
+
+            HttpClientResponseException exception = assertThrows(
+                HttpClientResponseException.class,
+                () -> task.run(runContext)
+            );
+
+            assertThat(exception.getResponse().getStatus().getCode(), is(417));
+            assertThat(exception.getMessage(), containsString("hello world"));
+            byte[] content = ((io.kestra.core.http.HttpRequest.ByteArrayRequestBody) exception.getRequest().getBody()).getContent();
+            assertThat(new String(content) , containsString("hello web"));
+        }
+    }
+
     @Test
     void selfSigned() throws Exception {
         try (
@@ -530,6 +561,13 @@ class RequestTest {
             return HttpResponse.status(HttpStatus.EXPECTATION_FAILED).body("{ \"hello\": \"world\" }");
         }
 
+        @Post("/markdown")
+        @Consumes(MediaType.TEXT_MARKDOWN)
+        @Produces(MediaType.TEXT_MARKDOWN)
+        HttpResponse<String> postMarkdown() {
+            return HttpResponse.status(HttpStatus.EXPECTATION_FAILED).body("# hello world");
+        }
+
         @Get("/redirect")
         HttpResponse<String> redirect() {
             return HttpResponse.redirect(URI.create("/hello"));
diff --git a/core/src/test/resources/flows/valids/sleep-long.yml b/core/src/test/resources/flows/valids/sleep-long.yml
new file mode 100644
index 000000000..c0ed77fb2
--- /dev/null
+++ b/core/src/test/resources/flows/valids/sleep-long.yml
@@ -0,0 +1,7 @@
+id: sleep-long
+namespace: io.kestra.tests
+
+tasks:
+  - id: sleep-long
+    type: io.kestra.plugin.core.flow.Sleep
+    duration: PT300S
\ No newline at end of file
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
index 303e50112..015aa0c59 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
@@ -1276,7 +1276,7 @@ class ExecutionControllerTest {
     @RetryingTest(5)
     void kill() throws TimeoutException, InterruptedException, QueueException {
         // Run execution until it is paused
-        Execution runningExecution = runnerUtils.runOneUntilRunning(null, TESTS_FLOW_NS, "sleep");
+        Execution runningExecution = runnerUtils.runOneUntilRunning(null, TESTS_FLOW_NS, "sleep-long");
         assertThat(runningExecution.getState().isRunning(), is(true));
 
         // listen to the execution queue