diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index 2d9fc98c..d509c2d0 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -46,4 +46,5 @@ jobs:
     with:
       skip-test: ${{ github.event.inputs.skip-test == 'true' }}
       kestra-version: ${{ github.event.inputs.kestra-version }}
+      java-version: '25'
     secrets: inherit
diff --git a/build.gradle b/build.gradle
index 0bd02cc4..19e236da 100644
--- a/build.gradle
+++ b/build.gradle
@@ -24,7 +24,7 @@ repositories {
     }
 }
 
-final targetJavaVersion = JavaVersion.VERSION_21
+final targetJavaVersion = JavaVersion.VERSION_25
 
 java {
     sourceCompatibility = targetJavaVersion
@@ -145,29 +145,6 @@ dependencies {
     testImplementation "io.floci:testcontainers-floci:2.11.0"
 }
 
-/**********************************************************************************************************************\
- * Allure Reports
- **********************************************************************************************************************/
-dependencies {
-    testAnnotationProcessor enforcedPlatform("io.kestra:platform:$kestraVersion")
-    testImplementation "io.qameta.allure:allure-junit5"
-}
-
-configurations {
-    agent {
-        canBeResolved = true
-        canBeConsumed = true
-    }
-}
-
-dependencies {
-    agent "org.aspectj:aspectjweaver:1.9.25.1"
-}
-
-test {
-    jvmArgs = [ "-javaagent:${configurations.agent.singleFile}" ]
-}
-
 /**********************************************************************************************************************\
  * Jacoco
  **********************************************************************************************************************/
diff --git a/gradle.properties b/gradle.properties
index 7f176bf3..88bbe9ac 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -1,2 +1,2 @@
 version=2.8.2-SNAPSHOT
-kestraVersion=1.3.19
+kestraVersion=2.0.0-SNAPSHOT
diff --git a/src/main/java/io/kestra/plugin/aws/cloudwatch/Trigger.java b/src/main/java/io/kestra/plugin/aws/cloudwatch/Trigger.java
index d15965c7..bd8e0587 100644
--- a/src/main/java/io/kestra/plugin/aws/cloudwatch/Trigger.java
+++ b/src/main/java/io/kestra/plugin/aws/cloudwatch/Trigger.java
@@ -37,12 +37,12 @@
             namespace: company.team
             tasks:
               - id: each
-                type: io.kestra.plugin.core.flow.ForEach
+                type: io.kestra.plugin.core.flow.Loop
                 values: "{{ trigger.series }}"
                 tasks:
                   - id: log
                     type: io.kestra.plugin.core.log.Log
-                    message: "Datapoint: {{ fromJson(taskrun.value) }}"
+                    message: "Datapoint: {{ fromJson(item.value) }}"
 
             triggers:
               - id: watch
diff --git a/src/main/java/io/kestra/plugin/aws/healthlake/Trigger.java b/src/main/java/io/kestra/plugin/aws/healthlake/Trigger.java
index eb9f14a0..388764ad 100644
--- a/src/main/java/io/kestra/plugin/aws/healthlake/Trigger.java
+++ b/src/main/java/io/kestra/plugin/aws/healthlake/Trigger.java
@@ -17,6 +17,8 @@
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.triggers.*;
 import io.kestra.core.runners.RunContext;
+import io.kestra.core.storages.kv.KVMetadata;
+import io.kestra.core.storages.kv.KVValueAndMetadata;
 import io.kestra.plugin.aws.AbstractConnectionInterface;
 import io.kestra.plugin.aws.ConnectionUtils;
 
@@ -132,7 +134,7 @@ public class Trigger extends AbstractTrigger implements PollingTriggerInterface,
         "COMPLETED", "COMPLETED_WITH_ERRORS", "FAILED", "CANCEL_COMPLETED", "CANCEL_FAILED"
     );
 
