fix: plugin v2 compatibility

.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

build.gradle

@@ -22,7 +22,7 @@ repositories {
     }
 }
 
-final targetJavaVersion = JavaVersion.VERSION_21
+final targetJavaVersion = JavaVersion.VERSION_25
 
 java {
     sourceCompatibility = targetJavaVersion

gradle.properties

@@ -1,2 +1,2 @@
 version=2.4.1-SNAPSHOT
-kestraVersion=1.3.13
+kestraVersion=2.0.0-SNAPSHOT

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: {{ json(taskrun.value) }}"
+                    message: "Datapoint: {{ fromJson(item.value) }}"
 
             triggers:
               - id: watch

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,20 +77,20 @@
 
                 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
                         accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
                         secretKeyId: "{{ secret('AWS_SECRET_KEY_ID') }}"
                         region: "eu-central-1"
                         bucket: "my-bucket"
-                        key: "{{ taskrun.value }}"
+                        key: "{{ item.value }}"
 
                 triggers:
                   - id: watch

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.AbstractLocalStackTest;
-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;
 
@@ -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")

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));

src/test/java/io/kestra/plugin/aws/kinesis/RealtimeTriggerTest.java

@@ -5,20 +5,17 @@
 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;
@@ -27,8 +24,7 @@
 @KestraTest(startRunner = true, startScheduler = true)
 class RealtimeTriggerTest extends AbstractKinesisTest {
     @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    QueueInterface<Execution> executionQueue;
+    DispatchQueueInterface<Execution> executionQueue;
 
     @Inject
     LocalFlowRepositoryLoader repositoryLoader;
@@ -37,56 +33,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: "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(localstack.getEndpoint().toString()))
-            .region(Property.ofValue(localstack.getRegion()))
-            .accessKeyId(Property.ofValue(localstack.getAccessKey()))
-            .secretKeyId(Property.ofValue(localstack.getSecretKey()))
-            .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(localstack.getEndpoint().toString()))
+                .region(Property.ofValue(localstack.getRegion()))
+                .accessKeyId(Property.ofValue(localstack.getAccessKey()))
+                .secretKeyId(Property.ofValue(localstack.getSecretKey()))
+                .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"));
     }
 }

src/test/java/io/kestra/plugin/aws/kinesis/TriggerTest.java

@@ -58,8 +58,8 @@ void evaluate() throws Exception {
             .endpointOverride(Property.ofValue(localstack.getEndpoint().toString()))
             .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));
     }