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

@@ -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
  **********************************************************************************************************************/

gradle.properties

@@ -1,2 +1,2 @@
 version=2.8.2-SNAPSHOT
-kestraVersion=1.3.19
+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: {{ fromJson(taskrun.value) }}"
+                    message: "Datapoint: {{ fromJson(item.value) }}"
 
             triggers:
               - id: watch

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

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

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

@@ -1,4 +1,4 @@
 package io.kestra.plugin.aws.cli;
 
 import java.util.Collections;
 import java.util.List;
@@ -14,16 +14,16 @@
 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;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 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 @@
         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

@@ -1,4 +1,4 @@
 package io.kestra.plugin.aws.cloudwatch;
 
 import java.time.Duration;
 import java.util.List;
@@ -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 @@
                 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/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));
     }