-    private static final String STATE_FILE = "healthlake-trigger-last-job-id.txt";
+    private static final String STATE_KV_PREFIX = "healthlake-trigger-last-job-id";
 
     @Override
     public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception {
@@ -142,10 +144,17 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
         var rDatastoreId = runContext.render(datastoreId).as(String.class).orElseThrow();
         var rJobType = runContext.render(jobType).as(JobType.class).orElse(JobType.IMPORT);
 
+        // trigger state moved from the removed RunContext#stateStore() to the namespace KV store in Kestra 2.0
+        var kvStore = runContext.namespaceKv(context.getNamespace());
+        var stateKey = STATE_KV_PREFIX + "-" + context.getFlowId() + "-" + this.getId();
+
         // Load last-fired jobId to avoid re-firing on the same terminal job every poll
         String lastFiredJobId = null;
-        try (var stateStream = runContext.stateStore().getState(context.getNamespace(), context.getFlowId(), STATE_FILE)) {
-            lastFiredJobId = new String(stateStream.readAllBytes(), StandardCharsets.UTF_8);
+        try {
+            var kvValue = kvStore.getValue(stateKey);
+            if (kvValue.isPresent() && kvValue.get().value() instanceof byte[] bytes && bytes.length > 0) {
+                lastFiredJobId = new String(bytes, StandardCharsets.UTF_8);
+            }
         } catch (Exception e) {
             logger.debug("No previous trigger state found, treating as first run");
         }
@@ -197,7 +206,10 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
             }
 
             // Persist the jobId so we don't fire on it again
-            runContext.stateStore().putState(context.getNamespace(), context.getFlowId(), STATE_FILE, jobId.getBytes(StandardCharsets.UTF_8));
+            kvStore.put(stateKey, new KVValueAndMetadata(
+                new KVMetadata("HealthLake trigger last-fired job id", (Duration) null),
+                jobId.getBytes(StandardCharsets.UTF_8)
+            ));
 
             logger.debug("Job '{}' reached terminal status={}, firing trigger", jobId, jobStatus);
             var output = Output.builder().jobId(jobId).jobStatus(jobStatus).build();
diff --git a/src/main/java/io/kestra/plugin/aws/msk/Trigger.java b/src/main/java/io/kestra/plugin/aws/msk/Trigger.java
index 889cfd07..3c25781a 100644
--- a/src/main/java/io/kestra/plugin/aws/msk/Trigger.java
+++ b/src/main/java/io/kestra/plugin/aws/msk/Trigger.java
@@ -14,6 +14,8 @@
 import io.kestra.core.models.triggers.TriggerOutput;
 import io.kestra.core.models.triggers.TriggerService;
 import io.kestra.core.runners.RunContext;
+import io.kestra.core.storages.kv.KVMetadata;
+import io.kestra.core.storages.kv.KVValueAndMetadata;
 import io.kestra.plugin.aws.AbstractConnectionInterface;
 import io.kestra.plugin.aws.ConnectionUtils;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -28,8 +30,6 @@
 import software.amazon.awssdk.services.kafka.model.ClusterState;
 import software.amazon.awssdk.services.kafka.model.DescribeClusterRequest;
 
-import java.io.BufferedReader;
-import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Optional;
@@ -78,8 +78,7 @@
 )
 public class Trigger extends AbstractTrigger implements PollingTriggerInterface, TriggerOutput<Trigger.Output>, AbstractConnectionInterface {
 
-    private static final String STATE_NAME = "msk-trigger";
-    private static final String STATE_SUB_NAME = "last-fired-state";
+    private static final String STATE_KV_PREFIX = "msk-trigger-last-fired-state";
 
     @Schema(title = "AWS access key ID", description = "Optional static credential. If omitted, the default credentials provider chain is used.")
     @PluginProperty(secret = true, group = "advanced")
@@ -151,14 +150,16 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
         var rArn = runContext.render(clusterArn).as(String.class).orElseThrow();
         var rTargetState = runContext.render(targetState).as(ClusterState.class).orElseThrow();
 
+        // trigger state moved from the removed RunContext#stateStore() to the namespace KV store in Kestra 2.0
+        var kvStore = runContext.namespaceKv(context.getNamespace());
+        var stateKey = STATE_KV_PREFIX + "-" + context.getFlowId() + "-" + this.getId();
+
         // Load last-fired state to avoid re-firing while cluster remains in target state
         String lastFiredState = null;
         try {
-            var stateStream = runContext.stateStore().getState(STATE_NAME, STATE_SUB_NAME, this.getId());
-            if (stateStream != null) {
-                try (var reader = new BufferedReader(new InputStreamReader(stateStream, StandardCharsets.UTF_8))) {
-                    lastFiredState = reader.readLine();
-                }
+            var kvValue = kvStore.getValue(stateKey);
+            if (kvValue.isPresent() && kvValue.get().value() instanceof byte[] bytes && bytes.length > 0) {
+                lastFiredState = new String(bytes, StandardCharsets.UTF_8);
             }
         } catch (Exception e) {
             logger.debug("No previous trigger state found, treating as first run");
@@ -175,7 +176,7 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
             if (currentState != rTargetState) {
                 // Clear persisted state so we re-fire if cluster transitions back into target later
                 if (lastFiredState != null) {
-                    runContext.stateStore().putState(STATE_NAME, STATE_SUB_NAME, this.getId(), new byte[0]);
+                    kvStore.delete(stateKey);
                 }
                 return Optional.empty();
             }
@@ -187,8 +188,10 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
             }
 
             // Persist fired state so subsequent polls don't re-fire
-            runContext.stateStore().putState(STATE_NAME, STATE_SUB_NAME, this.getId(),
-                rTargetState.toString().getBytes(StandardCharsets.UTF_8));
+            kvStore.put(stateKey, new KVValueAndMetadata(
+                new KVMetadata("MSK trigger last-fired state", (Duration) null),
+                rTargetState.toString().getBytes(StandardCharsets.UTF_8)
+            ));
 
             logger.debug("MSK cluster '{}' reached target state={}, firing trigger", rArn, rTargetState);
             var output = Output.builder()
diff --git a/src/main/java/io/kestra/plugin/aws/s3/Trigger.java b/src/main/java/io/kestra/plugin/aws/s3/Trigger.java
index b595088d..adf461fd 100644
--- a/src/main/java/io/kestra/plugin/aws/s3/Trigger.java
+++ b/src/main/java/io/kestra/plugin/aws/s3/Trigger.java
@@ -46,12 +46,12 @@
 
                 tasks:
                   - id: each
-                    type: io.kestra.plugin.core.flow.ForEach
+                    type: io.kestra.plugin.core.flow.Loop
                     values: "{{ trigger.objects | jq('.[].uri') }}"
                     tasks:
                       - id: return
                         type: io.kestra.plugin.core.debug.Return
-                        format: "{{ taskrun.value }}"
+                        format: "{{ item.value }}"
 
                 triggers:
                   - id: watch
@@ -77,12 +77,12 @@
 
                 tasks:
                   - id: each
-                    type: io.kestra.plugin.core.flow.ForEach
+                    type: io.kestra.plugin.core.flow.Loop
                     values: "{{ trigger.objects | jq('.[].key') }}"
                     tasks:
                       - id: return
                         type: io.kestra.plugin.core.debug.Return
-                        format: "{{ taskrun.value }}"
+                        format: "{{ item.value }}"
 
                       - id: delete
                         type: io.kestra.plugin.aws.s3.Delete
@@ -90,7 +90,7 @@
                         secretKeyId: "{{ secret('AWS_SECRET_KEY_ID') }}"
                         region: "eu-central-1"
                         bucket: "my-bucket"
-                        key: "{{ taskrun.value }}"
+                        key: "{{ item.value }}"
 
                 triggers:
                   - id: watch
diff --git a/src/test/java/io/kestra/plugin/aws/cli/AwsCLITest.java b/src/test/java/io/kestra/plugin/aws/cli/AwsCLITest.java
index 421a9c2c..1440c9d4 100644
--- a/src/test/java/io/kestra/plugin/aws/cli/AwsCLITest.java
+++ b/src/test/java/io/kestra/plugin/aws/cli/AwsCLITest.java
@@ -14,8 +14,8 @@
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.TestsUtils;
 import io.kestra.plugin.aws.AbstractFlociTest;
-import io.kestra.plugin.scripts.exec.scripts.models.DockerOptions;
 import io.kestra.plugin.scripts.exec.scripts.models.ScriptOutput;
+import io.kestra.plugin.scripts.runner.docker.Docker;
 
 import jakarta.inject.Inject;
 
@@ -23,7 +23,7 @@
 import static org.hamcrest.Matchers.hasEntry;
 import static org.hamcrest.Matchers.is;
 
-public class AwsCLITest extends AbstractFlociTest {
+class AwsCLITest extends AbstractFlociTest {
     @Inject
     private RunContextFactory runContextFactory;
 
@@ -36,8 +36,9 @@ void run() throws Exception {
         AwsCLI execute = AwsCLI.builder()
             .id(IdUtils.create())
             .type(AwsCLI.class.getName())
-            .docker(
-                DockerOptions.builder()
+            .taskRunner(
+                Docker.builder()
+                    .type(Docker.class.getName())
                     // needed to be able to reach localstack from inside the container
                     .networkMode("host")
                     .image("amazon/aws-cli")
diff --git a/src/test/java/io/kestra/plugin/aws/cloudwatch/TriggerTest.java b/src/test/java/io/kestra/plugin/aws/cloudwatch/TriggerTest.java
index 7be9cd4b..b8932f06 100644
--- a/src/test/java/io/kestra/plugin/aws/cloudwatch/TriggerTest.java
+++ b/src/test/java/io/kestra/plugin/aws/cloudwatch/TriggerTest.java
@@ -20,6 +20,7 @@
 import static org.mockito.Mockito.mockConstruction;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
+import io.kestra.core.models.conditions.ConditionContext;
 
 @KestraTest
 class TriggerTest {
@@ -59,8 +60,8 @@ void evaluate() throws Exception {
                 Query.class, (mock, context) -> when(mock.run(any())).thenReturn(output)
             )
         ) {
-            var conditionContext = io.kestra.core.utils.TestsUtils.mockTrigger(runContextFactory, trigger);
-            var execution = trigger.evaluate(conditionContext.getKey(), conditionContext.getValue());
+            Map.Entry<ConditionContext, io.kestra.core.scheduler.model.TriggerState> conditionContext = io.kestra.core.utils.TestsUtils.mockTrigger(runContextFactory, trigger);
+            var execution = trigger.evaluate(conditionContext.getKey(), conditionContext.getValue().context());
 
             assertThat(execution.isPresent(), is(true));
             assertThat(mockedQuery.constructed(), hasSize(1));
diff --git a/src/test/java/io/kestra/plugin/aws/healthlake/TriggerTest.java b/src/test/java/io/kestra/plugin/aws/healthlake/TriggerTest.java
index a4e38378..5b29ada8 100644
--- a/src/test/java/io/kestra/plugin/aws/healthlake/TriggerTest.java
+++ b/src/test/java/io/kestra/plugin/aws/healthlake/TriggerTest.java
@@ -38,7 +38,7 @@ void givenCompletedJob_whenEvaluate_thenReturnsExecution() throws Exception {
         doReturn(mockClient).when(spy).client(any(RunContext.class));
 
         var context = TestsUtils.mockTrigger(runContextFactory, spy);
-        var execution = spy.evaluate(context.getKey(), context.getValue());
+        var execution = spy.evaluate(context.getKey(), context.getValue().context());
 
         assertThat(execution.isPresent(), is(true));
         verify(mockClient).listFHIRImportJobs(any(ListFhirImportJobsRequest.class));
@@ -57,7 +57,7 @@ void givenInProgressJob_whenEvaluate_thenReturnsEmpty() throws Exception {
         doReturn(mockClient).when(spy).client(any(RunContext.class));
 
         var context = TestsUtils.mockTrigger(runContextFactory, spy);
-        var execution = spy.evaluate(context.getKey(), context.getValue());
+        var execution = spy.evaluate(context.getKey(), context.getValue().context());
 
         assertThat(execution.isEmpty(), is(true));
     }
@@ -70,7 +70,7 @@ void givenNoJobs_whenEvaluate_thenReturnsEmpty() throws Exception {
         doReturn(mockClient).when(spy).client(any(RunContext.class));
 
         var context = TestsUtils.mockTrigger(runContextFactory, spy);
-        var execution = spy.evaluate(context.getKey(), context.getValue());
+        var execution = spy.evaluate(context.getKey(), context.getValue().context());
 
         assertThat(execution.isEmpty(), is(true));
     }
diff --git a/src/test/java/io/kestra/plugin/aws/kinesis/RealtimeTriggerTest.java b/src/test/java/io/kestra/plugin/aws/kinesis/RealtimeTriggerTest.java
index df5062be..e15da7ff 100644
--- a/src/test/java/io/kestra/plugin/aws/kinesis/RealtimeTriggerTest.java
+++ b/src/test/java/io/kestra/plugin/aws/kinesis/RealtimeTriggerTest.java
@@ -5,21 +5,18 @@
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.jupiter.api.*;
 
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.property.Property;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.queues.DispatchQueueInterface;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
-import io.kestra.core.utils.TestsUtils;
 import io.kestra.plugin.aws.kinesis.model.Record;
 
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
-import reactor.core.publisher.Flux;
 import software.amazon.awssdk.services.kinesis.model.*;
 
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -28,8 +25,7 @@
 @KestraTest(startRunner = true, startScheduler = true)
 class RealtimeTriggerTest extends AbstractKinesisTest {
     @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    QueueInterface<Execution> executionQueue;
+    DispatchQueueInterface<Execution> executionQueue;
 
     @Inject
     LocalFlowRepositoryLoader repositoryLoader;
@@ -38,56 +34,59 @@ class RealtimeTriggerTest extends AbstractKinesisTest {
     void evaluate() throws Exception {
         String consumerArn = registerConsumer();
         CountDownLatch latch = new CountDownLatch(1);
-
-        Flux<Execution> received = TestsUtils.receive(executionQueue, e -> latch.countDown());
-
-        String yaml = """
-            id: realtime
-            namespace: company.team
-
-            tasks:
-              - id: log
-                type: io.kestra.plugin.core.log.Log
-                message: "{{ trigger.data }}"
-
-            triggers:
-              - id: realtime
-                type: io.kestra.plugin.aws.kinesis.RealtimeTrigger
-                streamName: "%s"
-                consumerArn: "%s"
-                region: "us-east-1"
-                accessKeyId: "test"
-                secretKeyId: "test"
-                endpointOverride: "%s"
-                iteratorType: TRIM_HORIZON
-            """
-            .formatted(streamName, consumerArn, endpointUrl());
-
-        File tempFlow = File.createTempFile("kinesis-realtime", ".yaml");
-        Files.writeString(tempFlow.toPath(), yaml);
-
-        repositoryLoader.load(tempFlow);
-
-        Record record = Record.builder()
-            .partitionKey("pk")
-            .data("hello")
-            .build();
-
-        var put = PutRecords.builder()
-            .endpointOverride(Property.ofValue(endpointUrl()))
-            .region(Property.ofValue(REGION))
-            .accessKeyId(Property.ofValue(ACCESS_KEY))
-            .secretKeyId(Property.ofValue(SECRET_KEY))
-            .streamName(Property.ofValue(streamName))
-            .records(List.of(record))
-            .build();
-
-        put.run(runContextFactory.of());
-
-        boolean done = latch.await(30, TimeUnit.SECONDS);
-        assertThat(done, is(true));
-
-        Execution exec = received.blockLast();
-        assertThat(exec.getTrigger().getVariables().get("data"), is("hello"));
+        AtomicReference<Execution> lastExecution = new AtomicReference<>();
+
+        executionQueue.addListener(e -> {
+            lastExecution.set(e);
+            latch.countDown();
+        });
+
+            String yaml = """
+                id: realtime
+                namespace: company.team
+
+                tasks:
+                  - id: log
+                    type: io.kestra.plugin.core.log.Log
+                    message: "{{ trigger.data }}"
+
+                triggers:
+                  - id: realtime
+                    type: io.kestra.plugin.aws.kinesis.RealtimeTrigger
+                    streamName: "%s"
+                    consumerArn: "%s"
+                    region: "us-east-1"
+                    accessKeyId: "test"
+                    secretKeyId: "test"
+                    endpointOverride: "http://localhost:4566"
+                    iteratorType: TRIM_HORIZON
+                """
+                .formatted(streamName, consumerArn);
+
+            File tempFlow = File.createTempFile("kinesis-realtime", ".yaml");
+            Files.writeString(tempFlow.toPath(), yaml);
+
+            repositoryLoader.load(tempFlow);
+
+            Record record = Record.builder()
+                .partitionKey("pk")
+                .data("hello")
+                .build();
+
+            var put = PutRecords.builder()
+                .endpointOverride(Property.ofValue(endpointUrl()))
+                .region(Property.ofValue(REGION))
+                .accessKeyId(Property.ofValue(ACCESS_KEY))
+                .secretKeyId(Property.ofValue(SECRET_KEY))
+                .streamName(Property.ofValue(streamName))
+                .records(List.of(record))
+                .build();
+
+            put.run(runContextFactory.of());
+
+            boolean done = latch.await(30, TimeUnit.SECONDS);
+            assertThat(done, is(true));
+
+            assertThat(lastExecution.get().getTrigger().getVariables().get("data"), is("hello"));
     }
 }
diff --git a/src/test/java/io/kestra/plugin/aws/kinesis/TriggerTest.java b/src/test/java/io/kestra/plugin/aws/kinesis/TriggerTest.java
index a11f9c6f..08429b0b 100644
--- a/src/test/java/io/kestra/plugin/aws/kinesis/TriggerTest.java
+++ b/src/test/java/io/kestra/plugin/aws/kinesis/TriggerTest.java
@@ -58,8 +58,8 @@ void evaluate() throws Exception {
             .endpointOverride(Property.ofValue(endpointUrl()))
             .build();
 
-        Map.Entry<ConditionContext, io.kestra.core.models.triggers.Trigger> context = TestsUtils.mockTrigger(runContextFactory, trigger);
-        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue());
+        Map.Entry<ConditionContext, io.kestra.core.scheduler.model.TriggerState> context = TestsUtils.mockTrigger(runContextFactory, trigger);
+        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue().context());
 
         assertThat(execution.isPresent(), is(true));
     }
diff --git a/src/test/java/io/kestra/plugin/aws/msk/TriggerTest.java b/src/test/java/io/kestra/plugin/aws/msk/TriggerTest.java
index 999bdc5c..27c734ff 100644
--- a/src/test/java/io/kestra/plugin/aws/msk/TriggerTest.java
+++ b/src/test/java/io/kestra/plugin/aws/msk/TriggerTest.java
@@ -59,7 +59,7 @@ void givenClusterAtTargetState_whenEvaluate_thenReturnsExecution() throws Except
         var spy = spy(trigger);
         doReturn(mockClient).when(spy).client(any(RunContext.class));
 
-        Optional<Execution> result = spy.evaluate(conditionContext.getKey(), conditionContext.getValue());
+        Optional<Execution> result = spy.evaluate(conditionContext.getKey(), conditionContext.getValue().context());
 
         assertThat(result.isPresent(), is(true));
     }
@@ -94,7 +94,7 @@ void givenClusterNotAtTargetState_whenEvaluate_thenReturnsEmpty() throws Excepti
         var spy = spy(trigger);
         doReturn(mockClient).when(spy).client(any(RunContext.class));
 
-        Optional<Execution> result = spy.evaluate(conditionContext.getKey(), conditionContext.getValue());
+        Optional<Execution> result = spy.evaluate(conditionContext.getKey(), conditionContext.getValue().context());
 
         assertThat(result.isPresent(), is(false));
     }
diff --git a/src/test/java/io/kestra/plugin/aws/s3/TriggerTest.java b/src/test/java/io/kestra/plugin/aws/s3/TriggerTest.java
index 7c2817a6..e477c134 100644
--- a/src/test/java/io/kestra/plugin/aws/s3/TriggerTest.java
+++ b/src/test/java/io/kestra/plugin/aws/s3/TriggerTest.java
@@ -9,23 +9,21 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
 
 import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.core.junit.annotations.LoadFlows;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.triggers.StatefulTriggerInterface;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
-import io.kestra.core.repositories.LocalFlowRepositoryLoader;
+import io.kestra.core.queues.DispatchQueueInterface;
+import io.kestra.core.runners.Scheduler;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.TestsUtils;
 import io.kestra.plugin.aws.s3.models.S3Object;
-
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
-import reactor.core.publisher.Flux;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
@@ -33,23 +31,24 @@
 @KestraTest(startRunner = true, startScheduler = true)
 class TriggerTest extends AbstractTest {
     @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    private QueueInterface<Execution> executionQueue;
+    private DispatchQueueInterface<Execution> executionQueue;
 
     @Inject
-    protected LocalFlowRepositoryLoader repositoryLoader;
+    protected Scheduler scheduler;
 
     @Test
+    @LoadFlows({"flows/s3/s3-listen.yaml"})
     void deleteAction() throws Exception {
+        Awaitility.await().atMost(Duration.ofSeconds(20)).pollInterval(Duration.ofMillis(100)).until(() -> scheduler.isActive());
+
         String bucket = "trigger-test";
         this.createBucket(bucket);
         List listTask = list().bucket(Property.ofValue(bucket)).build();
+
         CountDownLatch queueCount = new CountDownLatch(1);
         AtomicReference<Execution> last = new AtomicReference<>();
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, executionWithError ->
-        {
-            Execution execution = executionWithError.getLeft();
 
+        executionQueue.addListener(execution -> {
             if (execution.getFlowId().equals("s3-listen")) {
                 last.set(execution);
                 queueCount.countDown();
@@ -59,11 +58,8 @@ void deleteAction() throws Exception {
         upload("trigger/s3", bucket);
         upload("trigger/s3", bucket);
 
-        repositoryLoader.load(flowWithFlociEndpoint("flows/s3/s3-listen.yaml"));
-
-        boolean await = queueCount.await(15, TimeUnit.SECONDS);
+        boolean await = queueCount.await(1, TimeUnit.MINUTES);
         assertThat(await, is(true));
-        receive.blockLast();
 
         @SuppressWarnings("unchecked")
         java.util.List<S3Object> trigger = (java.util.List<S3Object>) last.get().getTrigger().getVariables().get("objects");
@@ -77,17 +73,17 @@ void deleteAction() throws Exception {
     }
 
     @Test
+    @LoadFlows({"flows/s3/s3-listen-none-action.yaml"})
     void noneAction() throws Exception {
+        Awaitility.await().atMost(Duration.ofSeconds(20)).pollInterval(Duration.ofMillis(100)).until(() -> scheduler.isActive());
+
         String bucket = "trigger-none-action-test";
         this.createBucket(bucket);
         List listTask = list().bucket(Property.ofValue(bucket)).build();
 
         CountDownLatch queueCount = new CountDownLatch(1);
         AtomicReference<Execution> last = new AtomicReference<>();
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, executionWithError ->
-        {
-            Execution execution = executionWithError.getLeft();
-
+        executionQueue.addListener(execution -> {
             if (execution.getFlowId().equals("s3-listen-none-action")) {
                 last.set(execution);
                 queueCount.countDown();
@@ -97,11 +93,8 @@ void noneAction() throws Exception {
         upload("trigger/s3", bucket);
         upload("trigger/s3", bucket);
 
-        repositoryLoader.load(flowWithFlociEndpoint("flows/s3/s3-listen-none-action.yaml"));
-
-        boolean await = queueCount.await(10, TimeUnit.SECONDS);
+        boolean await = queueCount.await(1, TimeUnit.MINUTES);
         assertThat(await, is(true));
-        receive.blockLast();
 
         @SuppressWarnings("unchecked")
         java.util.List<S3Object> trigger = (java.util.List<S3Object>) last.get().getTrigger().getVariables().get("objects");
@@ -115,17 +108,18 @@ void noneAction() throws Exception {
     }
 
     @Test
+    @LoadFlows({"flows/s3/s3-listen-localhost-force-path-style.yaml"})
     void forcePathStyleWithSimpleLocalhost() throws Exception {
+        Awaitility.await().atMost(Duration.ofSeconds(20)).pollInterval(Duration.ofMillis(100)).until(() -> scheduler.isActive());
+
         String bucket = "trigger-force-path-style-test";
         this.createBucket(bucket);
         List listTask = list().bucket(Property.ofValue(bucket)).build();
 
         CountDownLatch queueCount = new CountDownLatch(1);
         AtomicReference<Execution> last = new AtomicReference<>();
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, executionWithError ->
-        {
-            Execution execution = executionWithError.getLeft();
 
+        executionQueue.addListener(execution -> {
             if (execution.getFlowId().equals("s3-listen-localhost-force-path-style")) {
                 last.set(execution);
                 queueCount.countDown();
@@ -135,11 +129,8 @@ void forcePathStyleWithSimpleLocalhost() throws Exception {
         upload("trigger/s3", bucket);
         upload("trigger/s3", bucket);
 
-        repositoryLoader.load(flowWithFlociEndpoint("flows/s3/s3-listen-localhost-force-path-style.yaml"));
-
-        boolean await = queueCount.await(15, TimeUnit.SECONDS);
+        boolean await = queueCount.await(1, TimeUnit.MINUTES);
         assertThat("trigger should work with localhost endpoint + forcePathStyle", await, is(true));
-        receive.blockLast();
 
         @SuppressWarnings("unchecked")
         java.util.List<S3Object> trigger = (java.util.List<S3Object>) last.get().getTrigger().getVariables().get("objects");
@@ -155,7 +146,7 @@ void forcePathStyleWithSimpleLocalhost() throws Exception {
     private File flowWithFlociEndpoint(String resource) throws Exception {
         String yaml = new String(TriggerTest.class.getClassLoader().getResourceAsStream(resource).readAllBytes());
         yaml = yaml.replace("http://localhost:4566", endpointUrl())
-            .replace("http://127.0.0.1:4566", endpointUrl());
+                   .replace("http://127.0.0.1:4566", endpointUrl());
         File tempFlow = File.createTempFile("s3-trigger", ".yaml");
         Files.writeString(tempFlow.toPath(), yaml);
         return tempFlow;
@@ -184,9 +175,9 @@ void shouldExecuteOnCreate() throws Exception {
 
         upload("trigger/on-create", bucket);
 
-        Map.Entry<ConditionContext, io.kestra.core.models.triggers.Trigger> context = TestsUtils.mockTrigger(runContextFactory, trigger);
+        Map.Entry<ConditionContext, io.kestra.core.scheduler.model.TriggerState> context = TestsUtils.mockTrigger(runContextFactory, trigger);
 
-        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue());
+        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue().context());
         assertThat(execution.isPresent(), is(true));
     }
 
@@ -213,14 +204,14 @@ void shouldExecuteOnUpdate() throws Exception {
             .interval(Duration.ofSeconds(10))
             .build();
 
-        Map.Entry<ConditionContext, io.kestra.core.models.triggers.Trigger> context = TestsUtils.mockTrigger(runContextFactory, trigger);
+        Map.Entry<ConditionContext, io.kestra.core.scheduler.model.TriggerState> context = TestsUtils.mockTrigger(runContextFactory, trigger);
 
-        trigger.evaluate(context.getKey(), context.getValue());
+        trigger.evaluate(context.getKey(), context.getValue().context());
 
         update(key, bucket);
         Thread.sleep(2000);
 
-        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue());
+        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue().context());
         assertThat(execution.isPresent(), is(true));
     }
 
@@ -246,15 +237,15 @@ void shouldExecuteOnCreateOrUpdate() throws Exception {
 
         var key = upload("trigger/on-create-or-update", bucket);
 
-        Map.Entry<ConditionContext, io.kestra.core.models.triggers.Trigger> context = TestsUtils.mockTrigger(runContextFactory, trigger);
+        Map.Entry<ConditionContext, io.kestra.core.scheduler.model.TriggerState> context = TestsUtils.mockTrigger(runContextFactory, trigger);
 
-        Optional<Execution> createExecution = trigger.evaluate(context.getKey(), context.getValue());
+        Optional<Execution> createExecution = trigger.evaluate(context.getKey(), context.getValue().context());
         assertThat("Trigger should fire on CREATE", createExecution.isPresent(), is(true));
 
         update(key, bucket);
         Thread.sleep(2000);
 
-        Optional<Execution> updateExecution = trigger.evaluate(context.getKey(), context.getValue());
+        Optional<Execution> updateExecution = trigger.evaluate(context.getKey(), context.getValue().context());
         assertThat(updateExecution.isPresent(), is(true));
     }
 
@@ -285,9 +276,9 @@ void maxFilesExceeded() throws Exception {
             .interval(Duration.ofSeconds(10))
             .build();
 
-        Map.Entry<ConditionContext, io.kestra.core.models.triggers.Trigger> context = TestsUtils.mockTrigger(runContextFactory, trigger);
+        Map.Entry<ConditionContext, io.kestra.core.scheduler.model.TriggerState> context = TestsUtils.mockTrigger(runContextFactory, trigger);
 
-        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue());
+        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue().context());
         // When maxFiles exceeded, List returns first 3 files, so Trigger should fire
         assertThat(execution.isPresent(), is(true));
     }
@@ -319,9 +310,9 @@ void maxFilesNotExceeded() throws Exception {
             .interval(Duration.ofSeconds(10))
             .build();
 
-        Map.Entry<ConditionContext, io.kestra.core.models.triggers.Trigger> context = TestsUtils.mockTrigger(runContextFactory, trigger);
+        Map.Entry<ConditionContext, io.kestra.core.scheduler.model.TriggerState> context = TestsUtils.mockTrigger(runContextFactory, trigger);
 
-        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue());
+        Optional<Execution> execution = trigger.evaluate(context.getKey(), context.getValue().context());
         assertThat(execution.isPresent(), is(true));
     }
 }
diff --git a/src/test/java/io/kestra/plugin/aws/sqs/RealtimeTriggerTest.java b/src/test/java/io/kestra/plugin/aws/sqs/RealtimeTriggerTest.java
index ebfbd89e..45cea815 100644
--- a/src/test/java/io/kestra/plugin/aws/sqs/RealtimeTriggerTest.java
+++ b/src/test/java/io/kestra/plugin/aws/sqs/RealtimeTriggerTest.java
@@ -5,34 +5,25 @@
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.parallel.ExecutionMode;
-import org.junit.jupiter.api.parallel.ResourceLock;
 
 import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.property.Property;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.queues.DispatchQueueInterface;
 import io.kestra.core.repositories.LocalFlowRepositoryLoader;
-import io.kestra.core.utils.TestsUtils;
 import io.kestra.plugin.aws.sqs.model.Message;
-
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
-import reactor.core.publisher.Flux;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 
 @KestraTest(startRunner = true, startScheduler = true)
-@org.junit.jupiter.api.parallel.Execution(ExecutionMode.SAME_THREAD)
-@ResourceLock("kestra-sqs-trigger")
 class RealtimeTriggerTest extends AbstractSqsTest {
     @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    private QueueInterface<Execution> executionQueue;
+    private DispatchQueueInterface<Execution> executionQueue;
 
     @Inject
     protected LocalFlowRepositoryLoader repositoryLoader;
@@ -40,11 +31,11 @@ class RealtimeTriggerTest extends AbstractSqsTest {
     @Test
     void flow() throws Exception {
         CountDownLatch queueCount = new CountDownLatch(1);
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, execution ->
-        {
-            if (execution.isLeft() && "realtime".equals(execution.getLeft().getFlowId()) && "io.kestra.tests".equals(execution.getLeft().getNamespace())) {
-                queueCount.countDown();
-            }
+        AtomicReference<Execution> lastExecution = new AtomicReference<>();
+        executionQueue.addListener(execution -> {
+            lastExecution.set(execution);
+            queueCount.countDown();
+            assertThat(execution.getFlowId(), is("realtime"));
         });
 
         String yaml = """
@@ -82,12 +73,14 @@ void flow() throws Exception {
             )
             .build();
 
-        task.run(runContextFactory.of());
+        var runContext = runContextFactory.of();
+
+        task.run(runContext);
 
         boolean await = queueCount.await(1, TimeUnit.MINUTES);
         assertThat(await, is(true));
 
-        Execution last = receive.blockLast();
+        Execution last = lastExecution.get();
         assertThat(last.getTrigger().getVariables().size(), is(1));
         assertThat(last.getTrigger().getVariables().get("data"), is("Hello World"));
     }
diff --git a/src/test/java/io/kestra/plugin/aws/sqs/TriggerTest.java b/src/test/java/io/kestra/plugin/aws/sqs/TriggerTest.java
index bfbce0f1..2a75ae02 100644
--- a/src/test/java/io/kestra/plugin/aws/sqs/TriggerTest.java
+++ b/src/test/java/io/kestra/plugin/aws/sqs/TriggerTest.java
@@ -1,79 +1,52 @@
 package io.kestra.plugin.aws.sqs;
 
-import java.io.File;
-import java.nio.file.Files;
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.parallel.ResourceLock;
 
 import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.core.junit.annotations.LoadFlows;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.property.Property;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
-import io.kestra.core.repositories.LocalFlowRepositoryLoader;
+import io.kestra.core.queues.DispatchQueueInterface;
 import io.kestra.core.runners.RunContextFactory;
-import io.kestra.core.utils.TestsUtils;
+import io.kestra.core.runners.Scheduler;
 import io.kestra.plugin.aws.sqs.model.Message;
-
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
-import reactor.core.publisher.Flux;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 
 @KestraTest(startRunner = true, startScheduler = true)
-@ResourceLock("kestra-sqs-trigger")
 class TriggerTest extends AbstractSqsTest {
     @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    private QueueInterface<Execution> executionQueue;
+    private DispatchQueueInterface<Execution> executionQueue;
 
     @Inject
-    protected LocalFlowRepositoryLoader repositoryLoader;
+    protected Scheduler scheduler;
 
     @Inject
     protected RunContextFactory runContextFactory;
 
     @Test
+    @LoadFlows({"flows/sqs/sqs-listen.yaml"})
     void flow() throws Exception {
+        Awaitility.await().atMost(Duration.ofSeconds(20)).pollInterval(Duration.ofMillis(100)).until(() -> scheduler.isActive());
+
         CountDownLatch queueCount = new CountDownLatch(1);
-        Flux<Execution> receive = TestsUtils.receive(executionQueue, execution ->
-        {
-            if (execution.isLeft() && "sqs-listen".equals(execution.getLeft().getFlowId())) {
-                queueCount.countDown();
-            }
+        AtomicReference<Execution> lastExecution = new AtomicReference<>();
+        executionQueue.addListener(execution -> {
+            lastExecution.set(execution);
+            queueCount.countDown();
+            assertThat(execution.getFlowId(), is("sqs-listen"));
         });
 
-        String yaml = """
-            id: sqs-listen
-            namespace: io.kestra.tests
-
-            triggers:
-              - id: watch
-                type: io.kestra.plugin.aws.sqs.Trigger
-                endpointOverride: "%s"
-                queueUrl: "%s"
-                region: "us-east-1"
-                accessKeyId: "accesskey"
-                secretKeyId: "secretkey"
-                maxRecords: 2
-                interval: PT10S
-
-            tasks:
-              - id: end
-                type: io.kestra.plugin.core.debug.Return
-                format: "{{task.id}} > {{taskrun.startDate}}"
-            """.formatted(endpointUrl(), queueUrl());
-        File tempFlow = File.createTempFile("sqs-listen", ".yaml");
-        Files.writeString(tempFlow.toPath(), yaml);
-        repositoryLoader.load(tempFlow);
-
         Publish task = Publish.builder()
             .endpointOverride(Property.ofValue(endpointUrl()))
             .queueUrl(Property.ofValue(queueUrl()))
@@ -88,12 +61,14 @@ void flow() throws Exception {
             )
             .build();
 
-        task.run(runContextFactory.of());
+        var runContext = runContextFactory.of();
+
+        task.run(runContext);
 
         boolean await = queueCount.await(1, TimeUnit.MINUTES);
         assertThat(await, is(true));
 
-        Execution last = receive.filter(e -> "sqs-listen".equals(e.getFlowId())).blockLast();
+        Execution last = lastExecution.get();
         var count = (Integer) last.getTrigger().getVariables().get("count");
         var uri = (String) last.getTrigger().getVariables().get("uri");
         assertThat(count, is(2));
diff --git a/src/test/resources/allure.properties b/src/test/resources/allure.properties
deleted file mode 100644
index 4873f6d0..00000000
--- a/src/test/resources/allure.properties
+++ /dev/null
@@ -1 +0,0 @@
-allure.results.directory=build/allure-results
diff --git a/src/test/resources/application.yml b/src/test/resources/application.yml
index f1bb4dd5..ad450b21 100644
--- a/src/test/resources/application.yml
+++ b/src/test/resources/application.yml
@@ -9,3 +9,10 @@ kestra:
     type: memory
   repository:
     type: memory
+
+  worker:
+    controllers:
+      type: STATIC
+      static:
+        endpoints:
+          - host: localhost
\ No newline at end of file