refactor(core): use interface HasUID when possible

This commit adds a new HasUID interface for explicitly
managing classes that expose a unique identifier

core/src/main/java/io/kestra/core/models/HasUID.java

@@ -0,0 +1,21 @@
+package io.kestra.core.models;
+
+/**
+ * A UID is used to uniquely identify an entity across an entire Kestra's cluster.
+ * <p>
+ * A UID is either a unique random ID or composed of a combination of entity properties such as
+ * the associated tenant, namespace, and identifier.
+ * <p>
+ * For Kestra's queuing mechanism the UID can be used as routing/or partitioning key.
+ */
+public interface HasUID {
+
+    /**
+     * Gets the UID attached to this entity.
+     * <p>
+     * Be careful when modifying the implementation of this method for subclasses, as it should be consistent over time.
+     *
+     * @return the string uid.
+     */
+    String uid();
+}

core/src/main/java/io/kestra/core/models/executions/ExecutionKilled.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.kestra.core.models.HasUID;
 import io.kestra.core.models.TenantInterface;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -35,11 +36,9 @@
     @JsonSubTypes.Type(value = ExecutionKilledExecution.class, name = "execution"),
     @JsonSubTypes.Type(value = ExecutionKilledTrigger.class, name = "trigger"),
 })
-abstract public class ExecutionKilled implements TenantInterface {
+abstract public class ExecutionKilled implements TenantInterface, HasUID {
     abstract public String getType();
 
-    abstract public String uid();
-
     public enum State {
         REQUESTED,
         EXECUTED

core/src/main/java/io/kestra/core/models/executions/ExecutionKilledTrigger.java

@@ -36,7 +36,6 @@ public boolean isEqual(TriggerContext triggerContext) {
             triggerContext.getTriggerId().equals(this.triggerId);
     }
 
-
     @Override
     public String uid() {
         return IdUtils.fromParts(

core/src/main/java/io/kestra/core/models/flows/Flow.java

@@ -9,6 +9,7 @@
 import com.fasterxml.jackson.databind.introspect.AnnotatedMember;
 import com.fasterxml.jackson.databind.introspect.JacksonAnnotationIntrospector;
 import io.kestra.core.exceptions.InternalException;
+import io.kestra.core.models.HasUID;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.executions.Execution;
@@ -49,7 +50,7 @@
 @ToString
 @EqualsAndHashCode
 @FlowValidation
-public class Flow extends AbstractFlow {
+public class Flow extends AbstractFlow implements HasUID {
     private static final ObjectMapper NON_DEFAULT_OBJECT_MAPPER = JacksonMapper.ofYaml()
         .copy()
         .setSerializationInclusion(JsonInclude.Include.NON_DEFAULT);
@@ -121,6 +122,9 @@ public Logger logger() {
         return LoggerFactory.getLogger("flow." + this.id);
     }
 
+
+    /** {@inheritDoc **/
+    @Override
     @JsonIgnore
     public String uid() {
         return Flow.uid(this.getTenantId(), this.getNamespace(), this.getId(), Optional.ofNullable(this.revision));

core/src/main/java/io/kestra/core/models/namespaces/NamespaceInterface.java

@@ -1,10 +1,14 @@
 package io.kestra.core.models.namespaces;
 
 import io.kestra.core.models.DeletedInterface;
+import io.kestra.core.models.HasUID;
 
-public interface NamespaceInterface extends DeletedInterface {
+public interface NamespaceInterface extends DeletedInterface, HasUID {
     String getId();
 
+
+    /** {@inheritDoc **/
+    @Override
     default String uid() {
         return this.getId();
     }

core/src/main/java/io/kestra/core/models/templates/Template.java

@@ -7,6 +7,7 @@
 import com.fasterxml.jackson.databind.introspect.AnnotatedMember;
 import com.fasterxml.jackson.databind.introspect.JacksonAnnotationIntrospector;
 import io.kestra.core.models.DeletedInterface;
+import io.kestra.core.models.HasUID;
 import io.kestra.core.models.TenantInterface;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.validations.ManualConstraintViolation;
@@ -33,7 +34,7 @@
 @Introspected
 @ToString
 @EqualsAndHashCode
-public class Template implements DeletedInterface, TenantInterface {
+public class Template implements DeletedInterface, TenantInterface, HasUID {
     private static final ObjectMapper YAML_MAPPER = JacksonMapper.ofYaml().copy()
         .setAnnotationIntrospector(new JacksonAnnotationIntrospector() {
             @Override
@@ -70,6 +71,9 @@ public boolean hasIgnoreMarker(final AnnotatedMember m) {
     @Builder.Default
     private final boolean deleted = false;
 
+
+    /** {@inheritDoc **/
+    @Override
     @JsonIgnore
     public String uid() {
         return Template.uid(

core/src/main/java/io/kestra/core/models/topologies/FlowTopology.java

@@ -1,14 +1,15 @@
 package io.kestra.core.models.topologies;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import io.kestra.core.models.HasUID;
 import lombok.Builder;
 import lombok.Value;
 
 import jakarta.validation.constraints.NotNull;
 
 @Value
 @Builder
-public class FlowTopology {
+public class FlowTopology implements HasUID {
     @NotNull
     FlowNode source;
 
@@ -18,6 +19,9 @@ public class FlowTopology {
     @NotNull
     FlowNode destination;
 
+
+    /** {@inheritDoc **/
+    @Override
     @JsonIgnore
     public String uid() {
         // we use destination as prefix to enable prefixScan on FlowTopologyUpdateTransformer

core/src/main/java/io/kestra/core/models/triggers/Trigger.java

@@ -1,5 +1,6 @@
 package io.kestra.core.models.triggers;
 
+import io.kestra.core.models.HasUID;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
@@ -19,7 +20,7 @@
 @EqualsAndHashCode(callSuper = true)
 @Getter
 @NoArgsConstructor
-public class Trigger extends TriggerContext {
+public class Trigger extends TriggerContext implements HasUID {
     @Nullable
     private String executionId;
 
@@ -48,6 +49,9 @@ protected Trigger(TriggerBuilder<?, ?> b) {
         return new TriggerBuilderImpl();
     }
 
+
+    /** {@inheritDoc **/
+    @Override
     public String uid() {
         return uid(this);
     }

core/src/main/java/io/kestra/core/models/triggers/multipleflows/MultipleConditionWindow.java

@@ -1,6 +1,7 @@
 package io.kestra.core.models.triggers.multipleflows;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import io.kestra.core.models.HasUID;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.utils.IdUtils;
 import lombok.Builder;
@@ -13,7 +14,7 @@
 
 @Value
 @Builder
-public class MultipleConditionWindow {
+public class MultipleConditionWindow implements HasUID {
     String tenantId;
 
     String namespace;
@@ -28,6 +29,9 @@ public class MultipleConditionWindow {
 
     Map<String, Boolean> results;
 
+
+    /** {@inheritDoc **/
+    @Override
     @JsonIgnore
     public String uid() {
         return IdUtils.fromParts(

core/src/main/java/io/kestra/core/queues/QueueService.java

@@ -1,67 +1,34 @@
 package io.kestra.core.queues;
 
+import io.kestra.core.models.HasUID;
 import io.kestra.core.models.Setting;
 import io.kestra.core.models.executions.*;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.templates.Template;
-import io.kestra.core.models.topologies.FlowTopology;
-import io.kestra.core.models.triggers.Trigger;
-import io.kestra.core.models.triggers.multipleflows.MultipleConditionWindow;
 import io.kestra.core.runners.*;
-import io.kestra.core.server.ServiceInstance;
 import jakarta.inject.Singleton;
 
 @Singleton
 public class QueueService {
     public String key(Object object) {
-        if (object.getClass() == Execution.class) {
+        if (object instanceof HasUID hasUID) {
+            return hasUID.uid();
+        } else if (object.getClass() == Execution.class) {
             return ((Execution) object).getId();
-        } else if (object.getClass() == WorkerTask.class) {
-            return ((WorkerTask) object).getTaskRun().getId();
-        } else if (object.getClass() == WorkerTaskRunning.class) {
-            return ((WorkerTaskRunning) object).getTaskRun().getId();
         } else if (object.getClass() == WorkerInstance.class) {
             return ((WorkerInstance) object).getWorkerUuid();
-        } else if (object.getClass() == WorkerTaskResult.class) {
-            return ((WorkerTaskResult) object).getTaskRun().getId();
         } else if (object.getClass() == LogEntry.class) {
             return null;
-        } else if (object.getClass() == Flow.class) {
-            return ((Flow) object).uid();
-        } else if (object.getClass() == Template.class) {
-            return ((Template) object).uid();
-        } else if (object instanceof ExecutionKilled) {
-            return ((ExecutionKilled) object).uid();
-        } else if (object.getClass() == Trigger.class) {
-            return ((Trigger) object).uid();
-        } else if (object.getClass() == MultipleConditionWindow.class) {
-            return ((MultipleConditionWindow) object).uid();
         } else if (object.getClass() == SubflowExecution.class) {
             return ((SubflowExecution<?>) object).getExecution().getId();
         } else if (object.getClass() == SubflowExecutionResult.class) {
             return ((SubflowExecutionResult) object).getExecutionId();
-        } else if (object.getClass() == ExecutionDelay.class) {
-            return ((ExecutionDelay) object).uid();
         } else if (object.getClass() == ExecutorState.class) {
             return ((ExecutorState) object).getExecutionId();
         } else if (object.getClass() == Setting.class) {
             return ((Setting) object).getKey();
         } else if (object.getClass() == Executor.class) {
-          return ((Executor) object).getExecution().getId();
-        } else if (object.getClass() == FlowTopology.class) {
-            return ((FlowTopology) object).uid();
+            return ((Executor) object).getExecution().getId();
         } else if (object.getClass() == MetricEntry.class) {
             return null;
-        } else if (object.getClass() == WorkerTrigger.class) {
-            return ((WorkerTrigger) object).getTriggerContext().uid();
-        } else if (object.getClass() == WorkerTriggerRunning.class) {
-            return ((WorkerTriggerRunning) object).getTriggerContext().uid();
-        } else if (object.getClass() == WorkerTriggerResult.class) {
-            return ((WorkerTriggerResult) object).getTriggerContext().uid();
-        } else if (object.getClass() == ExecutionQueued.class) {
-            return ((ExecutionQueued) object).uid();
-        } else if (object.getClass() == ServiceInstance.class) {
-            return ((ServiceInstance) object).id();
         } else {
             throw new IllegalArgumentException("Unknown type '" + object.getClass().getName() + "'");
         }

core/src/main/java/io/kestra/core/repositories/ServiceInstanceRepositoryInterface.java

@@ -10,7 +10,6 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 /**
@@ -113,7 +112,7 @@ default ServiceStateTransition.Response mayTransitionServiceTo(final ServiceInst
                                                                    final String reason) {
         // This default method is not transactional and may lead to inconsistent state transition.
         synchronized (this) {
-            Optional<ServiceInstance> optional = findById(instance.id());
+            Optional<ServiceInstance> optional = findById(instance.uid());
             final ImmutablePair<ServiceInstance, ServiceInstance> beforeAndAfter;
             // UNKNOWN service
             if (optional.isEmpty()) {

core/src/main/java/io/kestra/core/runners/ExecutionDelay.java

@@ -1,6 +1,7 @@
 package io.kestra.core.runners;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import io.kestra.core.models.HasUID;
 import io.kestra.core.models.flows.State;
 import jakarta.validation.constraints.NotNull;
 import lombok.AllArgsConstructor;
@@ -12,7 +13,7 @@
 @Value
 @AllArgsConstructor
 @Builder
-public class ExecutionDelay {
+public class ExecutionDelay implements HasUID {
     @NotNull
     String taskRunId;
 
@@ -26,6 +27,7 @@ public class ExecutionDelay {
 
     @NotNull DelayType delayType;
 
+    @Override
     @JsonIgnore
     public String uid() {
         return String.join("_", executionId, taskRunId);

core/src/main/java/io/kestra/core/runners/ExecutionQueued.java

@@ -1,5 +1,6 @@
 package io.kestra.core.runners;
 
+import io.kestra.core.models.HasUID;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.utils.IdUtils;
 import lombok.AllArgsConstructor;
@@ -12,7 +13,7 @@
 @Value
 @AllArgsConstructor
 @Builder
-public class ExecutionQueued {
+public class ExecutionQueued implements HasUID {
     String tenantId;
 
     @NotNull
@@ -37,6 +38,8 @@ public static ExecutionQueued fromExecutionRunning(ExecutionRunning executionRun
         );
     }
 
+    /** {@inheritDoc **/
+    @Override
     public String uid() {
         return IdUtils.fromParts(this.tenantId, this.namespace, this.flowId, this.execution.getId());
     }

core/src/main/java/io/kestra/core/runners/WorkerJob.java

@@ -2,16 +2,14 @@
 
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.kestra.core.models.HasUID;
 
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", visible = true, include = JsonTypeInfo.As.EXISTING_PROPERTY, defaultImpl = WorkerTask.class)
 @JsonSubTypes({
     @JsonSubTypes.Type(value = WorkerTask.class, name = "task"),
     @JsonSubTypes.Type(value = WorkerTrigger.class, name = "trigger")
 })
-public abstract class WorkerJob {
+public abstract class WorkerJob implements HasUID {
     abstract public String getType();
 
-    abstract public String uid();
-
-    abstract public String taskRunId();
 }

core/src/main/java/io/kestra/core/runners/WorkerJobRunning.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.kestra.core.models.HasUID;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
@@ -16,7 +17,7 @@
 @Data
 @SuperBuilder
 @NoArgsConstructor
-public abstract class WorkerJobRunning {
+public abstract class WorkerJobRunning implements HasUID {
     @NotNull
     private WorkerInstance workerInstance;
 
@@ -25,5 +26,4 @@ public abstract class WorkerJobRunning {
 
     abstract public String getType();
 
-    abstract public String uid();
 }

core/src/main/java/io/kestra/core/runners/WorkerTask.java

@@ -38,13 +38,11 @@ public Logger logger() {
         );
     }
 
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public String uid() {
-        return this.taskRun.getTaskId();
-    }
-
-    @Override
-    public String taskRunId() {
         return this.taskRun.getId();
     }