fix: 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
@@ -97,6 +97,7 @@ dependencies {
     testImplementation group: "io.kestra", name: "core", version: kestraVersion
     testImplementation group: "io.kestra", name: "tests", version: kestraVersion
     testImplementation group: "io.kestra", name: "script", version: kestraVersion
+    testImplementation group: "io.kestra", name: "indexer", version: kestraVersion
     testImplementation group: "io.kestra", name: "repository-memory", version: kestraVersion
     testImplementation group: "io.kestra", name: "runner-memory", version: kestraVersion
     testImplementation group: "io.kestra", name: "storage-local", version: kestraVersion

gradle.properties

@@ -1,2 +1,2 @@
 version=1.5.1-SNAPSHOT
-kestraVersion=1.3.3
+kestraVersion=2.0.0-SNAPSHOT

src/test/java/io/kestra/plugin/git/NamespaceSyncTest.java

@@ -200,7 +200,7 @@ private void createFlowInKestra(String id, String namespace) {
 
             tasks:
               - id: say
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: hello
             """.formatted(id, namespace);
         GenericFlow f = GenericFlow.fromYaml(TENANT_ID, src);

src/test/java/io/kestra/plugin/git/PushFlowsTest.java

@@ -819,10 +819,10 @@ private FlowWithSource createFlow(String tenantId, String flowId, String namespa
 
             tasks:
               - id: my-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from my-task
               - id: subflow
-                type: io.kestra.core.tasks.flows.Subflow
+                type: io.kestra.plugin.core.flow.Subflow
                 namespace:\s""" + namespace + """
             .sub-namespace
                 flowId: another-flow

src/test/java/io/kestra/plugin/git/PushTest.java

@@ -629,7 +629,7 @@ private FlowWithSource createFlow(String tenantId, String namespace) {
 
             tasks:
               - id: my-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from my-task""";
         return flowRepositoryInterface.create(
             GenericFlow.fromYaml(tenantId, flowSource)

src/test/java/io/kestra/plugin/git/SyncFlowTest.java

@@ -75,7 +75,7 @@ void createNewFlow() throws Exception {
     void updateExistingFlow() throws Exception {
         RunContext runContext = runContext();
 
-        String initialSource = "id: second-flow\nnamespace: " + TARGET_NAMESPACE + "\ntasks:\n  - id: log\n    type: io.kestra.core.tasks.log.Log\n    message: 'v1'";
+        String initialSource = "id: second-flow\nnamespace: " + TARGET_NAMESPACE + "\ntasks:\n  - id: log\n    type: io.kestra.plugin.core.log.Log\n    message: 'v1'";
         GenericFlow flow = GenericFlow.fromYaml(TENANT_ID, initialSource);
 
         flowRepositoryInterface.create(

src/test/java/io/kestra/plugin/git/SyncFlowsTest.java

@@ -73,7 +73,7 @@ void defaultCase_WithDelete() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         GenericFlow flow = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(flow);
@@ -105,7 +105,7 @@ void defaultCase_WithDelete() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         flow = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(flow);
@@ -118,7 +118,7 @@ void defaultCase_WithDelete() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         GenericFlow invalidFlow = GenericFlow.builder()
             .id("validation-failed-flow")
@@ -202,7 +202,7 @@ void defaultCase_WithoutDelete() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         GenericFlow genericFlow = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(genericFlow);
@@ -235,7 +235,7 @@ void defaultCase_WithoutDelete() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         genericFlow = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(genericFlow);
@@ -281,7 +281,7 @@ void defaultCase_WithDeleteNoChildNs() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         GenericFlow genericFlow = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(genericFlow);
@@ -327,7 +327,7 @@ void defaultCase_WithDeleteNoChildNs() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         genericFlow = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(genericFlow);
@@ -389,7 +389,7 @@ void dryRun_WithDelete() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         GenericFlow genericFlow = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(genericFlow);
@@ -425,7 +425,7 @@ void dryRun_WithDelete() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         genericFlow = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(genericFlow);

src/test/java/io/kestra/plugin/git/SyncTest.java

@@ -22,8 +22,7 @@
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.GenericFlow;
 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.FlowRepositoryInterface;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.RunContextFactory;
@@ -34,8 +33,6 @@
 import io.kestra.core.utils.TestsUtils;
 
 import jakarta.inject.Inject;
-import jakarta.inject.Named;
-import reactor.core.publisher.Flux;
 
 import static io.kestra.core.utils.Rethrow.throwFunction;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -59,8 +56,7 @@ class SyncTest extends AbstractGitTest {
     private StorageInterface storageInterface;
 
     @Inject
-    @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
-    private QueueInterface<LogEntry> logQueue;
+    private DispatchQueueInterface<LogEntry> logQueue;
 
     @BeforeEach
     void init() throws IOException {
@@ -79,7 +75,7 @@ void reconcileNsFilesAndFlows() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         GenericFlow genericFlow = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(genericFlow);
@@ -235,7 +231,7 @@ void reconcile_MinimumSetup() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
         GenericFlow flowToDelete = GenericFlow.fromYaml(TENANT_ID, flowSource);
         flowRepositoryInterface.create(flowToDelete);
@@ -315,7 +311,7 @@ void reconcile_MinimumSetup() throws Exception {
     @Test
     void reconcile_DryRun_ShouldDoNothing() throws Exception {
         List<LogEntry> logs = new CopyOnWriteArrayList<>();
-        Flux<LogEntry> receive = TestsUtils.receive(logQueue, l -> logs.add(l.getLeft()));
+        logQueue.addListener(logs::add);
         String namespace = SyncTest.class.getName().toLowerCase();
 
         String flowSource = """
@@ -324,7 +320,7 @@ void reconcile_DryRun_ShouldDoNothing() throws Exception {
 
             tasks:
               - id: old-task
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: Hello from old-task""";
 
         GenericFlow genericFlow = GenericFlow.fromYaml(TENANT_ID, flowSource);
@@ -386,7 +382,6 @@ void reconcile_DryRun_ShouldDoNothing() throws Exception {
         assertHasInfoLog(logs, "~ " + toUpdateFilePath);
         assertHasInfoLog(logs, "- " + someFilePath);
         assertHasInfoLog(logs, "+ /cloned.json");
-        receive.blockLast();
     }
 
     private static void assertHasInfoLog(List<LogEntry> logs, String expectedMessage) {

src/test/java/io/kestra/plugin/git/TenantSyncTest.java

@@ -48,7 +48,7 @@ void shouldFailOnInvalidTaskDefinitionFromKestraExport() throws Exception {
 
             tasks:
               - id: log
-                type: io.kestra.core.tasks.log.Log
+                type: io.kestra.plugin.core.log.Log
                 message: hello
             """;
         var exportedZip = zippedYaml("my.namespace/exported-flow.yaml", yaml);

src/test/resources/application.yml

@@ -14,7 +14,13 @@ kestra:
     type: memory
   queue:
     type: memory
+  worker:
+    controllers:
+      type: STATIC
+      static:
+        endpoints:
+          - host: localhost
   storage:
     type: local
     local:
-      base-path: /tmp/unittest
+      base-path: /tmp/unittest