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.2.1-SNAPSHOT
-kestraVersion=1.2.5
+kestraVersion=2.0.0-SNAPSHOT

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/kinesis/RealtimeTriggerTest.java

@@ -1,45 +1,30 @@
 package io.kestra.plugin.aws.kinesis;
 
 import java.io.File;
 import java.nio.file.Files;
 import java.util.List;
-import java.util.UUID;
 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.runners.FlowListeners;
-import io.kestra.core.utils.TestsUtils;
-import io.kestra.jdbc.runner.JdbcScheduler;
 import io.kestra.plugin.aws.kinesis.model.Record;
-import io.kestra.scheduler.AbstractScheduler;
-import io.kestra.worker.DefaultWorker;
-
-import io.micronaut.context.ApplicationContext;
 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;
 import static org.hamcrest.Matchers.*;
 
+@KestraTest(startRunner = true, startScheduler = true)
 class RealtimeTriggerTest extends AbstractKinesisTest {
     @Inject
-    ApplicationContext applicationContext;
-
-    @Inject
-    FlowListeners flowListeners;
-
-    @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    QueueInterface<Execution> executionQueue;
+    DispatchQueueInterface<Execution> executionQueue;
 
     @Inject
     LocalFlowRepositoryLoader repositoryLoader;
@@ -48,14 +33,12 @@
     void evaluate() throws Exception {
         String consumerArn = registerConsumer();
         CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<Execution> lastExecution = new AtomicReference<>();
 
-        Flux<Execution> received = TestsUtils.receive(executionQueue, e -> latch.countDown());
-
-        DefaultWorker worker = applicationContext.createBean(DefaultWorker.class, UUID.randomUUID().toString(), 8, null);
-        try (AbstractScheduler scheduler = new JdbcScheduler(applicationContext, flowListeners)) {
-
-            worker.run();
-            scheduler.run();
+        executionQueue.addListener(e -> {
+            lastExecution.set(e);
+            latch.countDown();
+        });
 
             String yaml = """
                 id: realtime
@@ -103,8 +86,6 @@
             boolean done = latch.await(30, TimeUnit.SECONDS);
             assertThat(done, is(true));
 
-            Execution exec = received.blockLast();
-            assertThat(exec.getTrigger().getVariables().get("data"), is("hello"));
-        }
+            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));
     }