Repository: kestra-io/kestra (java)

Issue:
Add labels map to the new Executions condition

```yaml
  - id: flow3
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - id: flow_filters
        type: io.kestra.plugin.core.condition.ExecutionsCondition
        timeSLA:
          type: DURATION_WINDOW
          window: PT1D
        preconditions:
          - namespace: company.team
            flowId: myflow1
            states: [SUCCESS, WARNING]
            labels: 
              mykey: mylabelvalue
          - namespace: company.team
            flowId: myflow2
            states: [SUCCESS]
```


Pull request description:
feat(core): allow filtering on labels for ExecutionsCondition
Fixes #5836

Patch:
diff --git a/core/src/main/java/io/kestra/plugin/core/condition/ExecutionsCondition.java b/core/src/main/java/io/kestra/plugin/core/condition/ExecutionsCondition.java
index 0135ec03f..ca94f80ad 100644
--- a/core/src/main/java/io/kestra/plugin/core/condition/ExecutionsCondition.java
+++ b/core/src/main/java/io/kestra/plugin/core/condition/ExecutionsCondition.java
@@ -2,13 +2,13 @@ package io.kestra.plugin.core.condition;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import io.kestra.core.exceptions.InternalException;
+import io.kestra.core.models.Label;
 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.conditions.Condition;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.flows.State;
