Merge branch 'main' into dependabot/github_actions/docker/setup-qemu-action-3

Author: nthmost-orkes

.github/workflows/publish.yml

@@ -92,10 +92,10 @@ jobs:
         uses: docker/setup-qemu-action@v4
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4
 
       - name: Build and push Server
-        uses: docker/build-push-action@v3
+        uses: docker/build-push-action@v7
         with:
           context: .
           file: docker/server/Dockerfile

.github/workflows/publish_build.yaml

@@ -102,10 +102,10 @@ jobs:
         uses: docker/setup-qemu-action@v4
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@v4
 
       - name: Build and push Server
-        uses: docker/build-push-action@v3
+        uses: docker/build-push-action@v7
         with:
           context: .
           file: docker/server/Dockerfile

.github/workflows/ui-next-ci.yml

@@ -0,0 +1,61 @@
+name: UI v2 CI
+
+on:
+  pull_request:
+    branches:
+      - main
+    paths:
+      - "ui-next/**"
+
+permissions:
+  contents: read
+
+jobs:
+  lint-format-test:
+    name: Lint, Format & Test
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: ui-next
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: 22
+
+      - name: Setup pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          version: 10.32.0
+
+      - name: Get pnpm store directory
+        id: pnpm-cache
+        run: echo "store=$(pnpm store path)" >> $GITHUB_OUTPUT
+
+      - name: Cache pnpm store
+        uses: actions/cache@v4
+        with:
+          path: ${{ steps.pnpm-cache.outputs.store }}
+          key: ${{ runner.os }}-pnpm-${{ hashFiles('ui-next/pnpm-lock.yaml') }}
+          restore-keys: ${{ runner.os }}-pnpm-
+
+      - name: Install dependencies
+        run: pnpm install --frozen-lockfile
+
+      - name: Prettier check
+        run: pnpm prettier:check
+
+      - name: Lint
+        run: pnpm lint
+
+      - name: Type check
+        run: pnpm typecheck
+
+      - name: Test
+        run: pnpm test
+
+      - name: Build
+        run: pnpm build

annotations-processor/build.gradle