-import io.kestra.core.models.property.Property;
 import io.kestra.core.utils.ListUtils;
 import io.swagger.v3.oas.annotations.Hidden;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -112,13 +112,20 @@ public class ExecutionsCondition extends AbstractMultipleCondition {
     public static class UpstreamFlow {
         @NotNull
         @Schema(title = "The namespace of the flow.")
-        private Property<String> namespace;
+        @PluginProperty
+        private String namespace;
 
         @Schema(title = "The flow id.")
-        private Property<String> flowId;
+        @PluginProperty
+        private String flowId;
 
         @Schema(title = "The execution states.")
-        private Property<List<State.Type>> states;
+        @PluginProperty
+        private List<State.Type> states;
+
+        @Schema(title = "A key/value map of labels.")
+        @PluginProperty
+        private Map<String, Object> labels;
     }
 
     @Hidden
@@ -131,22 +138,28 @@ public class ExecutionsCondition extends AbstractMultipleCondition {
 
         @Override
         public boolean test(ConditionContext conditionContext) throws InternalException {
-            String namespace = conditionContext.getRunContext().render(upstreamFlow.namespace).as(String.class).orElse(null);
-            if (!conditionContext.getExecution().getNamespace().equals(namespace)) {
+            if (upstreamFlow.namespace != null && !conditionContext.getExecution().getNamespace().equals(upstreamFlow.namespace)) {
                 return false;
             }
 
-            if (upstreamFlow.flowId != null) {
-                String flowId = conditionContext.getRunContext().render(upstreamFlow.flowId).as(String.class).orElse(null);
-                if (!conditionContext.getExecution().getFlowId().equals(flowId)) {
-                    return false;
-                }
+            if (upstreamFlow.flowId != null && !conditionContext.getExecution().getFlowId().equals(upstreamFlow.flowId)) {
+                return false;
             }
 
             // we need to only evaluate on namespace and flow for simulated executions
-            if (upstreamFlow.states != null && !ListUtils.emptyOnNull(conditionContext.getExecution().getLabels()).contains(SIMULATED_EXECUTION)) {
-                List<State.Type> states = conditionContext.getRunContext().render(upstreamFlow.states).asList(State.Type.class);
-                return states.contains(conditionContext.getExecution().getState().getCurrent());
+            if (ListUtils.emptyOnNull(conditionContext.getExecution().getLabels()).contains(SIMULATED_EXECUTION)) {
+                return true;
+            }
+
+            if (upstreamFlow.states != null && !upstreamFlow.states.contains(conditionContext.getExecution().getState().getCurrent())) {
+                return false;
+            }
+
+            if (upstreamFlow.labels != null) {
+                boolean notMatched = upstreamFlow.labels.entrySet().stream()
+                    .map(entry -> new Label(entry.getKey(), String.valueOf(entry.getValue())))
+                    .anyMatch(label -> !conditionContext.getExecution().getLabels().contains(label));
+                return !notMatched;
             }
 
             return true;
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 24397a453..e0a271fe5 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
@@ -16,36 +16,8 @@ public class H2Queue<T> extends JdbcQueue<T> {
     }
 
     @Override
-    protected Result<Record> receiveFetch(DSLContext ctx, String consumerGroup, Integer offset, boolean forUpdate) {
-        var select = ctx.select(
-                AbstractJdbcRepository.field("value"),
-                AbstractJdbcRepository.field("offset")
-            )
-            .from(this.table)
-            .where(AbstractJdbcRepository.field("type").eq(this.cls.getName()));
-
-        if (consumerGroup != null) {
-            select = select.and(AbstractJdbcRepository.field("consumer_group").eq(consumerGroup));
-        } else {
-            select = select.and(AbstractJdbcRepository.field("consumer_group").isNull());
-        }
-
-        if (offset != 0) {
-            select = select.and(AbstractJdbcRepository.field("offset").gt(offset));
-        }
-
-        var limitSelect = select
-            .orderBy(AbstractJdbcRepository.field("offset").asc())
-            .limit(configuration.getPollSize());
-        ResultQuery<Record2<Object, Object>> configuredSelect = limitSelect;
-
-        if (forUpdate) {
-            configuredSelect = limitSelect.forUpdate().skipLocked();
-        }
-
-        return configuredSelect
-            .fetchMany()
-            .getFirst();
+    protected Condition buildTypeCondition(String type) {
+        return AbstractJdbcRepository.field("type").eq(type);
     }
 
     @Override
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 8926a27dd..8dabb6359 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
@@ -23,36 +23,8 @@ public class MysqlQueue<T> extends JdbcQueue<T> {
     }
 
     @Override
-    protected Result<Record> receiveFetch(DSLContext ctx, String consumerGroup, Integer offset, boolean forUpdate) {
-        var select = ctx.select(
-                AbstractJdbcRepository.field("value"),
-                AbstractJdbcRepository.field("offset")
-            )
-            .from(this.table)
-            .where(AbstractJdbcRepository.field("type").eq(this.cls.getName()));
-
-        if (offset != 0) {
-            select = select.and(AbstractJdbcRepository.field("offset").gt(offset));
-        }
-
-        if (consumerGroup != null) {
-            select = select.and(AbstractJdbcRepository.field("consumer_group").eq(consumerGroup));
-        } else {
-            select = select.and(AbstractJdbcRepository.field("consumer_group").isNull());
-        }
-
-        var limitSelect = select
-            .orderBy(AbstractJdbcRepository.field("offset").asc())
-            .limit(configuration.getPollSize());
-        ResultQuery<Record2<Object, Object>> configuredSelect = limitSelect;
-
-        if (forUpdate) {
-            configuredSelect = limitSelect.forUpdate().skipLocked();
-        }
-
-        return configuredSelect
-            .fetchMany()
-            .getFirst();
+    protected Condition buildTypeCondition(String type) {
+        return AbstractJdbcRepository.field("type").eq(type);
     }
 
     @Override
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 15ee0daf5..f592ddca2 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
@@ -42,36 +42,8 @@ public class PostgresQueue<T> extends JdbcQueue<T> {
     }
 
     @Override
-    protected Result<Record> receiveFetch(DSLContext ctx, String consumerGroup, @NonNull Integer offset, boolean forUpdate) {
-        var select = ctx.select(
-                AbstractJdbcRepository.field("value"),
-                AbstractJdbcRepository.field("offset")
-            )
-            .from(this.table)
-            .where(DSL.condition("type = CAST(? AS queue_type)", this.cls.getName()));
-
-        if (offset != 0) {
-            select = select.and(AbstractJdbcRepository.field("offset").gt(offset));
-        }
-
-        if (consumerGroup != null) {
-            select = select.and(AbstractJdbcRepository.field("consumer_group").eq(consumerGroup));
-        } else {
-            select = select.and(AbstractJdbcRepository.field("consumer_group").isNull());
-        }
-
-        var limitSelect = select
-            .orderBy(AbstractJdbcRepository.field("offset").asc())
-            .limit(configuration.getPollSize());
-        ResultQuery<Record2<Object, Object>> configuredSelect = limitSelect;
-
-        if (forUpdate) {
-            configuredSelect = limitSelect.forUpdate().skipLocked();
-        }
-
-        return configuredSelect
-            .fetchMany()
-            .getFirst();
+    protected Condition buildTypeCondition(String type) {
+        return DSL.condition("type = CAST(? AS queue_type)", type);
     }
 
     @Override
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 f79c66f7a..00fe19816 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java
@@ -22,12 +22,8 @@ import io.micronaut.context.annotation.ConfigurationProperties;
 import io.micronaut.transaction.exceptions.CannotCreateTransactionException;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
-import org.jooq.DSLContext;
-import org.jooq.Field;
-import org.jooq.JSONB;
+import org.jooq.*;
 import org.jooq.Record;
-import org.jooq.Result;
-import org.jooq.Table;
 import org.jooq.impl.DSL;
 
 import java.io.IOException;
@@ -173,7 +169,37 @@ public abstract class JdbcQueue<T> implements QueueInterface<T> {
         return this.receiveFetch(ctx, consumerGroup, offset, true);
     }
 
-    abstract protected Result<Record> receiveFetch(DSLContext ctx, String consumerGroup, Integer offset, boolean forUpdate);
+    protected Result<Record> receiveFetch(DSLContext ctx, String consumerGroup, Integer offset, boolean forUpdate) {
+        var select = ctx.select(
+                AbstractJdbcRepository.field("value"),
+                AbstractJdbcRepository.field("offset")
+            )
+            .from(this.table)
+            .where(buildTypeCondition(this.cls.getName()));
+
+        if (offset != 0) {
+            select = select.and(AbstractJdbcRepository.field("offset").gt(offset));
+        }
+
+        if (consumerGroup != null) {
+            select = select.and(AbstractJdbcRepository.field("consumer_group").eq(consumerGroup));
+        } else {
+            select = select.and(AbstractJdbcRepository.field("consumer_group").isNull());
+        }
+
+        var limitSelect = select
+            .orderBy(AbstractJdbcRepository.field("offset").asc())
+            .limit(configuration.getPollSize());
+        ResultQuery<Record2<Object, Object>> configuredSelect = limitSelect;
+
+        if (forUpdate) {
+            configuredSelect = limitSelect.forUpdate().skipLocked();
+        }
+
+        return configuredSelect
+            .fetchMany()
+            .getFirst();
+    }
 
     protected Result<Record> receiveFetch(DSLContext ctx, String consumerGroup, String queueType) {
         return this.receiveFetch(ctx, consumerGroup, queueType, true);
@@ -183,18 +209,26 @@ public abstract class JdbcQueue<T> implements QueueInterface<T> {
 
     abstract protected void updateGroupOffsets(DSLContext ctx, String consumerGroup, String queueType, List<Integer> offsets);
 
+    protected abstract Condition buildTypeCondition(String type);
+
     @Override
     public Runnable receive(String consumerGroup, Consumer<Either<T, DeserializationException>> consumer, boolean forUpdate) {
         AtomicInteger maxOffset = new AtomicInteger();
 
         // fetch max offset
         dslContextWrapper.transaction(configuration -> {
-            Integer integer = DSL
+            var select = DSL
                 .using(configuration)
                 .select(DSL.max(AbstractJdbcRepository.field("offset")).as("max"))
                 .from(table)
-                .fetchAny("max", Integer.class);
+                .where(buildTypeCondition(this.cls.getName()));
+            if (consumerGroup != null) {
+                select = select.and(AbstractJdbcRepository.field("consumer_group").eq(consumerGroup));
+            } else {
+                select = select.and(AbstractJdbcRepository.field("consumer_group").isNull());
+            }
 
+            Integer integer = select.fetchAny("max", Integer.class);
             if (integer != null) {
                 maxOffset.set(integer);
             }


Test patch:
diff --git a/core/src/test/resources/flows/valids/flow-trigger-basic-flow-a.yaml b/core/src/test/resources/flows/valids/flow-trigger-basic-flow-a.yaml
index 8a501c6c0..e29cb6e11 100644
--- a/core/src/test/resources/flows/valids/flow-trigger-basic-flow-a.yaml
+++ b/core/src/test/resources/flows/valids/flow-trigger-basic-flow-a.yaml
@@ -1,6 +1,9 @@
 id: flow-trigger-basic-flow-a
 namespace: io.kestra.tests.trigger.basic
 
+labels:
+  some: label
+
 tasks:
   - id: only
     type: io.kestra.plugin.core.debug.Return
diff --git a/core/src/test/resources/flows/valids/flow-trigger-basic-flow-listen.yaml b/core/src/test/resources/flows/valids/flow-trigger-basic-flow-listen.yaml
index 3a334c669..f66594ba7 100644
--- a/core/src/test/resources/flows/valids/flow-trigger-basic-flow-listen.yaml
+++ b/core/src/test/resources/flows/valids/flow-trigger-basic-flow-listen.yaml
@@ -12,6 +12,8 @@ triggers:
           - namespace: io.kestra.tests.trigger.basic
             flowId: flow-trigger-basic-flow-a
             states: [SUCCESS, WARNING]
+            labels:
+              some: label
           - namespace: io.kestra.tests.trigger.basic
             flowId: flow-trigger-basic-flow-b
             states: [SUCCESS]