@@ -5,7 +5,7 @@ sourceSets {
 
 dependencies {
     implementation project(':conductor-annotations')
-    api 'com.google.guava:guava:31.1-jre'
+    api 'com.google.guava:guava:33.5.0-jre'
     api 'com.squareup:javapoet:1.13.0'
     api 'com.github.jknack:handlebars:4.5.0'
     api "com.google.protobuf:protobuf-java:${revProtoBuf}"

common/src/main/java/com/netflix/conductor/common/metadata/workflow/WorkflowTask.java

@@ -21,6 +21,7 @@
 import java.util.Map;
 import java.util.Objects;
 
+import com.netflix.conductor.annotations.protogen.ProtoEnum;
 import com.netflix.conductor.annotations.protogen.ProtoField;
 import com.netflix.conductor.annotations.protogen.ProtoMessage;
 import com.netflix.conductor.common.metadata.tasks.TaskDef;
@@ -165,6 +166,16 @@ public void setTasks(List<WorkflowTask> tasks) {
     @ProtoField(id = 32)
     private boolean permissive;
 
+    /** Controls whether a JOIN task is evaluated synchronously (no backoff) or asynchronously. */
+    @ProtoEnum
+    public enum JoinMode {
+        SYNC,
+        ASYNC
+    }
+
+    @ProtoField(id = 34)
+    private JoinMode joinMode;
+
     /**
      * @return the name
      */
@@ -601,6 +612,20 @@ public void setPermissive(boolean permissive) {
         this.permissive = permissive;
     }
 
+    /**
+     * @return the join mode (SYNC or ASYNC)
+     */
+    public JoinMode getJoinMode() {
+        return joinMode;
+    }
+
+    /**
+     * @param joinMode the join mode to set
+     */
+    public void setJoinMode(JoinMode joinMode) {
+        this.joinMode = joinMode;
+    }
+
     private Collection<List<WorkflowTask>> children() {
         Collection<List<WorkflowTask>> workflowTaskLists = new LinkedList<>();
 

core/src/main/java/com/netflix/conductor/core/execution/tasks/Join.java

@@ -20,6 +20,7 @@
 import org.springframework.stereotype.Component;
 
 import com.netflix.conductor.annotations.VisibleForTesting;
+import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
 import com.netflix.conductor.common.utils.TaskUtils;
 import com.netflix.conductor.core.config.ConductorProperties;
 import com.netflix.conductor.core.execution.WorkflowExecutor;
@@ -125,6 +126,15 @@ public boolean execute(
 
     @Override
     public Optional<Long> getEvaluationOffset(TaskModel taskModel, long maxOffset) {
+        // Check if joinMode is set to SYNC — read directly from the workflow task definition
+        // rather than from input data so the value is never duplicated into the task's payload.
+        WorkflowTask workflowTask = taskModel.getWorkflowTask();
+        if (workflowTask != null && WorkflowTask.JoinMode.SYNC == workflowTask.getJoinMode()) {
+            // Synchronous mode: evaluate immediately every time (no backoff)
+            return Optional.of(0L);
+        }
+
+        // Asynchronous mode (default): use exponential backoff
         int pollCount = taskModel.getPollCount();
         // Assuming pollInterval = 50ms and evaluationOffsetThreshold = 200 this will cause
         // a JOIN task to be evaluated continuously during the first 10 seconds and the FORK/JOIN

core/src/test/java/com/netflix/conductor/core/execution/mapper/JoinTaskMapperTest.java

@@ -29,6 +29,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 
 public class JoinTaskMapperTest {
 
@@ -59,4 +60,37 @@ public void getMappedTasks() {
         assertNotNull(mappedTasks);
         assertEquals(TASK_TYPE_JOIN, mappedTasks.get(0).getTaskType());
     }
+
+    @Test
+    public void getMappedTasksWithJoinMode() {
+
+        WorkflowTask workflowTask = new WorkflowTask();
+        workflowTask.setType(TaskType.JOIN.name());
+        workflowTask.setJoinOn(Arrays.asList("task1", "task2"));
+        workflowTask.setJoinMode(WorkflowTask.JoinMode.SYNC);
+
+        String taskId = new IDGenerator().generate();
+
+        WorkflowDef wd = new WorkflowDef();
+        WorkflowModel workflow = new WorkflowModel();
+        workflow.setWorkflowDefinition(wd);
+
+        TaskMapperContext taskMapperContext =
+                TaskMapperContext.newBuilder()
+                        .withWorkflowModel(workflow)
+                        .withTaskDefinition(new TaskDef())
+                        .withWorkflowTask(workflowTask)
+                        .withRetryCount(0)
+                        .withTaskId(taskId)
+                        .build();
+
+        List<TaskModel> mappedTasks = new JoinTaskMapper().getMappedTasks(taskMapperContext);
+
+        assertNotNull(mappedTasks);
+        assertEquals(TASK_TYPE_JOIN, mappedTasks.get(0).getTaskType());
+        // joinMode is read directly from workflowTask, not injected into input data
+        assertNull(mappedTasks.get(0).getInputData().get("joinMode"));
+        assertEquals(
+                WorkflowTask.JoinMode.SYNC, mappedTasks.get(0).getWorkflowTask().getJoinMode());
+    }
 }

core/src/test/java/com/netflix/conductor/core/execution/tasks/JoinTest.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2022 Conductor Authors.
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package com.netflix.conductor.core.execution.tasks;
+
+import java.util.Optional;
+
+import org.junit.Test;
+
+import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
+import com.netflix.conductor.core.config.ConductorProperties;
+import com.netflix.conductor.model.TaskModel;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class JoinTest {
+
+    private static TaskModel taskWithJoinMode(WorkflowTask.JoinMode mode) {
+        WorkflowTask workflowTask = new WorkflowTask();
+        workflowTask.setJoinMode(mode);
+        TaskModel task = new TaskModel();
+        task.setWorkflowTask(workflowTask);
+        return task;
+    }
+
+    @Test
+    public void testSynchronousJoinModeEvaluationOffset() {
+        ConductorProperties properties = mock(ConductorProperties.class);
+        when(properties.getSystemTaskPostponeThreshold()).thenReturn(200);
+
+        Join join = new Join(properties);
+        TaskModel task = taskWithJoinMode(WorkflowTask.JoinMode.SYNC);
+
+        // Synchronous mode should always return 0 offset
+        task.setPollCount(100);
+        Optional<Long> offset = join.getEvaluationOffset(task, 10000L);
+        assertTrue(offset.isPresent());
+        assertEquals(0L, offset.get().longValue());
+
+        // Even with high poll count, SYNC mode returns 0
+        task.setPollCount(500);
+        offset = join.getEvaluationOffset(task, 10000L);
+        assertTrue(offset.isPresent());
+        assertEquals(0L, offset.get().longValue());
+    }
+
+    @Test
+    public void testAsynchronousJoinModeEvaluationOffset() {
+        ConductorProperties properties = mock(ConductorProperties.class);
+        when(properties.getSystemTaskPostponeThreshold()).thenReturn(200);
+
+        Join join = new Join(properties);
+        TaskModel task = taskWithJoinMode(WorkflowTask.JoinMode.ASYNC);
+
+        // Low poll count should return 0
+        task.setPollCount(100);
+        Optional<Long> offset = join.getEvaluationOffset(task, 10000L);
+        assertTrue(offset.isPresent());
+        assertEquals(0L, offset.get().longValue());
+
+        // High poll count should use exponential backoff
+        task.setPollCount(250);
+        offset = join.getEvaluationOffset(task, 10000L);
+        assertTrue(offset.isPresent());
+        assertTrue(offset.get() > 0L);
+    }
+
+    @Test
+    public void testDefaultAsyncBehavior() {
+        ConductorProperties properties = mock(ConductorProperties.class);
+        when(properties.getSystemTaskPostponeThreshold()).thenReturn(200);
+
+        Join join = new Join(properties);
+
+        // No joinMode on workflowTask — should default to async behavior
+        TaskModel task = new TaskModel();
+        task.setWorkflowTask(new WorkflowTask());
+
+        // Low poll count should return 0
+        task.setPollCount(100);
+        Optional<Long> offset = join.getEvaluationOffset(task, 10000L);
+        assertTrue(offset.isPresent());
+        assertEquals(0L, offset.get().longValue());
+
+        // High poll count should use exponential backoff (default async behavior)
+        task.setPollCount(250);
+        offset = join.getEvaluationOffset(task, 10000L);
+        assertTrue(offset.isPresent());
+        assertTrue(offset.get() > 0L);
+    }
+
+    @Test
+    public void testNullWorkflowTaskDefaultsToAsync() {
+        ConductorProperties properties = mock(ConductorProperties.class);
+        when(properties.getSystemTaskPostponeThreshold()).thenReturn(200);
+
+        Join join = new Join(properties);
+
+        // No workflowTask at all — should default to async behavior
+        TaskModel task = new TaskModel();
+
+        task.setPollCount(250);
+        Optional<Long> offset = join.getEvaluationOffset(task, 10000L);
+        assertTrue(offset.isPresent());
+        assertTrue(offset.get() > 0L);
+    }
+
+    @Test
+    public void testIsAsync() {
+        ConductorProperties properties = mock(ConductorProperties.class);
+        Join join = new Join(properties);
+
+        // isAsync should always return true
+        assertTrue(join.isAsync());
+    }
+}

grpc/src/main/java/com/netflix/conductor/grpc/AbstractProtoMapper.java

@@ -1721,6 +1721,9 @@ public WorkflowTaskPb.WorkflowTask toProto(WorkflowTask from) {
             to.setCacheConfig( toProto( from.getCacheConfig() ) );
         }
         to.setPermissive( from.isPermissive() );
+        if (from.getJoinMode() != null) {
+            to.setJoinMode( toProto( from.getJoinMode() ) );
+        }
         return to.build();
     }
 
@@ -1772,6 +1775,27 @@ public WorkflowTask fromProto(WorkflowTaskPb.WorkflowTask from) {
             to.setCacheConfig( fromProto( from.getCacheConfig() ) );
         }
         to.setPermissive( from.getPermissive() );
+        to.setJoinMode( fromProto( from.getJoinMode() ) );
+        return to;
+    }
+
+    public WorkflowTaskPb.WorkflowTask.JoinMode toProto(WorkflowTask.JoinMode from) {
+        WorkflowTaskPb.WorkflowTask.JoinMode to;
+        switch (from) {
+            case SYNC: to = WorkflowTaskPb.WorkflowTask.JoinMode.SYNC; break;
+            case ASYNC: to = WorkflowTaskPb.WorkflowTask.JoinMode.ASYNC; break;
+            default: throw new IllegalArgumentException("Unexpected enum constant: " + from);
+        }
+        return to;
+    }
+
+    public WorkflowTask.JoinMode fromProto(WorkflowTaskPb.WorkflowTask.JoinMode from) {
+        WorkflowTask.JoinMode to;
+        switch (from) {
+            case SYNC: to = WorkflowTask.JoinMode.SYNC; break;
+            case ASYNC: to = WorkflowTask.JoinMode.ASYNC; break;
+            default: throw new IllegalArgumentException("Unexpected enum constant: " + from);
+        }
         return to;
     }
 

grpc/src/main/proto/model/workflowtask.proto

@@ -11,6 +11,10 @@ option java_outer_classname = "WorkflowTaskPb";
 option go_package = "github.com/netflix/conductor/client/gogrpc/conductor/model";
 
 message WorkflowTask {
+    enum JoinMode {
+        SYNC = 0;
+        ASYNC = 1;
+    }
     message WorkflowTaskList {
         repeated WorkflowTask tasks = 1;
     }
@@ -46,4 +50,5 @@ message WorkflowTask {
     string join_status = 30;
     CacheConfig cache_config = 31;
     bool permissive = 32;
+    WorkflowTask.JoinMode join_mode = 34;
 }