From 6468ef314aeae643861cae0c17693834558252f6 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Thu, 13 Aug 2026 12:37:33 -0600 Subject: [PATCH 01/21] add test against oss on push of pr-to-main --- .github/workflows/ci.yml | 57 ++++++++++ CONTRIBUTING.md | 41 +++++++ .../client/http/SchedulerResource.java | 9 +- scripts/docker-compose-oss.yaml | 28 +++++ scripts/run-integration-oss.sh | 106 ++++++++++++++++++ .../client/ServiceRegistryClientTest.java | 3 + .../conductor/client/WorkflowRetryTest.java | 3 + .../client/http/AuthorizationClientTests.java | 3 + .../client/http/EnvironmentClientTests.java | 3 + .../client/http/EventClientTests.java | 6 +- .../client/http/MetadataClientTests.java | 42 ++++++- .../client/http/PromptClientTests.java | 3 + .../client/http/SchedulerClientTests.java | 5 + .../client/http/SchemaClientTests.java | 3 + .../client/http/SecretClientTests.java | 3 + .../http/ServiceRegistryClientTests.java | 3 + .../client/http/TaskClientTests.java | 34 +++++- .../client/http/TokenClientTest.java | 3 + .../client/http/WorkflowClientTests.java | 5 + .../client/http/WorkflowStateUpdateTests.java | 5 + .../orkes/conductor/client/util/TestUtil.java | 25 +++++ .../orkes/conductor/sdk/WorkflowSDKTests.java | 14 ++- 22 files changed, 388 insertions(+), 16 deletions(-) create mode 100644 scripts/docker-compose-oss.yaml create mode 100755 scripts/run-integration-oss.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ecca05d7a..ef7d836a9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,6 +8,11 @@ on: branches: - main workflow_dispatch: + inputs: + oss_conductor_version: + description: 'OSS Conductor image tag (falls back to E2E_TEST_OSS_CONDUCTOR_VERSION org var)' + required: false + type: string concurrency: group: ${{ github.workflow }}-${{ github.ref }} @@ -153,3 +158,55 @@ jobs: - name: Check Tests Status if: steps.tests.outcome == 'failure' run: exit 1 + + integration-tests-oss: + runs-on: ubuntu-latest + name: Integration Tests (OSS) + timeout-minutes: 30 + env: + CONDUCTOR_SERVER_URL: http://localhost:8080/api + CONDUCTOR_SERVER_TYPE: oss + OSS_CONDUCTOR_VERSION: ${{ inputs.oss_conductor_version || vars.E2E_TEST_OSS_CONDUCTOR_VERSION }} + + steps: + - name: Verify OSS Conductor version is set + run: | + if [ -z "$OSS_CONDUCTOR_VERSION" ]; then + echo "::error::No Conductor OSS image tag resolved. Set the E2E_TEST_OSS_CONDUCTOR_VERSION organization variable (and ensure its repository access policy includes this repo), or pass the oss_conductor_version input via workflow_dispatch." + exit 1 + fi + echo "Using conductoross/conductor:$OSS_CONDUCTOR_VERSION" + + - name: Checkout + uses: actions/checkout@v6 + + - name: Set up Zulu JDK 21 + uses: actions/setup-java@v5 + with: + distribution: "zulu" + java-version: "21" + + - name: Start Conductor OSS stack + run: docker compose -f scripts/docker-compose-oss.yaml up -d + + - name: Wait for Conductor to be healthy + run: timeout 120 bash -c 'until curl -sf http://localhost:8080/health; do sleep 5; done' + + - name: Run integration tests (OSS) + id: integration_tests + continue-on-error: true + run: ./gradlew :tests:test -PIntegrationTests + + - name: Dump Conductor logs + if: failure() || steps.integration_tests.outcome == 'failure' + run: docker compose -f scripts/docker-compose-oss.yaml logs conductor-server + + - name: Publish Test Report + if: always() + uses: mikepenz/action-junit-report@v6 + with: + report_paths: '**/tests/build/test-results/test/TEST-*.xml' + + - name: Check Integration Tests Status + if: steps.integration_tests.outcome == 'failure' + run: exit 1 \ No newline at end of file diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 7bfa1458c..bb0172b6d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -23,6 +23,47 @@ Run the SDK test suite: ./gradlew test jacocoTestReport ``` +### Running the OSS integration suite locally + +The `tests` module also has an integration suite (`-PIntegrationTests`) that runs against a +real Conductor server, separate from the unit suite above. `scripts/run-integration-oss.sh` +mirrors the `integration-tests-oss` job in `ci.yml`: it starts a local Conductor OSS + +Postgres stack (defined in `scripts/docker-compose-oss.yaml`), waits for `/health`, runs the +integration suite, and tears the stack down on exit. + +```shell +scripts/run-integration-oss.sh # against `latest` +scripts/run-integration-oss.sh --version 3.32.0-rc18 +scripts/run-integration-oss.sh --keep-up # leave the stack running afterwards +scripts/run-integration-oss.sh --include-gated # also run tests normally skipped as Orkes-only +``` + +The script always prints the resolved `conductoross/conductor` tag and pulls it before +starting the stack, since `latest` (the local default) is a mutable tag — without an +explicit pull, `docker compose up` would silently reuse a stale cached image instead of +fetching the current one. It also always runs Gradle with `--rerun-tasks`, since the `test` +task's up-to-date check doesn't account for env vars like `CONDUCTOR_SERVER_TYPE` or the +state of the live server underneath — without it, a rerun after changing gating or switching +server versions could silently report a stale cached result instead of executing anything. + +The script doesn't pin a JDK itself, but CI runs on Zulu 21. If your local default JDK is +newer (e.g. 23) you may hit `Unsupported class file major version` errors compiling tests — +set `JAVA_HOME` explicitly to match CI: + +```shell +JAVA_HOME=/Library/Java/JavaVirtualMachines/zulu-21.jdk/Contents/Home ./scripts/run-integration-oss.sh +``` + +Tests annotated `@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = +"oss")` skip themselves against plain OSS because they exercise Orkes-managed-only features +(e.g. the Service Registry, Authorization, Prompts/Integrations, Environment Variables, and +Secrets APIs, plus a handful of task/workflow endpoints OSS doesn't implement or that hit +known Postgres-persistence bugs). Each annotation's `disabledReason` documents the specific, +empirically-confirmed gap — treat those as the source of truth rather than a list here, since +they can drift as OSS gains features. If you add or remove that annotation, re-verify against +a freshly-pulled image first: a test that fails against a stale local image may pass against +current OSS, and vice versa. + Compile the maintained agent examples when changing their APIs or documentation: ```shell diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java b/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java index fe9d7db38..71e0d030b 100644 --- a/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java +++ b/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java @@ -167,14 +167,19 @@ public void resumeSchedule(String name) { /** * Enterprise scheduler endpoints accept GET while OSS accepts PUT. Retry only * a method-not-allowed response so application and authentication failures - * retain their original behavior. + * retain their original behavior. Orkes Enterprise reports this as a proper + * 405; plain OSS Conductor instead reports it as a 500 with a "Request + * method '...' is not supported" message (confirmed empirically) -- treat + * both as a signal to retry with PUT. */ private void executeGetThenPutOnMethodNotAllowed( ConductorClientRequest getRequest, ConductorClientRequest putRequest) { try { client.execute(getRequest); } catch (ConductorClientException e) { - if (e.getStatus() != 405) { + if (e.getStatus() != 405 + && !(e.getStatus() == 500 && e.getMessage() != null + && e.getMessage().contains("is not supported"))) { throw e; } client.execute(putRequest); diff --git a/scripts/docker-compose-oss.yaml b/scripts/docker-compose-oss.yaml new file mode 100644 index 000000000..efc517329 --- /dev/null +++ b/scripts/docker-compose-oss.yaml @@ -0,0 +1,28 @@ +services: + conductor-server: + image: conductoross/conductor:${OSS_CONDUCTOR_VERSION:-latest} + environment: + - CONFIG_PROP=config-postgres.properties + ports: + - "8080:8080" + healthcheck: + test: ["CMD", "curl", "-I", "-XGET", "http://localhost:8080/health"] + interval: 10s + timeout: 10s + retries: 20 + links: + - conductor-postgres:postgresdb + depends_on: + conductor-postgres: + condition: service_healthy + + conductor-postgres: + image: postgres:16 + environment: + - POSTGRES_USER=conductor + - POSTGRES_PASSWORD=conductor + healthcheck: + test: timeout 5 bash -c 'cat < /dev/null > /dev/tcp/localhost/5432' + interval: 5s + timeout: 5s + retries: 12 diff --git a/scripts/run-integration-oss.sh b/scripts/run-integration-oss.sh new file mode 100755 index 000000000..1a17b6d3f --- /dev/null +++ b/scripts/run-integration-oss.sh @@ -0,0 +1,106 @@ +#!/usr/bin/env bash +# +# Spin up a local Conductor OSS stack and run the `tests` module's +# integration suite against it, mirroring the `integration-tests-oss` job in +# .github/workflows/integration-tests-oss.yml. Orkes-Enterprise-only test +# classes are annotated with +# @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss") +# so they skip themselves when it's set (see the individual test files for +# the empirically-confirmed gaps). +# +# The stack (Conductor OSS + Postgres) is defined in +# scripts/docker-compose-oss.yaml and is torn down automatically on exit. The +# image is always pulled before starting, since `latest` (the local default) +# is a mutable tag and a cached copy would otherwise go stale silently. +# +# Usage: +# scripts/run-integration-oss.sh [--keep-up] [--version ] [--include-gated] [-- gradle args] +# Examples: +# scripts/run-integration-oss.sh +# scripts/run-integration-oss.sh --version 3.32.0-rc18 +# scripts/run-integration-oss.sh --keep-up +# scripts/run-integration-oss.sh --include-gated # also run tests normally skipped as Orkes-only +# scripts/run-integration-oss.sh -- --tests "*WorkflowClientTests" +set -euo pipefail + +KEEP_UP=0 +INCLUDE_GATED=0 +extra=() + +while [[ $# -gt 0 ]]; do + case "$1" in + --keep-up) KEEP_UP=1; shift ;; + --version) OSS_CONDUCTOR_VERSION="${2:?--version needs a tag}"; shift 2 ;; + --include-gated) INCLUDE_GATED=1; shift ;; + -h|--help) + echo "Usage: $0 [--keep-up] [--version ] [--include-gated] [-- gradle args]" + exit 0 + ;; + --) shift; extra=("$@"); break ;; + *) echo "Unknown argument: $1" >&2; exit 1 ;; + esac +done + +export OSS_CONDUCTOR_VERSION="${OSS_CONDUCTOR_VERSION:-latest}" + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" +COMPOSE_FILE="${SCRIPT_DIR}/docker-compose-oss.yaml" +cd "${REPO_ROOT}" + +compose() { docker compose -f "${COMPOSE_FILE}" "$@"; } + +cleanup() { + if [[ "${KEEP_UP}" == "1" ]]; then + echo "--keep-up set: leaving the OSS stack running. Tear down with:" + echo " docker compose -f ${COMPOSE_FILE} down -v" + return + fi + echo "Tearing down Conductor OSS stack..." + compose down -v || true +} +trap cleanup EXIT + +echo "Using conductoross/conductor:${OSS_CONDUCTOR_VERSION}" + +# `docker compose up` only pulls an image when it is missing locally, so a +# previously-cached `latest` (or any other mutable tag) would silently be +# reused instead of getting the current version. Pull unconditionally so the +# stack always reflects the tag we just printed. +echo "Pulling conductoross/conductor:${OSS_CONDUCTOR_VERSION} to ensure it's current..." +compose pull conductor-server + +echo "Starting Conductor OSS stack..." +compose up -d + +echo "Waiting for Conductor to be healthy..." +HEALTH_TIMEOUT="${HEALTH_TIMEOUT:-180}" +deadline=$(( SECONDS + HEALTH_TIMEOUT )) +until curl -sf http://localhost:8080/health >/dev/null 2>&1; do + if (( SECONDS >= deadline )); then + echo "Error: Conductor did not become healthy within ${HEALTH_TIMEOUT}s." >&2 + compose logs conductor-server || true + exit 1 + fi + sleep 5 +done +echo "Conductor is up." + +export CONDUCTOR_SERVER_URL="http://localhost:8080/api" + +if [[ "${INCLUDE_GATED}" == "1" ]]; then + echo "--include-gated set: leaving CONDUCTOR_SERVER_TYPE unset, so tests normally" \ + "skipped as Orkes-only will run against OSS too." + unset CONDUCTOR_SERVER_TYPE || true +else + export CONDUCTOR_SERVER_TYPE="oss" +fi + + +# --rerun-tasks: the `test` task's up-to-date check only considers the compiled +# test classpath, not env vars like CONDUCTOR_SERVER_URL/CONDUCTOR_SERVER_TYPE +# or the state of the live server underneath. Without this, Gradle can report +# BUILD SUCCESSFUL while silently reusing a stale cached result from a +# previous run against a different server/tag/gating state instead of +# actually executing anything. +./gradlew :tests:test -PIntegrationTests --rerun-tasks ${extra[@]+"${extra[@]}"} diff --git a/tests/src/test/java/io/orkes/conductor/client/ServiceRegistryClientTest.java b/tests/src/test/java/io/orkes/conductor/client/ServiceRegistryClientTest.java index 76922097e..64fc95d0f 100644 --- a/tests/src/test/java/io/orkes/conductor/client/ServiceRegistryClientTest.java +++ b/tests/src/test/java/io/orkes/conductor/client/ServiceRegistryClientTest.java @@ -19,6 +19,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.common.model.OrkesCircuitBreakerConfig; import com.netflix.conductor.common.model.ServiceMethod; @@ -30,6 +31,8 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; +@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the Service Registry API (/registry/service) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/registry/service')") public class ServiceRegistryClientTest { private static final String PROTO_FILENAME = "compiled.bin"; diff --git a/tests/src/test/java/io/orkes/conductor/client/WorkflowRetryTest.java b/tests/src/test/java/io/orkes/conductor/client/WorkflowRetryTest.java index e3793c077..739367075 100644 --- a/tests/src/test/java/io/orkes/conductor/client/WorkflowRetryTest.java +++ b/tests/src/test/java/io/orkes/conductor/client/WorkflowRetryTest.java @@ -18,6 +18,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.common.metadata.tasks.TaskDef; import com.netflix.conductor.common.metadata.tasks.TaskResult; @@ -36,6 +37,8 @@ import lombok.extern.slf4j.Slf4j; @Slf4j +@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "workflowClient.uploadCompletedWorkflows() (/workflow/document-store/upload) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/workflow/document-store/upload')") public class WorkflowRetryTest { private final OrkesMetadataClient metadataClient; private final OrkesWorkflowClient workflowClient; diff --git a/tests/src/test/java/io/orkes/conductor/client/http/AuthorizationClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/AuthorizationClientTests.java index d40e829bc..fa896a45d 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/AuthorizationClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/AuthorizationClientTests.java @@ -25,6 +25,7 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.client.exception.ConductorClientException; import com.netflix.conductor.common.metadata.workflow.WorkflowDef; @@ -49,6 +50,8 @@ import io.orkes.conductor.client.util.ClientTestUtil; import io.orkes.conductor.client.util.Commons; +@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the Authorization APIs (applications/users/groups/roles/permissions) are not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/applications|users|groups|...')") public class AuthorizationClientTests { private static AuthorizationClient authorizationClient; private static String applicationId; diff --git a/tests/src/test/java/io/orkes/conductor/client/http/EnvironmentClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/EnvironmentClientTests.java index 6dd0e3359..d1de9535b 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/EnvironmentClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/EnvironmentClientTests.java @@ -19,11 +19,14 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import java.util.List; import java.util.Optional; import java.util.UUID; +@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "environment variable writes are not supported by plain OSS Conductor, confirmed empirically: OSS added read-only GET /environment in 3.32.0-rc.9 but PUT /environment/{key} still 405s ('Request method 'PUT' is not supported')") public class EnvironmentClientTests { private static EnvironmentClient envClient; diff --git a/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java index c1e21d90b..47c2d28af 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java @@ -35,7 +35,11 @@ void testEventHandler() { try { eventClient.unregisterEventHandler(EVENT_NAME); } catch (ConductorClientException e) { - if (e.getStatus() != 404) { + // Best-effort cleanup: tolerate "doesn't exist" regardless of how the + // server reports it. Orkes Enterprise returns 404; plain OSS Conductor + // returns a 500 with a "not found" message instead (confirmed + // empirically) -- treat both as success for this purpose. + if (e.getStatus() != 404 && !e.getMessage().contains("not found")) { throw e; } } diff --git a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java index 1e91169d9..837e276ef 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java @@ -16,6 +16,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.client.exception.ConductorClientException; import com.netflix.conductor.common.metadata.tasks.TaskDef; @@ -38,7 +39,11 @@ void taskDefinition() { try { metadataClient.unregisterTaskDef(Commons.TASK_NAME); } catch (ConductorClientException e) { - if (e.getStatus() != 404) { + // Best-effort cleanup: tolerate "doesn't exist" regardless of how the + // server reports it. Orkes Enterprise returns 404; plain OSS Conductor + // returns a 500 with a "No such task definition" message instead + // (confirmed empirically) -- treat both as success for this purpose. + if (e.getStatus() != 404 && !e.getMessage().contains("No such task definition")) { throw e; } } @@ -54,16 +59,41 @@ void workflow() { try { metadataClient.unregisterWorkflowDef(Commons.WORKFLOW_NAME, Commons.WORKFLOW_VERSION); } catch (ConductorClientException e) { - if (e.getStatus() != 404) { + // Best-effort cleanup: tolerate "doesn't exist" regardless of how the + // server reports it. Orkes Enterprise returns 404; plain OSS Conductor + // returns a 500 with a "No such workflow definition" message instead + // (confirmed empirically) -- treat both as success for this purpose. + if (e.getStatus() != 404 && !e.getMessage().contains("No such workflow definition")) { throw e; } } metadataClient.registerTaskDefs(List.of(Commons.getTaskDef())); WorkflowDef workflowDef = WorkflowUtil.getWorkflowDef(); - metadataClient.registerWorkflowDef(workflowDef); + try { + metadataClient.registerWorkflowDef(workflowDef); + } catch (ConductorClientException e) { + // Commons.WORKFLOW_NAME/VERSION is shared fixture data used by several + // test classes in this suite; tolerate an "already exists" collision + // here since the update/overwrite calls below re-establish the + // intended definition regardless of which class registered it first. + if (e.getStatus() != 500 || !e.getMessage().contains("already exists")) { + throw e; + } + } metadataClient.updateWorkflowDefs(List.of(workflowDef)); metadataClient.updateWorkflowDefs(List.of(workflowDef), true); - metadataClient.registerWorkflowDef(workflowDef, true); + try { + metadataClient.registerWorkflowDef(workflowDef, true); + } catch (ConductorClientException e) { + // The overwrite=true query param on POST /metadata/workflow is not + // honored by plain OSS Conductor, confirmed empirically (it still + // rejects an existing name+version instead of overwriting); the + // updateWorkflowDefs(..., true) call above already re-established + // the intended definition. + if (e.getStatus() != 500 || !e.getMessage().contains("already exists")) { + throw e; + } + } ((OrkesMetadataClient) metadataClient) .getWorkflowDefWithMetadata(Commons.WORKFLOW_NAME, Commons.WORKFLOW_VERSION); WorkflowDef receivedWorkflowDef = metadataClient.getWorkflowDef(Commons.WORKFLOW_NAME, @@ -73,6 +103,8 @@ void workflow() { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "task tagging (/metadata/task/{name}/tags) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/metadata/task/{name}/tags')") void tagTask() throws Exception { metadataClient.registerTaskDefs(List.of(Commons.getTaskDef())); try { @@ -98,6 +130,8 @@ void tagTask() throws Exception { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "workflow tagging (/metadata/workflow/{name}/tags) is not implemented by plain OSS Conductor, confirmed empirically (the {version} path segment ends up matching the literal string \"tags\" instead, a server-side routing collision)") void tagWorkflow() { TagObject tagObject = Commons.getTagObject(); try { diff --git a/tests/src/test/java/io/orkes/conductor/client/http/PromptClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/PromptClientTests.java index 53d427232..2ce260590 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/PromptClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/PromptClientTests.java @@ -20,6 +20,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.client.exception.ConductorClientException; @@ -32,6 +33,8 @@ import org.conductoross.conductor.client.model.ai.PromptTemplate; import io.orkes.conductor.client.util.ClientTestUtil; +@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the Prompts and Integrations APIs (/prompts, /integrations) are not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/prompts|integrations/...')") public class PromptClientTests { private static final String PROMPT_NAME = "test-sdk-java-prompt"; private static final String PROMPT_DESCRIPTION = "Test prompt for Java SDK"; diff --git a/tests/src/test/java/io/orkes/conductor/client/http/SchedulerClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/SchedulerClientTests.java index f86c332d2..9bbc030a7 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/SchedulerClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/SchedulerClientTests.java @@ -16,6 +16,7 @@ import java.util.UUID; import org.junit.jupiter.api.*; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.common.model.BulkResponse; @@ -51,6 +52,10 @@ void afterEach() { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "GET /scheduler/search does not 404 on plain OSS Conductor, but confirmed empirically to " + + "always return zero results (even after polling for 30s) -- schedules aren't surfaced via search " + + "on plain OSS the way they are on Orkes Enterprise") void testMethods() { schedulerClient.deleteSchedule(SCHEDULE_1); Assertions.assertTrue(schedulerClient.getNextFewSchedules(CRON_EXPRESSION_1, 0L, 0L, 0).isEmpty()); diff --git a/tests/src/test/java/io/orkes/conductor/client/http/SchemaClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/SchemaClientTests.java index b74b318fc..8dd145160 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/SchemaClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/SchemaClientTests.java @@ -17,6 +17,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.client.exception.ConductorClientException; import com.netflix.conductor.common.metadata.SchemaDef; @@ -24,6 +25,8 @@ import io.orkes.conductor.client.SchemaClient; import io.orkes.conductor.client.util.ClientTestUtil; +@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the Schema API (/schema) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/schema')") public class SchemaClientTests { private static final String SCHEMA_NAME = "test-sdk-java-schema"; diff --git a/tests/src/test/java/io/orkes/conductor/client/http/SecretClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/SecretClientTests.java index 1a8daecf8..33ce31ee7 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/SecretClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/SecretClientTests.java @@ -16,6 +16,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.client.exception.ConductorClientException; @@ -24,6 +25,8 @@ import io.orkes.conductor.client.util.ClientTestUtil; +@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "secret writes are not supported by plain OSS Conductor, confirmed empirically: its env-var-backed secrets DAO is read-only, so putSecret() 501s ('env-backed secrets are read-only')") public class SecretClientTests { private final String SECRET_NAME = "test-sdk-java-secret_name"; private final String SECRET_KEY = "test-sdk-java-secret_key"; diff --git a/tests/src/test/java/io/orkes/conductor/client/http/ServiceRegistryClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/ServiceRegistryClientTests.java index d5253fd8a..a09ba6c85 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/ServiceRegistryClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/ServiceRegistryClientTests.java @@ -18,6 +18,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.client.exception.ConductorClientException; import com.netflix.conductor.common.model.CircuitBreakerTransitionResponse; @@ -32,6 +33,8 @@ import static org.junit.jupiter.api.Assertions.assertNull; +@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the Service Registry API (/registry/service) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/registry/service')") public class ServiceRegistryClientTests { private static final String SERVICE_NAME = "test-sdk-java-service"; private static final String SERVICE_URI = "localhost:50051"; diff --git a/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java index 22f9098e7..19abbc1ae 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java @@ -30,6 +30,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable; import org.testcontainers.shaded.com.google.common.util.concurrent.Uninterruptibles; @@ -146,6 +147,8 @@ public void testUpdateByRefName() { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the sync task-update endpoint (POST /tasks/{workflowId}/{taskRefName}/{status}/sync) never returns the updated workflow on plain OSS Conductor, confirmed empirically (caller times out waiting for terminal status)") public void testUpdateByRefNameSync() { StartWorkflowRequest request = new StartWorkflowRequest(); request.setName(workflowName); @@ -332,6 +335,8 @@ private void completeWorkflow(String workflowId) throws Exception { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically") void testSyncTargetWorkflow() throws Exception { String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.TARGET_WORKFLOW); @@ -346,6 +351,8 @@ void testSyncTargetWorkflow() throws Exception { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically") void testSyncBlockingWorkflow() throws Exception { String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.BLOCKING_WORKFLOW); @@ -360,6 +367,8 @@ void testSyncBlockingWorkflow() throws Exception { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically") void testSyncBlockingTask() throws Exception { String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.BLOCKING_TASK); @@ -391,6 +400,8 @@ void testSyncBlockingTaskList() throws Exception { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically") void testSyncBlockingTaskInput() throws Exception { String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.BLOCKING_TASK_INPUT); @@ -408,7 +419,10 @@ void testSyncBlockingTaskInput() throws Exception { private static final String REGION_DURABLE_ENABLED = "CONDUCTOR_REGION_DURABLE_ENABLED"; @Test - @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true") + @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true", + disabledReason = "target server has no region replication configured") + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically") void testDurableTargetWorkflow() throws Exception { String workflowId = startComplexWorkflow(Consistency.REGION_DURABLE, ReturnStrategy.TARGET_WORKFLOW); @@ -423,7 +437,10 @@ void testDurableTargetWorkflow() throws Exception { } @Test - @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true") + @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true", + disabledReason = "target server has no region replication configured") + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically") void testDurableBlockingWorkflow() throws Exception { String workflowId = startComplexWorkflow(Consistency.REGION_DURABLE, ReturnStrategy.BLOCKING_WORKFLOW); @@ -438,6 +455,8 @@ void testDurableBlockingWorkflow() throws Exception { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically") void testDurableBlockingTask() throws Exception { String workflowId = startComplexWorkflow(Consistency.DURABLE, ReturnStrategy.BLOCKING_TASK); @@ -452,7 +471,10 @@ void testDurableBlockingTask() throws Exception { } @Test - @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true") + @EnabledIfEnvironmentVariable(named = REGION_DURABLE_ENABLED, matches = "true", + disabledReason = "target server has no region replication configured") + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically") void testDurableBlockingTaskInput() throws Exception { String workflowId = startComplexWorkflow(Consistency.REGION_DURABLE, ReturnStrategy.BLOCKING_TASK_INPUT); @@ -467,6 +489,8 @@ void testDurableBlockingTaskInput() throws Exception { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the SYNCHRONOUS/REGION_DURABLE consistency + BLOCKING_* return-strategy signal API (POST /tasks/{workflowId}/{status}/signal/sync) is not implemented by plain OSS Conductor, confirmed empirically") void testDefaultReturnStrategy() throws Exception { String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.TARGET_WORKFLOW); @@ -745,6 +769,8 @@ void testRequeuePendingTasksByTaskType() { // ==================== Search Tests ==================== @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "GET /tasks/search fails on plain OSS Conductor with a Postgres persistence layer, confirmed empirically (ERROR: column \"workflow_id\" does not exist)") void testSearchTasks() { StartWorkflowRequest request = new StartWorkflowRequest(); request.setName(workflowName); @@ -781,6 +807,8 @@ void testSearchV2Tasks() { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "GET /tasks/search fails on plain OSS Conductor with a Postgres persistence layer, confirmed empirically (ERROR: column \"workflow_id\" does not exist)") void testPaginatedSearchTasks() { StartWorkflowRequest request = new StartWorkflowRequest(); request.setName(workflowName); diff --git a/tests/src/test/java/io/orkes/conductor/client/http/TokenClientTest.java b/tests/src/test/java/io/orkes/conductor/client/http/TokenClientTest.java index 06b2f25dc..7f6ed98db 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/TokenClientTest.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/TokenClientTest.java @@ -16,11 +16,14 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import io.orkes.conductor.client.model.GenerateTokenRequest; import io.orkes.conductor.client.model.TokenResponse; import io.orkes.conductor.client.util.ClientTestUtil; +@DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "the auth Token API (/token) is not implemented by plain OSS Conductor (which has no authentication layer), confirmed empirically (404 'No static resource api/token')") public class TokenClientTest { public static OrkesTokenClient tokenClient; diff --git a/tests/src/test/java/io/orkes/conductor/client/http/WorkflowClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/WorkflowClientTests.java index a8d906940..c4c5e8e13 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/WorkflowClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/WorkflowClientTests.java @@ -22,6 +22,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.common.metadata.tasks.TaskDef; import com.netflix.conductor.common.metadata.tasks.TaskResult; @@ -102,6 +103,8 @@ public void startWorkflow() { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "batch correlation-id search (POST /workflow/correlated/batch) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/workflow/correlated/batch')") public void testSearchByCorrelationIds() { List correlationIds = new ArrayList<>(); Set workflowNames = new HashSet<>(); @@ -188,6 +191,8 @@ public void testSkipTaskFromWorkflow() throws Exception { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "POST /workflow/{workflowId}/variables is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/workflow/{id}/variables')") public void testUpdateVariables() { ConductorWorkflow workflow = new ConductorWorkflow<>(workflowExecutor); workflow.add(new SimpleTask("simple_task", "simple_task_ref")); diff --git a/tests/src/test/java/io/orkes/conductor/client/http/WorkflowStateUpdateTests.java b/tests/src/test/java/io/orkes/conductor/client/http/WorkflowStateUpdateTests.java index 286d428a5..0faad3086 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/WorkflowStateUpdateTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/WorkflowStateUpdateTests.java @@ -22,6 +22,7 @@ import org.conductoross.conductor.common.model.WorkflowRun; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable; import com.netflix.conductor.client.exception.ConductorClientException; import com.netflix.conductor.common.metadata.tasks.Task; @@ -91,6 +92,8 @@ public String startWorkflow() { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "POST /workflow/{workflowId}/state (updateWorkflow) is not implemented by plain OSS Conductor, confirmed empirically (404 'No static resource api/workflow/{id}/state')") public void test() { String workflowId = startWorkflow(); System.out.println(workflowId); @@ -135,6 +138,8 @@ public void test() { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "workflow start idempotency keys are not honored by plain OSS Conductor, confirmed empirically (RETURN_EXISTING starts a brand-new run instead of returning the original workflowId)") public void testIdempotency() { StartWorkflowRequest startWorkflowRequest = new StartWorkflowRequest(); startWorkflowRequest.setName("sync_task_variable_updates"); diff --git a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java index 65b13f72e..a0c4025b1 100644 --- a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java +++ b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java @@ -18,6 +18,7 @@ import java.time.Duration; import java.util.concurrent.Callable; import java.util.concurrent.TimeoutException; +import java.util.function.Predicate; import com.netflix.conductor.common.config.ObjectMapperProvider; import com.netflix.conductor.common.metadata.workflow.WorkflowDef; @@ -160,4 +161,28 @@ private static boolean isTerminalFailure(Workflow workflow) { return workflow.getStatus() == Workflow.WorkflowStatus.FAILED || workflow.getStatus() == Workflow.WorkflowStatus.TERMINATED; } + + /** + * Repeatedly invokes {@code supplier} until {@code condition} accepts its result, or the + * time budget is exhausted, sleeping {@code pollIntervalMs} between attempts. Useful for + * assertions against eventually-consistent state (e.g. search-index writes) instead of a + * single point-in-time check. + * + * @return the first result accepted by {@code condition} + * @throws TimeoutException if no result satisfies {@code condition} within maxWaitTimeMs + */ + public static T waitUntil(Callable supplier, Predicate condition, + long maxWaitTimeMs, long pollIntervalMs) throws Exception { + long endTime = System.currentTimeMillis() + maxWaitTimeMs; + T last = supplier.call(); + while (!condition.test(last)) { + if (System.currentTimeMillis() >= endTime) { + throw new TimeoutException( + String.format("Condition not met within %dms. Last value: %s", maxWaitTimeMs, last)); + } + Thread.sleep(pollIntervalMs); + last = supplier.call(); + } + return last; + } } diff --git a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java index 452da6872..883855e31 100644 --- a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java +++ b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java @@ -14,9 +14,6 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -33,12 +30,13 @@ import com.netflix.conductor.sdk.workflow.task.WorkerTask; import io.orkes.conductor.client.util.ClientTestUtil; +import io.orkes.conductor.client.util.TestUtil; public class WorkflowSDKTests { @Test - public void testCreateWorkflow() { + public void testCreateWorkflow() throws Exception { ConductorClient client = ClientTestUtil.getClient(); AnnotatedWorkerExecutor workerExecutor = new AnnotatedWorkerExecutor(new TaskClient(client), new WorkerConfiguration()); @@ -57,10 +55,14 @@ public void testCreateWorkflow() { CompletableFuture result = workflow.execute(Map.of("name", "orkes")); Assertions.assertNotNull(result); try { - Workflow executedWorkflow = result.get(3, TimeUnit.SECONDS); + // Poll with a time budget instead of a single point-in-time get(): worker + // registration + polling + task execution can take longer than a couple of + // seconds under load (e.g. running alongside the rest of the integration suite). + TestUtil.waitUntil(result::isDone, Boolean::booleanValue, 30_000, 3_000); + Workflow executedWorkflow = result.get(); Assertions.assertNotNull(executedWorkflow); Assertions.assertEquals(Workflow.WorkflowStatus.COMPLETED, executedWorkflow.getStatus()); - } catch (InterruptedException | ExecutionException | TimeoutException e) { + } catch (Exception e) { Assertions.fail(e.getMessage()); } } From 867f988c1c90e2fd6bd2bacc197898ca7cdee89a Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Thu, 13 Aug 2026 12:50:00 -0600 Subject: [PATCH 02/21] give permission to update test report --- .github/workflows/ci.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ef7d836a9..87a16f1e7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,6 +18,10 @@ concurrency: group: ${{ github.workflow }}-${{ github.ref }} cancel-in-progress: true +permissions: + contents: read + checks: write + jobs: documentation-validation: runs-on: ubuntu-latest From 9bcfcfc466f5029bbc1dc78ef733f426dcd12e0b Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 09:34:09 -0600 Subject: [PATCH 03/21] increase timeout on a test that fails --- .github/workflows/ci.yml | 8 ++++++-- .../java/io/orkes/conductor/sdk/WorkflowSDKTests.java | 5 +++-- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 87a16f1e7..1a3317137 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -161,7 +161,9 @@ jobs: - name: Check Tests Status if: steps.tests.outcome == 'failure' - run: exit 1 + run: | + echo "::error::Tests failed. See the 'Run Tests' step above for the Gradle/test output, and the 'Publish Test Report' step's JUnit summary for which test(s) failed." + exit 1 integration-tests-oss: runs-on: ubuntu-latest @@ -213,4 +215,6 @@ jobs: - name: Check Integration Tests Status if: steps.integration_tests.outcome == 'failure' - run: exit 1 \ No newline at end of file + run: | + echo "::error::Integration tests (OSS) failed. See the 'Run integration tests (OSS)' step above for the Gradle/test output, the 'Dump Conductor logs' step for server-side logs, and the 'Publish Test Report' step's JUnit summary for which test(s) failed." + exit 1 \ No newline at end of file diff --git a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java index 883855e31..0686fda89 100644 --- a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java +++ b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java @@ -57,8 +57,9 @@ public void testCreateWorkflow() throws Exception { try { // Poll with a time budget instead of a single point-in-time get(): worker // registration + polling + task execution can take longer than a couple of - // seconds under load (e.g. running alongside the rest of the integration suite). - TestUtil.waitUntil(result::isDone, Boolean::booleanValue, 30_000, 3_000); + // seconds under load (e.g. running alongside the rest of the integration suite, + // or on a shared/slower CI runner -- 30s was observed to be marginal in CI). + TestUtil.waitUntil(result::isDone, Boolean::booleanValue, 60_000, 3_000); Workflow executedWorkflow = result.get(); Assertions.assertNotNull(executedWorkflow); Assertions.assertEquals(Workflow.WorkflowStatus.COMPLETED, executedWorkflow.getStatus()); From d7998b196e7e484d95e9105d5d15b68db74c3cb0 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 10:03:32 -0600 Subject: [PATCH 04/21] add helper for test tolernace of varying behavior between oss and enterprise for a few endpoints, but the 404 vs the oss-way explicitly --- .../client/http/EventClientTests.java | 11 +++--- .../client/http/MetadataClientTests.java | 36 ++++++------------- .../orkes/conductor/client/util/TestUtil.java | 28 +++++++++++++++ 3 files changed, 42 insertions(+), 33 deletions(-) diff --git a/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java index 47c2d28af..bbfdb195e 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java @@ -24,6 +24,7 @@ import io.orkes.conductor.client.util.ClientTestUtil; import io.orkes.conductor.client.util.Commons; +import io.orkes.conductor.client.util.TestUtil; public class EventClientTests { private static final String EVENT_NAME = "test_sdk_java_event_name"; @@ -35,13 +36,9 @@ void testEventHandler() { try { eventClient.unregisterEventHandler(EVENT_NAME); } catch (ConductorClientException e) { - // Best-effort cleanup: tolerate "doesn't exist" regardless of how the - // server reports it. Orkes Enterprise returns 404; plain OSS Conductor - // returns a 500 with a "not found" message instead (confirmed - // empirically) -- treat both as success for this purpose. - if (e.getStatus() != 404 && !e.getMessage().contains("not found")) { - throw e; - } + // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server + // we're running against actually reports it. + TestUtil.assertNotFoundOrRethrow(e, "not found"); } EventHandler eventHandler = getEventHandler(); eventClient.registerEventHandler(eventHandler); diff --git a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java index 837e276ef..bfe82d8d0 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java @@ -39,13 +39,9 @@ void taskDefinition() { try { metadataClient.unregisterTaskDef(Commons.TASK_NAME); } catch (ConductorClientException e) { - // Best-effort cleanup: tolerate "doesn't exist" regardless of how the - // server reports it. Orkes Enterprise returns 404; plain OSS Conductor - // returns a 500 with a "No such task definition" message instead - // (confirmed empirically) -- treat both as success for this purpose. - if (e.getStatus() != 404 && !e.getMessage().contains("No such task definition")) { - throw e; - } + // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server + // we're running against actually reports it. + TestUtil.assertNotFoundOrRethrow(e, "No such task definition"); } TaskDef taskDef = Commons.getTaskDef(); metadataClient.registerTaskDefs(List.of(taskDef)); @@ -59,13 +55,9 @@ void workflow() { try { metadataClient.unregisterWorkflowDef(Commons.WORKFLOW_NAME, Commons.WORKFLOW_VERSION); } catch (ConductorClientException e) { - // Best-effort cleanup: tolerate "doesn't exist" regardless of how the - // server reports it. Orkes Enterprise returns 404; plain OSS Conductor - // returns a 500 with a "No such workflow definition" message instead - // (confirmed empirically) -- treat both as success for this purpose. - if (e.getStatus() != 404 && !e.getMessage().contains("No such workflow definition")) { - throw e; - } + // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server + // we're running against actually reports it. + TestUtil.assertNotFoundOrRethrow(e, "No such workflow definition"); } metadataClient.registerTaskDefs(List.of(Commons.getTaskDef())); WorkflowDef workflowDef = WorkflowUtil.getWorkflowDef(); @@ -82,18 +74,10 @@ void workflow() { } metadataClient.updateWorkflowDefs(List.of(workflowDef)); metadataClient.updateWorkflowDefs(List.of(workflowDef), true); - try { - metadataClient.registerWorkflowDef(workflowDef, true); - } catch (ConductorClientException e) { - // The overwrite=true query param on POST /metadata/workflow is not - // honored by plain OSS Conductor, confirmed empirically (it still - // rejects an existing name+version instead of overwriting); the - // updateWorkflowDefs(..., true) call above already re-established - // the intended definition. - if (e.getStatus() != 500 || !e.getMessage().contains("already exists")) { - throw e; - } - } + // Both Orkes Enterprise and plain OSS Conductor honor overwrite=true on an existing + // name+version and succeed outright (verified empirically against a freshly-pulled + // OSS image; an earlier assumption that OSS rejected this with a 500 no longer holds). + metadataClient.registerWorkflowDef(workflowDef, true); ((OrkesMetadataClient) metadataClient) .getWorkflowDefWithMetadata(Commons.WORKFLOW_NAME, Commons.WORKFLOW_VERSION); WorkflowDef receivedWorkflowDef = metadataClient.getWorkflowDef(Commons.WORKFLOW_NAME, diff --git a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java index a0c4025b1..c52712fb1 100644 --- a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java +++ b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java @@ -20,6 +20,7 @@ import java.util.concurrent.TimeoutException; import java.util.function.Predicate; +import com.netflix.conductor.client.exception.ConductorClientException; import com.netflix.conductor.common.config.ObjectMapperProvider; import com.netflix.conductor.common.metadata.workflow.WorkflowDef; import com.netflix.conductor.common.run.Workflow; @@ -185,4 +186,31 @@ public static T waitUntil(Callable supplier, Predicate condition, } return last; } + + /** + * Whether the suite is currently running against plain OSS Conductor rather than Orkes + * Enterprise, per the same {@code CONDUCTOR_SERVER_TYPE} signal that + * {@code @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss")} + * checks for test gating. + */ + public static boolean isOssServer() { + return "oss".equals(System.getenv("CONDUCTOR_SERVER_TYPE")); + } + + /** + * Asserts a caught exception represents "resource doesn't exist", in the shape specific to + * whichever server type {@code CONDUCTOR_SERVER_TYPE} says we're running against: Orkes + * Enterprise reports a proper 404; plain OSS Conductor instead reports a 500 whose message + * contains {@code ossMessageSubstring} (empirically confirmed per endpoint). Anything else + * is rethrown, since it isn't the "doesn't exist" case this is meant to tolerate. + */ + public static void assertNotFoundOrRethrow(ConductorClientException e, String ossMessageSubstring) { + if (isOssServer()) { + if (e.getStatus() != 500 || e.getMessage() == null || !e.getMessage().contains(ossMessageSubstring)) { + throw e; + } + } else if (e.getStatus() != 404) { + throw e; + } + } } From b602cec3d79926ad1012b000132fd61de99bf1c3 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 10:21:08 -0600 Subject: [PATCH 05/21] remove unneeded oss-vs-orkes tolerance --- .../conductor/client/http/MetadataClientTests.java | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java index bfe82d8d0..5210001b1 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java @@ -61,17 +61,7 @@ void workflow() { } metadataClient.registerTaskDefs(List.of(Commons.getTaskDef())); WorkflowDef workflowDef = WorkflowUtil.getWorkflowDef(); - try { - metadataClient.registerWorkflowDef(workflowDef); - } catch (ConductorClientException e) { - // Commons.WORKFLOW_NAME/VERSION is shared fixture data used by several - // test classes in this suite; tolerate an "already exists" collision - // here since the update/overwrite calls below re-establish the - // intended definition regardless of which class registered it first. - if (e.getStatus() != 500 || !e.getMessage().contains("already exists")) { - throw e; - } - } + metadataClient.registerWorkflowDef(workflowDef); metadataClient.updateWorkflowDefs(List.of(workflowDef)); metadataClient.updateWorkflowDefs(List.of(workflowDef), true); // Both Orkes Enterprise and plain OSS Conductor honor overwrite=true on an existing From 981adfef3fc93a7abfb0cf0401ba1ef28081bfcb Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 12:23:58 -0600 Subject: [PATCH 06/21] remove unnecessary retry utility, add a try/except to a thing that is scheduleAtFixedRate'd --- .../workflow/executor/WorkflowExecutor.java | 14 ++++++++--- .../orkes/conductor/client/util/TestUtil.java | 25 ------------------- .../orkes/conductor/sdk/WorkflowSDKTests.java | 10 +++----- 3 files changed, 13 insertions(+), 36 deletions(-) diff --git a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java index 33654d78d..b88d85150 100644 --- a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java +++ b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java @@ -174,10 +174,16 @@ private void initMonitor() { for (Map.Entry> entry : runningWorkflowFutures.entrySet()) { String workflowId = entry.getKey(); CompletableFuture future = entry.getValue(); - Workflow workflow = workflowClient.getWorkflow(workflowId, true); - if (workflow.getStatus().isTerminal()) { - future.complete(workflow); - runningWorkflowFutures.remove(workflowId); + try { + Workflow workflow = workflowClient.getWorkflow(workflowId, true); + if (workflow.getStatus().isTerminal()) { + future.complete(workflow); + runningWorkflowFutures.remove(workflowId); + } + } catch (Exception e) { + // scheduleAtFixedRate silently kills all future ticks on any uncaught exception, so catch here instead of letting one transient error stop completion-tracking forever. + LOGGER.warn("Error polling workflow {} for completion; will retry on " + + "the next tick", workflowId, e); } } }, diff --git a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java index c52712fb1..6827f61ad 100644 --- a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java +++ b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java @@ -18,7 +18,6 @@ import java.time.Duration; import java.util.concurrent.Callable; import java.util.concurrent.TimeoutException; -import java.util.function.Predicate; import com.netflix.conductor.client.exception.ConductorClientException; import com.netflix.conductor.common.config.ObjectMapperProvider; @@ -163,30 +162,6 @@ private static boolean isTerminalFailure(Workflow workflow) { || workflow.getStatus() == Workflow.WorkflowStatus.TERMINATED; } - /** - * Repeatedly invokes {@code supplier} until {@code condition} accepts its result, or the - * time budget is exhausted, sleeping {@code pollIntervalMs} between attempts. Useful for - * assertions against eventually-consistent state (e.g. search-index writes) instead of a - * single point-in-time check. - * - * @return the first result accepted by {@code condition} - * @throws TimeoutException if no result satisfies {@code condition} within maxWaitTimeMs - */ - public static T waitUntil(Callable supplier, Predicate condition, - long maxWaitTimeMs, long pollIntervalMs) throws Exception { - long endTime = System.currentTimeMillis() + maxWaitTimeMs; - T last = supplier.call(); - while (!condition.test(last)) { - if (System.currentTimeMillis() >= endTime) { - throw new TimeoutException( - String.format("Condition not met within %dms. Last value: %s", maxWaitTimeMs, last)); - } - Thread.sleep(pollIntervalMs); - last = supplier.call(); - } - return last; - } - /** * Whether the suite is currently running against plain OSS Conductor rather than Orkes * Enterprise, per the same {@code CONDUCTOR_SERVER_TYPE} signal that diff --git a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java index 0686fda89..a9091edc1 100644 --- a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java +++ b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java @@ -14,6 +14,7 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -30,7 +31,6 @@ import com.netflix.conductor.sdk.workflow.task.WorkerTask; import io.orkes.conductor.client.util.ClientTestUtil; -import io.orkes.conductor.client.util.TestUtil; public class WorkflowSDKTests { @@ -55,12 +55,8 @@ public void testCreateWorkflow() throws Exception { CompletableFuture result = workflow.execute(Map.of("name", "orkes")); Assertions.assertNotNull(result); try { - // Poll with a time budget instead of a single point-in-time get(): worker - // registration + polling + task execution can take longer than a couple of - // seconds under load (e.g. running alongside the rest of the integration suite, - // or on a shared/slower CI runner -- 30s was observed to be marginal in CI). - TestUtil.waitUntil(result::isDone, Boolean::booleanValue, 60_000, 3_000); - Workflow executedWorkflow = result.get(); + // WorkflowExecutor's monitor thread polls every 100ms (see initMonitor()), so 10s is a generous margin. + Workflow executedWorkflow = result.get(10, TimeUnit.SECONDS); Assertions.assertNotNull(executedWorkflow); Assertions.assertEquals(Workflow.WorkflowStatus.COMPLETED, executedWorkflow.getStatus()); } catch (Exception e) { From 7eb28c74fb94445d1ac2aea45af2e0e3428eb0a2 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 12:41:39 -0600 Subject: [PATCH 07/21] restore schedulerresource to prior state --- .../orkes/conductor/client/http/SchedulerResource.java | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java b/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java index 71e0d030b..fe9d7db38 100644 --- a/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java +++ b/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java @@ -167,19 +167,14 @@ public void resumeSchedule(String name) { /** * Enterprise scheduler endpoints accept GET while OSS accepts PUT. Retry only * a method-not-allowed response so application and authentication failures - * retain their original behavior. Orkes Enterprise reports this as a proper - * 405; plain OSS Conductor instead reports it as a 500 with a "Request - * method '...' is not supported" message (confirmed empirically) -- treat - * both as a signal to retry with PUT. + * retain their original behavior. */ private void executeGetThenPutOnMethodNotAllowed( ConductorClientRequest getRequest, ConductorClientRequest putRequest) { try { client.execute(getRequest); } catch (ConductorClientException e) { - if (e.getStatus() != 405 - && !(e.getStatus() == 500 && e.getMessage() != null - && e.getMessage().contains("is not supported"))) { + if (e.getStatus() != 405) { throw e; } client.execute(putRequest); From 88f6807802990126a881f2866fc245d16ccdf811 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 12:48:44 -0600 Subject: [PATCH 08/21] remove unnecessary comments --- .../io/orkes/conductor/client/http/MetadataClientTests.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java index 5210001b1..d7430a97b 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java @@ -64,9 +64,6 @@ void workflow() { metadataClient.registerWorkflowDef(workflowDef); metadataClient.updateWorkflowDefs(List.of(workflowDef)); metadataClient.updateWorkflowDefs(List.of(workflowDef), true); - // Both Orkes Enterprise and plain OSS Conductor honor overwrite=true on an existing - // name+version and succeed outright (verified empirically against a freshly-pulled - // OSS image; an earlier assumption that OSS rejected this with a 500 no longer holds). metadataClient.registerWorkflowDef(workflowDef, true); ((OrkesMetadataClient) metadataClient) .getWorkflowDefWithMetadata(Commons.WORKFLOW_NAME, Commons.WORKFLOW_VERSION); From d78c219c61da88832570f298faee90de1be25fe3 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Mon, 17 Aug 2026 09:38:20 -0600 Subject: [PATCH 09/21] attempt to improve flaky test by removing a redundant call to startPolling --- .../src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java index a9091edc1..25d5d303f 100644 --- a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java +++ b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java @@ -40,8 +40,8 @@ public void testCreateWorkflow() throws Exception { ConductorClient client = ClientTestUtil.getClient(); AnnotatedWorkerExecutor workerExecutor = new AnnotatedWorkerExecutor(new TaskClient(client), new WorkerConfiguration()); + // initWorkers() already starts polling; a redundant extra startPolling() call here used to race with AnnotatedWorkerExecutor's own double-start (likely a real bug there) and could drop the first polled task. workerExecutor.initWorkers("io.orkes.conductor.sdk"); - workerExecutor.startPolling(); WorkflowExecutor executor = new WorkflowExecutor(client, workerExecutor); From 4076f65b27cfa7e014ba39ab83befa35004cb56d Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Mon, 17 Aug 2026 10:10:46 -0600 Subject: [PATCH 10/21] testing removal of redundant call to startPolling --- .../workflow/executor/task/AnnotatedWorkerExecutor.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java index e2ed6d107..349a38ba9 100644 --- a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java +++ b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java @@ -76,8 +76,9 @@ public AnnotatedWorkerExecutor(TaskClient taskClient, WorkerConfiguration worker * implementation */ public synchronized void initWorkers(String... basePackages) { + // scanWorkers() -> initWorkersFromClasses() -> initWorkersFromInstances() already calls startPolling(); + // an extra call here used to race with that one and could drop a task polled by the runner it replaces. scanWorkers(basePackages); - startPolling(); } public synchronized void initWorkersFromInstances(List workerInstances) { @@ -157,7 +158,10 @@ private void scanWorkers(String... basePackages) { initWorkersFromClasses(classes); } catch (Exception e) { - LOGGER.error("Error while scanning for workers: ", e); + // Rethrow (unchecked) rather than swallow: initWorkers() no longer has its own startPolling() + // fallback, so a swallowed failure here would otherwise leave the caller believing workers are + // running when none were ever started. + throw new RuntimeException("Error while scanning for workers", e); } } From be18d53dba2ada95d59087e53a5f1f71a4e3a97b Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 11:41:07 -0600 Subject: [PATCH 11/21] make startPolling idempotent instead of dropping it from initWorkers startPolling() builds a new TaskRunnerConfigurer, init()s it, and only then shuts the previous one down, so calling it twice leaves two runners polling the same task types. A task leased by the outgoing runner can be left in-progress until its response timeout expires -- the cause of the WorkflowSDKTests flakiness against OSS (that run logged three startPolling invocations). Removing the call from initWorkers() only fixed callers that don't also call startPolling() themselves, which the docs and examples/old/.../taskdomains/Main both do, and it forced scanWorkers() to rethrow so a scan failure wouldn't silently leave nothing polling. Guarding inside startPolling() instead fixes every caller shape and leaves initWorkers()'s contract alone, so scanWorkers() goes back to logging. startPolling() is now synchronized, matching the init* methods that call it, so the worker-set flag it reads is not raced. Co-Authored-By: Claude Opus 5 (1M context) --- .../task/AnnotatedWorkerExecutor.java | 39 +++++++++++--- .../executor/task/AnnotatedWorkerTests.java | 54 +++++++++++++++++++ 2 files changed, 86 insertions(+), 7 deletions(-) diff --git a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java index 349a38ba9..1894ce162 100644 --- a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java +++ b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java @@ -52,6 +52,12 @@ public class AnnotatedWorkerExecutor { private final Set scannedPackages = new HashSet<>(); + /** + * Set whenever a worker is added, cleared whenever {@link #startPolling()} builds a task runner. + * Lets startPolling() distinguish a real (re)start from a redundant duplicate call. + */ + private boolean workersChanged = false; + private final WorkerConfiguration workerConfiguration; public AnnotatedWorkerExecutor(TaskClient taskClient) { @@ -76,9 +82,11 @@ public AnnotatedWorkerExecutor(TaskClient taskClient, WorkerConfiguration worker * implementation */ public synchronized void initWorkers(String... basePackages) { - // scanWorkers() -> initWorkersFromClasses() -> initWorkersFromInstances() already calls startPolling(); - // an extra call here used to race with that one and could drop a task polled by the runner it replaces. scanWorkers(basePackages); + // scanWorkers() -> initWorkersFromClasses() -> initWorkersFromInstances() already reaches + // startPolling(). This second call is therefore redundant, but startPolling() is idempotent + // while the worker set is unchanged, so it is a no-op rather than a task runner restart. + startPolling(); } public synchronized void initWorkersFromInstances(List workerInstances) { @@ -158,10 +166,7 @@ private void scanWorkers(String... basePackages) { initWorkersFromClasses(classes); } catch (Exception e) { - // Rethrow (unchecked) rather than swallow: initWorkers() no longer has its own startPolling() - // fallback, so a swallowed failure here would otherwise leave the caller believing workers are - // running when none were ever started. - throw new RuntimeException("Error while scanning for workers", e); + LOGGER.error("Error while scanning for workers: ", e); } } @@ -226,6 +231,7 @@ private void addMethod(WorkerTask annotation, Method method, Object bean) { for (int i = 0; i < pollerCount; i++) { workers.add(executor); } + workersChanged = true; LOGGER.info( "Adding worker for task {}, method {} with threadCount {} and polling interval set to {} ms", @@ -235,11 +241,28 @@ private void addMethod(WorkerTask annotation, Method method, Object bean) { pollingInterval); } - public void startPolling() { + /** + * Builds a {@link TaskRunnerConfigurer} over the currently registered workers and starts polling. + * + *

Idempotent: if a task runner is already polling and no worker has been added since it was + * built, this returns without doing anything. Restarting unnecessarily would stand up a second + * runner polling the same task types and only shut the first one down afterwards, so a task + * already leased by the outgoing runner could be left in-progress until its response timeout + * expired. Callers that add workers and call this again still get the intended restart. + */ + public synchronized void startPolling() { if (workers.isEmpty()) { return; } + if (taskRunner != null && !workersChanged) { + LOGGER.debug( + "Task runner is already polling {} workers and the worker set is unchanged; " + + "skipping redundant restart.", + workers.size()); + return; + } + LOGGER.info("Starting {} with threadCount {}", workers.stream().map(Worker::getTaskDefName).toList(), workerToThreadCount); LOGGER.info("Worker domains {}", workerDomains); LOGGER.info("Worker workerToPollTimeout (in millis) {}", workerToPollTimeout); @@ -257,6 +280,8 @@ public void startPolling() { taskRunner = builder.build(); taskRunner.init(); + workersChanged = false; + oldTaskRunner.ifPresent(taskRunner -> { LOGGER.trace("Shutting down previous task runner with {} workers.", taskRunner.getWorkerCount()); taskRunner.shutdown(); diff --git a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java index abc4a19fd..2e5a3676c 100644 --- a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java +++ b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java @@ -84,6 +84,60 @@ void directInstanceSupply() { ))); } + @Test + @DisplayName("initWorkers should leave exactly one task runner polling") + void initWorkersStartsASingleTaskRunner() { + var executor = new AnnotatedWorkerExecutor(mock(TaskClient.class)); + executor.initWorkers("com.netflix.conductor.sdk.workflow.executor.task.workers1"); + + var runner = executor.getTaskRunner(); + assertNotNull(runner); + + // The shape used by examples/old/.../taskdomains/Main.java and by callers following the + // docs: initWorkers() followed by an explicit startPolling(). This must not stand up a + // replacement runner alongside the live one. + executor.startPolling(); + assertSame(runner, executor.getTaskRunner()); + + executor.shutdown(); + } + + @Test + @DisplayName("startPolling should be a no-op while the worker set is unchanged") + void startPollingIsIdempotent() { + var executor = new AnnotatedWorkerExecutor(mock(TaskClient.class)); + executor.addBean(new MultipleInputParams()); + + executor.startPolling(); + var first = executor.getTaskRunner(); + assertNotNull(first); + + executor.startPolling(); + executor.startPolling(); + assertSame(first, executor.getTaskRunner()); + + executor.shutdown(); + } + + @Test + @DisplayName("startPolling should still rebuild the task runner once new workers are added") + void startPollingRebuildsWhenWorkersAreAdded() { + var executor = new AnnotatedWorkerExecutor(mock(TaskClient.class)); + executor.addBean(new MultipleInputParams()); + executor.startPolling(); + var first = executor.getTaskRunner(); + assertEquals(1, first.getWorkerCount()); + + executor.addBean(new AnotherAnnotationInput()); + executor.startPolling(); + var second = executor.getTaskRunner(); + + assertNotSame(first, second); + assertEquals(2, second.getWorkerCount()); + + executor.shutdown(); + } + @Test @DisplayName("it should handle null values when InputParam is a List") void nullListAsInputParam() throws NoSuchMethodException { From 05c248e2b281ca08f23e64563bb3646e3466dfa8 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 11:41:28 -0600 Subject: [PATCH 12/21] bound the workflow monitor's retries instead of warning on every tick scheduleAtFixedRate cancels all future ticks on an uncaught exception, so the monitor must not let a failed getWorkflow escape. Catching unconditionally traded that for a workflow id that never resolves -- a purged workflow, expired credentials -- spinning at the 100ms poll interval forever, logging a stack trace each time while its caller blocks with no signal. Track when a run of consecutive failures started per workflow id: warn once, stay at DEBUG while it continues, and after a minute give up, drop the entry and completeExceptionally the future so the caller learns instead of hanging. Co-Authored-By: Claude Opus 5 (1M context) --- .../workflow/executor/WorkflowExecutor.java | 49 +++++++- .../WorkflowExecutorMonitorTests.java | 105 ++++++++++++++++++ 2 files changed, 151 insertions(+), 3 deletions(-) create mode 100644 conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java diff --git a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java index b88d85150..1a7b96228 100644 --- a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java +++ b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java @@ -66,6 +66,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.annotations.VisibleForTesting; public class WorkflowExecutor { @@ -77,9 +78,16 @@ public class WorkflowExecutor { private final TypeReference> listOfTaskDefs = new TypeReference<>() { }; + private static final long DEFAULT_MONITOR_FAILURE_GIVE_UP_MILLIS = TimeUnit.MINUTES.toMillis(1); + private final Map> runningWorkflowFutures = new ConcurrentHashMap<>(); + /** When the current run of consecutive polling failures started, per workflow id. */ + private final Map monitorFailingSince = new ConcurrentHashMap<>(); + + private volatile long monitorFailureGiveUpMillis = DEFAULT_MONITOR_FAILURE_GIVE_UP_MILLIS; + private final ObjectMapper objectMapper = new ObjectMapperProvider().getObjectMapper(); private final TaskClient taskClient; @@ -176,14 +184,18 @@ private void initMonitor() { CompletableFuture future = entry.getValue(); try { Workflow workflow = workflowClient.getWorkflow(workflowId, true); + monitorFailingSince.remove(workflowId); if (workflow.getStatus().isTerminal()) { future.complete(workflow); runningWorkflowFutures.remove(workflowId); } } catch (Exception e) { - // scheduleAtFixedRate silently kills all future ticks on any uncaught exception, so catch here instead of letting one transient error stop completion-tracking forever. - LOGGER.warn("Error polling workflow {} for completion; will retry on " - + "the next tick", workflowId, e); + // scheduleAtFixedRate silently kills all future ticks on any uncaught + // exception, so one transient error here would otherwise stop completion + // tracking for every workflow, forever. Catch, but do not retry forever: + // a workflow id that never becomes resolvable would spin at the polling + // interval indefinitely while its caller blocks with no signal. + handleMonitorFailure(workflowId, future, e); } } }, @@ -192,6 +204,37 @@ private void initMonitor() { TimeUnit.MILLISECONDS); } + private void handleMonitorFailure(String workflowId, CompletableFuture future, Exception e) { + long now = System.currentTimeMillis(); + Long failingSince = monitorFailingSince.putIfAbsent(workflowId, now); + + if (failingSince == null) { + LOGGER.warn("Error polling workflow {} for completion; will retry on the next tick", + workflowId, e); + return; + } + + long failingForMillis = now - failingSince; + if (failingForMillis < monitorFailureGiveUpMillis) { + // Already warned once for this run of failures. Staying at DEBUG keeps a persistently + // unresolvable workflow from emitting a stack trace on every tick. + LOGGER.debug("Still failing to poll workflow {} for completion ({} ms so far)", + workflowId, failingForMillis, e); + return; + } + + LOGGER.error("Giving up polling workflow {} for completion after {} ms of consecutive " + + "failures; completing its future exceptionally", workflowId, failingForMillis, e); + monitorFailingSince.remove(workflowId); + runningWorkflowFutures.remove(workflowId); + future.completeExceptionally(e); + } + + @VisibleForTesting + void setMonitorFailureGiveUpMillis(long monitorFailureGiveUpMillis) { + this.monitorFailureGiveUpMillis = monitorFailureGiveUpMillis; + } + public void initWorkers(String... packagesToScan) { annotatedWorkerExecutor.initWorkers(packagesToScan); } diff --git a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java new file mode 100644 index 000000000..37a60236d --- /dev/null +++ b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java @@ -0,0 +1,105 @@ +/* + * Copyright 2026 Conductor Authors. + *

+ * 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 + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * 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.sdk.workflow.executor; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import com.netflix.conductor.client.http.MetadataClient; +import com.netflix.conductor.client.http.TaskClient; +import com.netflix.conductor.client.http.WorkflowClient; +import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest; +import com.netflix.conductor.common.run.Workflow; +import com.netflix.conductor.sdk.workflow.executor.task.AnnotatedWorkerExecutor; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Covers the completion-tracking monitor started by {@link WorkflowExecutor}'s constructors. The + * monitor runs under {@code scheduleAtFixedRate}, which cancels all future ticks on any uncaught + * exception, so a single failed {@code getWorkflow} call must not be allowed to escape. + */ +public class WorkflowExecutorMonitorTests { + + private static final String WORKFLOW_ID = "test-workflow-id"; + + private WorkflowExecutor executorFor(WorkflowClient workflowClient) { + return new WorkflowExecutor( + mock(TaskClient.class), + workflowClient, + mock(MetadataClient.class), + mock(AnnotatedWorkerExecutor.class)); + } + + @Test + @DisplayName("the monitor should keep polling after a transient getWorkflow failure") + void monitorSurvivesTransientPollFailure() throws Exception { + Workflow completed = new Workflow(); + completed.setStatus(Workflow.WorkflowStatus.COMPLETED); + + WorkflowClient workflowClient = mock(WorkflowClient.class); + when(workflowClient.startWorkflow(any(StartWorkflowRequest.class))).thenReturn(WORKFLOW_ID); + when(workflowClient.getWorkflow(anyString(), anyBoolean())) + .thenThrow(new RuntimeException("transient failure")) + .thenReturn(completed); + + WorkflowExecutor executor = executorFor(workflowClient); + try { + CompletableFuture future = executor.executeWorkflow("wf", 1, Map.of()); + + Workflow result = future.get(5, TimeUnit.SECONDS); + + assertEquals(Workflow.WorkflowStatus.COMPLETED, result.getStatus()); + } finally { + executor.shutdown(); + } + } + + @Test + @DisplayName("the monitor should give up and fail the future once the failure budget is spent") + void monitorGivesUpOnPersistentPollFailure() { + WorkflowClient workflowClient = mock(WorkflowClient.class); + when(workflowClient.startWorkflow(any(StartWorkflowRequest.class))).thenReturn(WORKFLOW_ID); + when(workflowClient.getWorkflow(anyString(), anyBoolean())) + .thenThrow(new RuntimeException("permanent failure")); + + WorkflowExecutor executor = executorFor(workflowClient); + try { + // Zero budget: give up on the tick after the first failure, so the test does not have + // to sit out the production budget. + executor.setMonitorFailureGiveUpMillis(0); + + CompletableFuture future = executor.executeWorkflow("wf", 1, Map.of()); + + ExecutionException thrown = assertThrows( + ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); + + assertInstanceOf(RuntimeException.class, thrown.getCause()); + assertEquals("permanent failure", thrown.getCause().getMessage()); + } finally { + executor.shutdown(); + } + } +} From 8138cfb1d4f84eb4304fe1f80d14c3d574902c88 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 11:41:28 -0600 Subject: [PATCH 13/21] keep testCreateWorkflow on the documented worker-startup shape, and report why it fails initWorkers() followed by an explicit startPolling() is what the docs and examples do, so the integration test should exercise it; startPolling() is now idempotent, so the second call is a no-op rather than a runner restart. fail(e) rather than fail(e.getMessage()): a TimeoutException carries a null message, so this test's only CI failure surfaced as a bare AssertionFailedError with nothing to diagnose. Co-Authored-By: Claude Opus 5 (1M context) --- .../java/io/orkes/conductor/sdk/WorkflowSDKTests.java | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java index 25d5d303f..391cfd926 100644 --- a/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java +++ b/tests/src/test/java/io/orkes/conductor/sdk/WorkflowSDKTests.java @@ -40,8 +40,12 @@ public void testCreateWorkflow() throws Exception { ConductorClient client = ClientTestUtil.getClient(); AnnotatedWorkerExecutor workerExecutor = new AnnotatedWorkerExecutor(new TaskClient(client), new WorkerConfiguration()); - // initWorkers() already starts polling; a redundant extra startPolling() call here used to race with AnnotatedWorkerExecutor's own double-start (likely a real bug there) and could drop the first polled task. workerExecutor.initWorkers("io.orkes.conductor.sdk"); + // Redundant -- initWorkers() already reaches startPolling() -- but kept deliberately: this is + // the shape the docs and examples use, and startPolling() is now idempotent, so it must not + // restart the runner out from under an in-flight poll. See + // AnnotatedWorkerTests#initWorkersStartsASingleTaskRunner. + workerExecutor.startPolling(); WorkflowExecutor executor = new WorkflowExecutor(client, workerExecutor); @@ -60,7 +64,9 @@ public void testCreateWorkflow() throws Exception { Assertions.assertNotNull(executedWorkflow); Assertions.assertEquals(Workflow.WorkflowStatus.COMPLETED, executedWorkflow.getStatus()); } catch (Exception e) { - Assertions.fail(e.getMessage()); + // fail(e), not fail(e.getMessage()): a TimeoutException carries a null message, which + // previously surfaced in CI as a bare AssertionFailedError with nothing to diagnose. + Assertions.fail(e); } } From 54ba4047a565aef8a651974e8801ab897c04a439 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 12:40:17 -0600 Subject: [PATCH 14/21] run the OSS integration suite from ci.yml, and make the helper script reproduce it - distinct check_name per action-junit-report step; all of them defaulted to "JUnit Test Report", so the build job's report and the OSS job's landed on a single check run and overwrote each other - name the compose project, so the stack does not collide with the identically located compose file in the other SDK repos, on both project name and port - unset CONDUCTOR_AUTH_KEY/SECRET before the run: OSS has no /token endpoint, and ClientTestUtil builds its client with useEnvVariables(true), so a shell still holding Orkes credentials sent the whole run through an auth flow the local server cannot serve - point the script header at ci.yml, where the job actually lives Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/ci.yml | 9 +++++++-- scripts/docker-compose-oss.yaml | 11 +++++++++++ scripts/run-integration-oss.sh | 9 ++++++++- 3 files changed, 26 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1a3317137..5c4242b02 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -158,7 +158,11 @@ jobs: uses: mikepenz/action-junit-report@v6 with: report_paths: '**/build/test-results/test/TEST-*.xml' - + # Distinct per suite: the default is 'JUnit Test Report' for every caller, so this + # report, the OSS one below, and the one in integration-tests.yml all landed on a + # single check run and overwrote each other. + check_name: Unit Test Report + - name: Check Tests Status if: steps.tests.outcome == 'failure' run: | @@ -212,9 +216,10 @@ jobs: uses: mikepenz/action-junit-report@v6 with: report_paths: '**/tests/build/test-results/test/TEST-*.xml' + check_name: OSS Integration Test Report - name: Check Integration Tests Status if: steps.integration_tests.outcome == 'failure' run: | echo "::error::Integration tests (OSS) failed. See the 'Run integration tests (OSS)' step above for the Gradle/test output, the 'Dump Conductor logs' step for server-side logs, and the 'Publish Test Report' step's JUnit summary for which test(s) failed." - exit 1 \ No newline at end of file + exit 1 diff --git a/scripts/docker-compose-oss.yaml b/scripts/docker-compose-oss.yaml index efc517329..c5edf8c17 100644 --- a/scripts/docker-compose-oss.yaml +++ b/scripts/docker-compose-oss.yaml @@ -1,3 +1,14 @@ +# Conductor OSS stack used to run the SDK integration tests against open-source Conductor. +# Shared by scripts/run-integration-oss.sh and the integration-tests-oss job in +# .github/workflows/ci.yml. +# +# OSS_CONDUCTOR_VERSION defaults to `latest` for local runs; CI pins it via the +# E2E_TEST_OSS_CONDUCTOR_VERSION org variable (or a workflow_dispatch input). +# +# Per-repo name; unnamed, the project defaults to this file's dir (`scripts`) in every SDK repo +# and stacks collide -- both on the project name and on port 8080. +name: java-sdk-oss-e2e + services: conductor-server: image: conductoross/conductor:${OSS_CONDUCTOR_VERSION:-latest} diff --git a/scripts/run-integration-oss.sh b/scripts/run-integration-oss.sh index 1a17b6d3f..c5a45ae5f 100755 --- a/scripts/run-integration-oss.sh +++ b/scripts/run-integration-oss.sh @@ -2,7 +2,7 @@ # # Spin up a local Conductor OSS stack and run the `tests` module's # integration suite against it, mirroring the `integration-tests-oss` job in -# .github/workflows/integration-tests-oss.yml. Orkes-Enterprise-only test +# .github/workflows/ci.yml. Orkes-Enterprise-only test # classes are annotated with # @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss") # so they skip themselves when it's set (see the individual test files for @@ -88,6 +88,13 @@ echo "Conductor is up." export CONDUCTOR_SERVER_URL="http://localhost:8080/api" +# Plain OSS Conductor has no authentication layer and no /token endpoint. ClientTestUtil builds +# its client with useEnvVariables(true), and ApiClient.applyEnvVariables() attaches credentials +# whenever both of these are present -- so a shell that still has them exported for the Orkes +# suite would send the whole run through an auth flow the local server cannot serve. +unset CONDUCTOR_AUTH_KEY CONDUCTOR_AUTH_SECRET +unset CONDUCTOR_SERVER_AUTH_KEY CONDUCTOR_SERVER_AUTH_SECRET + if [[ "${INCLUDE_GATED}" == "1" ]]; then echo "--include-gated set: leaving CONDUCTOR_SERVER_TYPE unset, so tests normally" \ "skipped as Orkes-only will run against OSS too." From 4b13a476a1484c63d01254e10255f93ce5026c69 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 13:31:20 -0600 Subject: [PATCH 15/21] adjustments for feedback suggestions --- .github/workflows/ci.yml | 70 +------------- .github/workflows/integration-tests.yml | 96 ++++++++++++++++++- CONTRIBUTING.md | 2 +- .../workflow/executor/WorkflowExecutor.java | 28 +++++- .../task/AnnotatedWorkerExecutor.java | 22 ++++- .../WorkflowExecutorMonitorTests.java | 45 ++++++++- .../executor/task/AnnotatedWorkerTests.java | 24 +++++ scripts/docker-compose-oss.yaml | 6 +- scripts/run-integration-oss.sh | 2 +- .../orkes/conductor/client/util/TestUtil.java | 37 ++++--- 10 files changed, 225 insertions(+), 107 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5c4242b02..6a5b6fa5c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,20 +8,11 @@ on: branches: - main workflow_dispatch: - inputs: - oss_conductor_version: - description: 'OSS Conductor image tag (falls back to E2E_TEST_OSS_CONDUCTOR_VERSION org var)' - required: false - type: string concurrency: group: ${{ github.workflow }}-${{ github.ref }} cancel-in-progress: true -permissions: - contents: read - checks: write - jobs: documentation-validation: runs-on: ubuntu-latest @@ -158,9 +149,9 @@ jobs: uses: mikepenz/action-junit-report@v6 with: report_paths: '**/build/test-results/test/TEST-*.xml' - # Distinct per suite: the default is 'JUnit Test Report' for every caller, so this - # report, the OSS one below, and the one in integration-tests.yml all landed on a - # single check run and overwrote each other. + # Distinct per suite: the action defaults to 'JUnit Test Report' for every caller, so + # this report and the ones published by integration-tests.yml all landed on a single + # check run and overwrote each other. check_name: Unit Test Report - name: Check Tests Status @@ -168,58 +159,3 @@ jobs: run: | echo "::error::Tests failed. See the 'Run Tests' step above for the Gradle/test output, and the 'Publish Test Report' step's JUnit summary for which test(s) failed." exit 1 - - integration-tests-oss: - runs-on: ubuntu-latest - name: Integration Tests (OSS) - timeout-minutes: 30 - env: - CONDUCTOR_SERVER_URL: http://localhost:8080/api - CONDUCTOR_SERVER_TYPE: oss - OSS_CONDUCTOR_VERSION: ${{ inputs.oss_conductor_version || vars.E2E_TEST_OSS_CONDUCTOR_VERSION }} - - steps: - - name: Verify OSS Conductor version is set - run: | - if [ -z "$OSS_CONDUCTOR_VERSION" ]; then - echo "::error::No Conductor OSS image tag resolved. Set the E2E_TEST_OSS_CONDUCTOR_VERSION organization variable (and ensure its repository access policy includes this repo), or pass the oss_conductor_version input via workflow_dispatch." - exit 1 - fi - echo "Using conductoross/conductor:$OSS_CONDUCTOR_VERSION" - - - name: Checkout - uses: actions/checkout@v6 - - - name: Set up Zulu JDK 21 - uses: actions/setup-java@v5 - with: - distribution: "zulu" - java-version: "21" - - - name: Start Conductor OSS stack - run: docker compose -f scripts/docker-compose-oss.yaml up -d - - - name: Wait for Conductor to be healthy - run: timeout 120 bash -c 'until curl -sf http://localhost:8080/health; do sleep 5; done' - - - name: Run integration tests (OSS) - id: integration_tests - continue-on-error: true - run: ./gradlew :tests:test -PIntegrationTests - - - name: Dump Conductor logs - if: failure() || steps.integration_tests.outcome == 'failure' - run: docker compose -f scripts/docker-compose-oss.yaml logs conductor-server - - - name: Publish Test Report - if: always() - uses: mikepenz/action-junit-report@v6 - with: - report_paths: '**/tests/build/test-results/test/TEST-*.xml' - check_name: OSS Integration Test Report - - - name: Check Integration Tests Status - if: steps.integration_tests.outcome == 'failure' - run: | - echo "::error::Integration tests (OSS) failed. See the 'Run integration tests (OSS)' step above for the Gradle/test output, the 'Dump Conductor logs' step for server-side logs, and the 'Publish Test Report' step's JUnit summary for which test(s) failed." - exit 1 diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 9ffbd9081..1c52d45aa 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -5,6 +5,12 @@ on: workflows: ["Java Client Build"] types: - completed + workflow_dispatch: + inputs: + oss_conductor_version: + description: 'OSS Conductor image tag (falls back to E2E_TEST_OSS_CONDUCTOR_VERSION org var)' + required: false + type: string # allow this workflow to update the status of the PR that triggered it permissions: @@ -44,7 +50,11 @@ jobs: if: always() with: report_paths: '**/build/test-results/test/TEST-*.xml' - + # Distinct per suite: the action defaults to 'JUnit Test Report' for every caller, so + # this report, the OSS one below, and ci.yml's unit report all landed on a single + # check run and overwrote each other. + check_name: Integration Test Report + - name: Update PR Status if: always() uses: actions/github-script@v8 @@ -60,4 +70,88 @@ jobs: description: 'Integration tests ${{ job.status }}' }); + # Runs against a throwaway Conductor OSS stack rather than the Orkes deployment the job above + # targets, so it needs no secrets and no `environment`. It is deliberately a sibling of that + # job, not a dependent: an OSS failure must not suppress the enterprise suite, or vice versa. + integration-tests-oss: + runs-on: ubuntu-latest + name: Integration Tests (OSS) + timeout-minutes: 30 + if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success' }} + env: + CONDUCTOR_SERVER_URL: http://localhost:8080/api + CONDUCTOR_SERVER_TYPE: oss + OSS_CONDUCTOR_VERSION: ${{ inputs.oss_conductor_version || vars.E2E_TEST_OSS_CONDUCTOR_VERSION }} + + steps: + - name: Verify OSS Conductor version is set + run: | + if [ -z "$OSS_CONDUCTOR_VERSION" ]; then + echo "::error::No Conductor OSS image tag resolved. Set the E2E_TEST_OSS_CONDUCTOR_VERSION organization variable (and ensure its repository access policy includes this repo), or pass the oss_conductor_version input via workflow_dispatch." + exit 1 + fi + echo "Using conductoross/conductor:$OSS_CONDUCTOR_VERSION" + + - name: Checkout + uses: actions/checkout@v6 + with: + # workflow_run carries the triggering run's head; workflow_dispatch has neither, and + # falls back to the ref the dispatch was made against. + ref: ${{ github.event.workflow_run.head_sha || github.sha }} + repository: ${{ github.event.workflow_run.repository.full_name || github.repository }} + + - name: Set up Zulu JDK 21 + uses: actions/setup-java@v5 + with: + distribution: "zulu" + java-version: "21" + + - name: Start Conductor OSS stack + run: docker compose -f scripts/docker-compose-oss.yaml up -d + + - name: Wait for Conductor to be healthy + # Matches HEALTH_TIMEOUT in scripts/run-integration-oss.sh, and stays under the compose + # healthcheck's own ~200s budget. + run: timeout 180 bash -c 'until curl -sf http://localhost:8080/health; do sleep 5; done' + + - name: Run integration tests (OSS) + id: integration_tests + continue-on-error: true + run: ./gradlew :tests:test -PIntegrationTests + + - name: Dump Conductor logs + if: failure() || steps.integration_tests.outcome == 'failure' + run: docker compose -f scripts/docker-compose-oss.yaml logs conductor-server + + - name: Publish Test Report + if: always() + uses: mikepenz/action-junit-report@v6 + with: + report_paths: 'tests/build/test-results/test/TEST-*.xml' + check_name: OSS Integration Test Report + + # Before Update PR Status, not after: the test step is continue-on-error, so job.status is + # still 'success' until this step's exit 1 flips it. + - name: Check Integration Tests Status + if: steps.integration_tests.outcome == 'failure' + run: | + echo "::error::Integration tests (OSS) failed. See the 'Run integration tests (OSS)' step above for the Gradle/test output, the 'Dump Conductor logs' step for server-side logs, and the 'Publish Test Report' step's JUnit summary for which test(s) failed." + exit 1 + + - name: Update PR Status + # Skipped on workflow_dispatch: there is no triggering run, so there is no PR head to + # report against and context.payload.workflow_run would be undefined. + if: ${{ always() && github.event_name == 'workflow_run' }} + uses: actions/github-script@v8 + with: + script: | + const { owner, repo } = context.repo; + const sha = context.payload.workflow_run.head_sha; + await github.rest.repos.createCommitStatus({ + owner, repo, sha, + state: '${{ job.status }}' === 'success' ? 'success' : 'failure', + context: 'Integration Tests (OSS)', + target_url: `${context.serverUrl}/${owner}/${repo}/actions/runs/${context.runId}`, + description: 'OSS integration tests ${{ job.status }}' + }); diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index bb0172b6d..7876321c5 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -27,7 +27,7 @@ Run the SDK test suite: The `tests` module also has an integration suite (`-PIntegrationTests`) that runs against a real Conductor server, separate from the unit suite above. `scripts/run-integration-oss.sh` -mirrors the `integration-tests-oss` job in `ci.yml`: it starts a local Conductor OSS + +mirrors the `integration-tests-oss` job in `integration-tests.yml`: it starts a local Conductor OSS + Postgres stack (defined in `scripts/docker-compose-oss.yaml`), waits for `/health`, runs the integration suite, and tears the stack down on exit. diff --git a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java index 1a7b96228..ff9dbfeb7 100644 --- a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java +++ b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java @@ -66,7 +66,6 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; -import com.google.common.annotations.VisibleForTesting; public class WorkflowExecutor { @@ -78,7 +77,8 @@ public class WorkflowExecutor { private final TypeReference> listOfTaskDefs = new TypeReference<>() { }; - private static final long DEFAULT_MONITOR_FAILURE_GIVE_UP_MILLIS = TimeUnit.MINUTES.toMillis(1); + /** Zero, i.e. never give up. See {@link #setMonitorFailureGiveUpMillis(long)}. */ + private static final long DEFAULT_MONITOR_FAILURE_GIVE_UP_MILLIS = 0; private final Map> runningWorkflowFutures = new ConcurrentHashMap<>(); @@ -214,8 +214,9 @@ private void handleMonitorFailure(String workflowId, CompletableFuture return; } + long giveUpMillis = monitorFailureGiveUpMillis; long failingForMillis = now - failingSince; - if (failingForMillis < monitorFailureGiveUpMillis) { + if (giveUpMillis <= 0 || failingForMillis < giveUpMillis) { // Already warned once for this run of failures. Staying at DEBUG keeps a persistently // unresolvable workflow from emitting a stack trace on every tick. LOGGER.debug("Still failing to poll workflow {} for completion ({} ms so far)", @@ -230,11 +231,28 @@ private void handleMonitorFailure(String workflowId, CompletableFuture future.completeExceptionally(e); } - @VisibleForTesting - void setMonitorFailureGiveUpMillis(long monitorFailureGiveUpMillis) { + /** + * How long the completion monitor keeps retrying a workflow whose status cannot be fetched + * before giving up on it. + * + * @param monitorFailureGiveUpMillis zero or negative (the default) to never give up: the + * monitor retries such a workflow for as long as this executor lives, so a server outage + * longer than any fixed budget — a rolling restart, a failover — does not strand futures + * that would otherwise have completed once the server came back. A positive value bounds + * that: once a workflow has failed to poll continuously for this long, the monitor stops + * tracking it and completes its future exceptionally, so a caller blocked in + * {@code executeWorkflow(...).get()} sees an {@link java.util.concurrent.ExecutionException} + * rather than blocking indefinitely on a workflow id that will never resolve. + */ + public void setMonitorFailureGiveUpMillis(long monitorFailureGiveUpMillis) { this.monitorFailureGiveUpMillis = monitorFailureGiveUpMillis; } + /** @see #setMonitorFailureGiveUpMillis(long) */ + public long getMonitorFailureGiveUpMillis() { + return monitorFailureGiveUpMillis; + } + public void initWorkers(String... packagesToScan) { annotatedWorkerExecutor.initWorkers(packagesToScan); } diff --git a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java index 1894ce162..634f9399f 100644 --- a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java +++ b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java @@ -118,10 +118,18 @@ public synchronized void initWorkersFromClasses(List> classes } - /** Shuts down the workers */ - public void shutdown() { + /** + * Shuts down the workers. + * + *

Clears the task runner as well as shutting it down, so a later {@link #startPolling()} + * builds a fresh one and resumes. A {@link TaskRunnerConfigurer} is single-use — its + * shutdown closes the executor backing it — so leaving the field set would make + * startPolling() take its already-polling fast path and never poll again. + */ + public synchronized void shutdown() { if (taskRunner != null) { taskRunner.shutdown(); + taskRunner = null; } } @@ -177,7 +185,15 @@ private boolean classBelongsToPackage(List packagesToScan, String classN return false; } - public void addBean(Object bean) { + /** + * Registers every {@link WorkerTask}-annotated method on the bean as a worker. + * + *

Synchronized because it is the public entry point that mutates the worker set and the + * {@code workersChanged} flag {@link #startPolling()} reads to decide whether a restart is + * needed. Without a common lock, a worker added on another thread could be missed and never + * polled. + */ + public synchronized void addBean(Object bean) { Class clazz = bean.getClass(); for (Method method : clazz.getMethods()) { WorkerTask annotation = method.getAnnotation(WorkerTask.class); diff --git a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java index 37a60236d..7e2fabfd1 100644 --- a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java +++ b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java @@ -16,6 +16,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; @@ -33,6 +34,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -45,6 +47,8 @@ public class WorkflowExecutorMonitorTests { private static final String WORKFLOW_ID = "test-workflow-id"; + private static final String OTHER_WORKFLOW_ID = "other-test-workflow-id"; + private WorkflowExecutor executorFor(WorkflowClient workflowClient) { return new WorkflowExecutor( mock(TaskClient.class), @@ -87,9 +91,10 @@ void monitorGivesUpOnPersistentPollFailure() { WorkflowExecutor executor = executorFor(workflowClient); try { - // Zero budget: give up on the tick after the first failure, so the test does not have - // to sit out the production budget. - executor.setMonitorFailureGiveUpMillis(0); + // 1ms budget: the monitor ticks every 100ms, so the second consecutive failure is + // already past it. Keeps the test off the wall clock while still exercising a real + // (positive, opt-in) budget rather than the never-give-up default. + executor.setMonitorFailureGiveUpMillis(1); CompletableFuture future = executor.executeWorkflow("wf", 1, Map.of()); @@ -102,4 +107,38 @@ void monitorGivesUpOnPersistentPollFailure() { executor.shutdown(); } } + + @Test + @DisplayName("by default the monitor should never give up, and one bad workflow should not stall the rest") + void monitorDoesNotGiveUpByDefault() throws Exception { + Workflow completed = new Workflow(); + completed.setStatus(Workflow.WorkflowStatus.COMPLETED); + + WorkflowClient workflowClient = mock(WorkflowClient.class); + when(workflowClient.startWorkflow(any(StartWorkflowRequest.class))) + .thenReturn(WORKFLOW_ID) + .thenReturn(OTHER_WORKFLOW_ID); + when(workflowClient.getWorkflow(eq(WORKFLOW_ID), anyBoolean())) + .thenThrow(new RuntimeException("permanent failure")); + when(workflowClient.getWorkflow(eq(OTHER_WORKFLOW_ID), anyBoolean())) + .thenReturn(completed); + + WorkflowExecutor executor = executorFor(workflowClient); + try { + assertEquals(0, executor.getMonitorFailureGiveUpMillis(), "give-up should be off by default"); + + CompletableFuture failing = executor.executeWorkflow("wf", 1, Map.of()); + CompletableFuture healthy = executor.executeWorkflow("wf", 1, Map.of()); + + // The unresolvable workflow must not take the monitor down with it: a sibling + // registered on the same tick loop still completes. + assertEquals(Workflow.WorkflowStatus.COMPLETED, healthy.get(5, TimeUnit.SECONDS).getStatus()); + + // ...and the unresolvable one stays pending rather than being failed, which is the + // pre-bounded-retry behavior callers depend on across a server restart. + assertThrows(TimeoutException.class, () -> failing.get(1, TimeUnit.SECONDS)); + } finally { + executor.shutdown(); + } + } } diff --git a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java index 2e5a3676c..37c0b4c7c 100644 --- a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java +++ b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerTests.java @@ -138,6 +138,30 @@ void startPollingRebuildsWhenWorkersAreAdded() { executor.shutdown(); } + @Test + @DisplayName("startPolling should build a fresh task runner after a shutdown") + void startPollingRestartsAfterShutdown() { + var executor = new AnnotatedWorkerExecutor(mock(TaskClient.class)); + executor.addBean(new MultipleInputParams()); + executor.startPolling(); + var first = executor.getTaskRunner(); + assertNotNull(first); + + // A TaskRunnerConfigurer is single-use, so shutdown() has to clear the field as well as + // shut the runner down. Otherwise startPolling()'s idempotence check sees a non-null + // runner with an unchanged worker set and silently declines to poll ever again. + executor.shutdown(); + assertNull(executor.getTaskRunner()); + + executor.startPolling(); + var second = executor.getTaskRunner(); + assertNotNull(second); + assertNotSame(first, second); + assertEquals(1, second.getWorkerCount()); + + executor.shutdown(); + } + @Test @DisplayName("it should handle null values when InputParam is a List") void nullListAsInputParam() throws NoSuchMethodException { diff --git a/scripts/docker-compose-oss.yaml b/scripts/docker-compose-oss.yaml index c5edf8c17..5ecced59c 100644 --- a/scripts/docker-compose-oss.yaml +++ b/scripts/docker-compose-oss.yaml @@ -1,13 +1,9 @@ # Conductor OSS stack used to run the SDK integration tests against open-source Conductor. # Shared by scripts/run-integration-oss.sh and the integration-tests-oss job in -# .github/workflows/ci.yml. +# .github/workflows/integration-tests.yml. # # OSS_CONDUCTOR_VERSION defaults to `latest` for local runs; CI pins it via the # E2E_TEST_OSS_CONDUCTOR_VERSION org variable (or a workflow_dispatch input). -# -# Per-repo name; unnamed, the project defaults to this file's dir (`scripts`) in every SDK repo -# and stacks collide -- both on the project name and on port 8080. -name: java-sdk-oss-e2e services: conductor-server: diff --git a/scripts/run-integration-oss.sh b/scripts/run-integration-oss.sh index c5a45ae5f..1d0d27ac7 100755 --- a/scripts/run-integration-oss.sh +++ b/scripts/run-integration-oss.sh @@ -2,7 +2,7 @@ # # Spin up a local Conductor OSS stack and run the `tests` module's # integration suite against it, mirroring the `integration-tests-oss` job in -# .github/workflows/ci.yml. Orkes-Enterprise-only test +# .github/workflows/integration-tests.yml. Orkes-Enterprise-only test # classes are annotated with # @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss") # so they skip themselves when it's set (see the individual test files for diff --git a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java index 6827f61ad..786e7e8a2 100644 --- a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java +++ b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java @@ -163,29 +163,24 @@ private static boolean isTerminalFailure(Workflow workflow) { } /** - * Whether the suite is currently running against plain OSS Conductor rather than Orkes - * Enterprise, per the same {@code CONDUCTOR_SERVER_TYPE} signal that - * {@code @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss")} - * checks for test gating. - */ - public static boolean isOssServer() { - return "oss".equals(System.getenv("CONDUCTOR_SERVER_TYPE")); - } - - /** - * Asserts a caught exception represents "resource doesn't exist", in the shape specific to - * whichever server type {@code CONDUCTOR_SERVER_TYPE} says we're running against: Orkes - * Enterprise reports a proper 404; plain OSS Conductor instead reports a 500 whose message - * contains {@code ossMessageSubstring} (empirically confirmed per endpoint). Anything else - * is rethrown, since it isn't the "doesn't exist" case this is meant to tolerate. + * Tolerates a caught exception that represents "resource doesn't exist", in either shape a + * Conductor server reports it in: a proper 404, or -- on plain OSS Conductor, for the + * endpoints where this has been empirically confirmed -- a 500 whose message contains + * {@code ossMessageSubstring}. Anything else is rethrown, since it isn't the "doesn't exist" + * case this is meant to tolerate. + * + *

Both shapes are accepted regardless of {@code CONDUCTOR_SERVER_TYPE}. Keying off that + * variable would break two ways: {@code run-integration-oss.sh --include-gated} runs against + * OSS with it deliberately unset, and OSS returning a correct 404 for one of these endpoints + * should not start failing the suite. */ public static void assertNotFoundOrRethrow(ConductorClientException e, String ossMessageSubstring) { - if (isOssServer()) { - if (e.getStatus() != 500 || e.getMessage() == null || !e.getMessage().contains(ossMessageSubstring)) { - throw e; - } - } else if (e.getStatus() != 404) { - throw e; + if (e.getStatus() == 404) { + return; + } + if (e.getStatus() == 500 && e.getMessage() != null && e.getMessage().contains(ossMessageSubstring)) { + return; } + throw e; } } From 0bcc28c9a75734456641c6cae9d5732bb156366e Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 13:34:31 -0600 Subject: [PATCH 16/21] try to get job running in the gh wf i want it in but while still on the PR --- .github/workflows/integration-tests.yml | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 1c52d45aa..04234cc85 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -12,13 +12,29 @@ on: required: false type: string + # --------------------------------------------------------------------------- + # TEMPORARY -- REMOVE BEFORE MERGE. + # + # A workflow_run-triggered run always executes the copy of this file on the + # default branch, so the integration-tests-oss job below cannot be exercised + # from a pull request. push runs the branch's own copy, so this trigger is + # what lets the job actually run while it is being developed. + # + # Deleting this block is sufficient to revert; nothing else below depends on it. + # --------------------------------------------------------------------------- + push: + branches: + - e2e-against-conductor-with-local-script + # allow this workflow to update the status of the PR that triggered it permissions: statuses: write checks: write concurrency: - group: ${{ github.workflow }}-${{ github.event.workflow_run.head_branch }} + # head_branch is only set for workflow_run; fall back to the ref so runs from + # other triggers do not all collapse into one unnamed concurrency group. + group: ${{ github.workflow }}-${{ github.event.workflow_run.head_branch || github.ref }} cancel-in-progress: true jobs: @@ -77,7 +93,9 @@ jobs: runs-on: ubuntu-latest name: Integration Tests (OSS) timeout-minutes: 30 - if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success' }} + # The build-succeeded gate only makes sense for workflow_run, which is the only trigger that + # has a triggering run to inspect. Any other trigger is a direct request to run this suite. + if: ${{ github.event_name != 'workflow_run' || github.event.workflow_run.conclusion == 'success' }} env: CONDUCTOR_SERVER_URL: http://localhost:8080/api CONDUCTOR_SERVER_TYPE: oss From b9ed4607464a0b139fbb6cad42098cf4d1125a54 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Mon, 31 Aug 2026 10:20:46 -0600 Subject: [PATCH 17/21] address feedback from self review --- .github/workflows/ci.yml | 10 +-- .../workflow/executor/WorkflowExecutor.java | 70 ++----------------- .../WorkflowExecutorMonitorTests.java | 58 ++++----------- scripts/docker-compose-oss.yaml | 6 +- .../client/http/EventClientTests.java | 2 +- .../client/http/MetadataClientTests.java | 4 +- .../orkes/conductor/client/util/TestUtil.java | 2 +- 7 files changed, 29 insertions(+), 123 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6a5b6fa5c..ecca05d7a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -149,13 +149,7 @@ jobs: uses: mikepenz/action-junit-report@v6 with: report_paths: '**/build/test-results/test/TEST-*.xml' - # Distinct per suite: the action defaults to 'JUnit Test Report' for every caller, so - # this report and the ones published by integration-tests.yml all landed on a single - # check run and overwrote each other. - check_name: Unit Test Report - + - name: Check Tests Status if: steps.tests.outcome == 'failure' - run: | - echo "::error::Tests failed. See the 'Run Tests' step above for the Gradle/test output, and the 'Publish Test Report' step's JUnit summary for which test(s) failed." - exit 1 + run: exit 1 diff --git a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java index ff9dbfeb7..f185e6e6d 100644 --- a/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java +++ b/conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java @@ -77,17 +77,9 @@ public class WorkflowExecutor { private final TypeReference> listOfTaskDefs = new TypeReference<>() { }; - /** Zero, i.e. never give up. See {@link #setMonitorFailureGiveUpMillis(long)}. */ - private static final long DEFAULT_MONITOR_FAILURE_GIVE_UP_MILLIS = 0; - private final Map> runningWorkflowFutures = new ConcurrentHashMap<>(); - /** When the current run of consecutive polling failures started, per workflow id. */ - private final Map monitorFailingSince = new ConcurrentHashMap<>(); - - private volatile long monitorFailureGiveUpMillis = DEFAULT_MONITOR_FAILURE_GIVE_UP_MILLIS; - private final ObjectMapper objectMapper = new ObjectMapperProvider().getObjectMapper(); private final TaskClient taskClient; @@ -184,18 +176,19 @@ private void initMonitor() { CompletableFuture future = entry.getValue(); try { Workflow workflow = workflowClient.getWorkflow(workflowId, true); - monitorFailingSince.remove(workflowId); if (workflow.getStatus().isTerminal()) { future.complete(workflow); runningWorkflowFutures.remove(workflowId); } } catch (Exception e) { // scheduleAtFixedRate silently kills all future ticks on any uncaught - // exception, so one transient error here would otherwise stop completion - // tracking for every workflow, forever. Catch, but do not retry forever: - // a workflow id that never becomes resolvable would spin at the polling - // interval indefinitely while its caller blocks with no signal. - handleMonitorFailure(workflowId, future, e); + // exception, so this must not escape: one failed poll would otherwise + // stop completion tracking for every workflow, forever. Stop tracking + // this one and let its caller see the failure. + LOGGER.error("Error polling workflow {} for completion; completing its " + + "future exceptionally", workflowId, e); + runningWorkflowFutures.remove(workflowId); + future.completeExceptionally(e); } } }, @@ -204,55 +197,6 @@ private void initMonitor() { TimeUnit.MILLISECONDS); } - private void handleMonitorFailure(String workflowId, CompletableFuture future, Exception e) { - long now = System.currentTimeMillis(); - Long failingSince = monitorFailingSince.putIfAbsent(workflowId, now); - - if (failingSince == null) { - LOGGER.warn("Error polling workflow {} for completion; will retry on the next tick", - workflowId, e); - return; - } - - long giveUpMillis = monitorFailureGiveUpMillis; - long failingForMillis = now - failingSince; - if (giveUpMillis <= 0 || failingForMillis < giveUpMillis) { - // Already warned once for this run of failures. Staying at DEBUG keeps a persistently - // unresolvable workflow from emitting a stack trace on every tick. - LOGGER.debug("Still failing to poll workflow {} for completion ({} ms so far)", - workflowId, failingForMillis, e); - return; - } - - LOGGER.error("Giving up polling workflow {} for completion after {} ms of consecutive " - + "failures; completing its future exceptionally", workflowId, failingForMillis, e); - monitorFailingSince.remove(workflowId); - runningWorkflowFutures.remove(workflowId); - future.completeExceptionally(e); - } - - /** - * How long the completion monitor keeps retrying a workflow whose status cannot be fetched - * before giving up on it. - * - * @param monitorFailureGiveUpMillis zero or negative (the default) to never give up: the - * monitor retries such a workflow for as long as this executor lives, so a server outage - * longer than any fixed budget — a rolling restart, a failover — does not strand futures - * that would otherwise have completed once the server came back. A positive value bounds - * that: once a workflow has failed to poll continuously for this long, the monitor stops - * tracking it and completes its future exceptionally, so a caller blocked in - * {@code executeWorkflow(...).get()} sees an {@link java.util.concurrent.ExecutionException} - * rather than blocking indefinitely on a workflow id that will never resolve. - */ - public void setMonitorFailureGiveUpMillis(long monitorFailureGiveUpMillis) { - this.monitorFailureGiveUpMillis = monitorFailureGiveUpMillis; - } - - /** @see #setMonitorFailureGiveUpMillis(long) */ - public long getMonitorFailureGiveUpMillis() { - return monitorFailureGiveUpMillis; - } - public void initWorkers(String... packagesToScan) { annotatedWorkerExecutor.initWorkers(packagesToScan); } diff --git a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java index 7e2fabfd1..56f684ac4 100644 --- a/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java +++ b/conductor-client/src/test/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutorMonitorTests.java @@ -16,7 +16,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; @@ -58,59 +57,30 @@ private WorkflowExecutor executorFor(WorkflowClient workflowClient) { } @Test - @DisplayName("the monitor should keep polling after a transient getWorkflow failure") - void monitorSurvivesTransientPollFailure() throws Exception { - Workflow completed = new Workflow(); - completed.setStatus(Workflow.WorkflowStatus.COMPLETED); - - WorkflowClient workflowClient = mock(WorkflowClient.class); - when(workflowClient.startWorkflow(any(StartWorkflowRequest.class))).thenReturn(WORKFLOW_ID); - when(workflowClient.getWorkflow(anyString(), anyBoolean())) - .thenThrow(new RuntimeException("transient failure")) - .thenReturn(completed); - - WorkflowExecutor executor = executorFor(workflowClient); - try { - CompletableFuture future = executor.executeWorkflow("wf", 1, Map.of()); - - Workflow result = future.get(5, TimeUnit.SECONDS); - - assertEquals(Workflow.WorkflowStatus.COMPLETED, result.getStatus()); - } finally { - executor.shutdown(); - } - } - - @Test - @DisplayName("the monitor should give up and fail the future once the failure budget is spent") - void monitorGivesUpOnPersistentPollFailure() { + @DisplayName("a failed poll should complete that workflow's future exceptionally") + void monitorFailsTheFutureOnPollFailure() { WorkflowClient workflowClient = mock(WorkflowClient.class); when(workflowClient.startWorkflow(any(StartWorkflowRequest.class))).thenReturn(WORKFLOW_ID); when(workflowClient.getWorkflow(anyString(), anyBoolean())) - .thenThrow(new RuntimeException("permanent failure")); + .thenThrow(new RuntimeException("poll failure")); WorkflowExecutor executor = executorFor(workflowClient); try { - // 1ms budget: the monitor ticks every 100ms, so the second consecutive failure is - // already past it. Keeps the test off the wall clock while still exercising a real - // (positive, opt-in) budget rather than the never-give-up default. - executor.setMonitorFailureGiveUpMillis(1); - CompletableFuture future = executor.executeWorkflow("wf", 1, Map.of()); ExecutionException thrown = assertThrows( ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); assertInstanceOf(RuntimeException.class, thrown.getCause()); - assertEquals("permanent failure", thrown.getCause().getMessage()); + assertEquals("poll failure", thrown.getCause().getMessage()); } finally { executor.shutdown(); } } @Test - @DisplayName("by default the monitor should never give up, and one bad workflow should not stall the rest") - void monitorDoesNotGiveUpByDefault() throws Exception { + @DisplayName("one workflow's poll failure should not stop the monitor tracking the rest") + void monitorKeepsTrackingOtherWorkflowsAfterAFailure() throws Exception { Workflow completed = new Workflow(); completed.setStatus(Workflow.WorkflowStatus.COMPLETED); @@ -119,24 +89,20 @@ void monitorDoesNotGiveUpByDefault() throws Exception { .thenReturn(WORKFLOW_ID) .thenReturn(OTHER_WORKFLOW_ID); when(workflowClient.getWorkflow(eq(WORKFLOW_ID), anyBoolean())) - .thenThrow(new RuntimeException("permanent failure")); + .thenThrow(new RuntimeException("poll failure")); when(workflowClient.getWorkflow(eq(OTHER_WORKFLOW_ID), anyBoolean())) .thenReturn(completed); WorkflowExecutor executor = executorFor(workflowClient); try { - assertEquals(0, executor.getMonitorFailureGiveUpMillis(), "give-up should be off by default"); - CompletableFuture failing = executor.executeWorkflow("wf", 1, Map.of()); - CompletableFuture healthy = executor.executeWorkflow("wf", 1, Map.of()); + assertThrows(ExecutionException.class, () -> failing.get(5, TimeUnit.SECONDS)); - // The unresolvable workflow must not take the monitor down with it: a sibling - // registered on the same tick loop still completes. + // Registered only after the failure has already happened, so it can complete at all + // only if the tick loop survived it -- scheduleAtFixedRate would have cancelled every + // future tick had the exception been allowed to escape. + CompletableFuture healthy = executor.executeWorkflow("wf", 1, Map.of()); assertEquals(Workflow.WorkflowStatus.COMPLETED, healthy.get(5, TimeUnit.SECONDS).getStatus()); - - // ...and the unresolvable one stays pending rather than being failed, which is the - // pre-bounded-retry behavior callers depend on across a server restart. - assertThrows(TimeoutException.class, () -> failing.get(1, TimeUnit.SECONDS)); } finally { executor.shutdown(); } diff --git a/scripts/docker-compose-oss.yaml b/scripts/docker-compose-oss.yaml index 5ecced59c..cb1f36f9b 100644 --- a/scripts/docker-compose-oss.yaml +++ b/scripts/docker-compose-oss.yaml @@ -2,8 +2,10 @@ # Shared by scripts/run-integration-oss.sh and the integration-tests-oss job in # .github/workflows/integration-tests.yml. # -# OSS_CONDUCTOR_VERSION defaults to `latest` for local runs; CI pins it via the -# E2E_TEST_OSS_CONDUCTOR_VERSION org variable (or a workflow_dispatch input). +# OSS_CONDUCTOR_VERSION defaults to `latest` for local runs. CI resolves it from the +# E2E_TEST_OSS_CONDUCTOR_VERSION org variable (or a workflow_dispatch input); that variable is +# currently set to `latest` too, so CI tracks whatever `latest` resolves to at run time rather +# than a fixed version. Set the org variable to a real tag if the job needs to be deterministic. services: conductor-server: diff --git a/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java index bbfdb195e..104e13ca5 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/EventClientTests.java @@ -38,7 +38,7 @@ void testEventHandler() { } catch (ConductorClientException e) { // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server // we're running against actually reports it. - TestUtil.assertNotFoundOrRethrow(e, "not found"); + TestUtil.tolerateNotFound(e, "EventHandler with name"); } EventHandler eventHandler = getEventHandler(); eventClient.registerEventHandler(eventHandler); diff --git a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java index d7430a97b..8338aa67e 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/MetadataClientTests.java @@ -41,7 +41,7 @@ void taskDefinition() { } catch (ConductorClientException e) { // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server // we're running against actually reports it. - TestUtil.assertNotFoundOrRethrow(e, "No such task definition"); + TestUtil.tolerateNotFound(e, "No such task definition"); } TaskDef taskDef = Commons.getTaskDef(); metadataClient.registerTaskDefs(List.of(taskDef)); @@ -57,7 +57,7 @@ void workflow() { } catch (ConductorClientException e) { // Best-effort cleanup: tolerate "doesn't exist" in whichever shape the server // we're running against actually reports it. - TestUtil.assertNotFoundOrRethrow(e, "No such workflow definition"); + TestUtil.tolerateNotFound(e, "No such workflow definition"); } metadataClient.registerTaskDefs(List.of(Commons.getTaskDef())); WorkflowDef workflowDef = WorkflowUtil.getWorkflowDef(); diff --git a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java index 786e7e8a2..bbb77c76d 100644 --- a/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java +++ b/tests/src/test/java/io/orkes/conductor/client/util/TestUtil.java @@ -174,7 +174,7 @@ private static boolean isTerminalFailure(Workflow workflow) { * OSS with it deliberately unset, and OSS returning a correct 404 for one of these endpoints * should not start failing the suite. */ - public static void assertNotFoundOrRethrow(ConductorClientException e, String ossMessageSubstring) { + public static void tolerateNotFound(ConductorClientException e, String ossMessageSubstring) { if (e.getStatus() == 404) { return; } From 8104ceee221e991fff4155f01a41d499dac86140 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Tue, 8 Sep 2026 12:03:37 -0600 Subject: [PATCH 18/21] local run script can dump server logs if tests fail --- scripts/run-integration-oss.sh | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/scripts/run-integration-oss.sh b/scripts/run-integration-oss.sh index 1d0d27ac7..8f44f8e8b 100755 --- a/scripts/run-integration-oss.sh +++ b/scripts/run-integration-oss.sh @@ -51,6 +51,11 @@ cd "${REPO_ROOT}" compose() { docker compose -f "${COMPOSE_FILE}" "$@"; } cleanup() { + local status=$? + if [[ "${status}" -ne 0 ]]; then + echo "Dumping conductor-server logs (exit ${status})..." >&2 + compose logs conductor-server || true + fi if [[ "${KEEP_UP}" == "1" ]]; then echo "--keep-up set: leaving the OSS stack running. Tear down with:" echo " docker compose -f ${COMPOSE_FILE} down -v" @@ -79,7 +84,6 @@ deadline=$(( SECONDS + HEALTH_TIMEOUT )) until curl -sf http://localhost:8080/health >/dev/null 2>&1; do if (( SECONDS >= deadline )); then echo "Error: Conductor did not become healthy within ${HEALTH_TIMEOUT}s." >&2 - compose logs conductor-server || true exit 1 fi sleep 5 From 98486f3d8f5bd32fa0f5308514d9cb51df31d8b2 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Tue, 8 Sep 2026 12:16:38 -0600 Subject: [PATCH 19/21] skip a new test that won't work on oss --- .../java/io/orkes/conductor/client/http/TaskClientTests.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java b/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java index 19abbc1ae..bfc0bc0d0 100644 --- a/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java +++ b/tests/src/test/java/io/orkes/conductor/client/http/TaskClientTests.java @@ -383,6 +383,8 @@ void testSyncBlockingTask() throws Exception { } @Test + @DisabledIfEnvironmentVariable(named = "CONDUCTOR_SERVER_TYPE", matches = "oss", + disabledReason = "BLOCKING_TASK_LIST is not a value of the server-side WorkflowSignalReturnStrategy enum in any OSS Conductor release, so OSS rejects the returnStrategy parameter outright, on top of not implementing the signal/sync API the other BLOCKING_* tests are skipped for") void testSyncBlockingTaskList() throws Exception { String workflowId = startComplexWorkflow(Consistency.SYNCHRONOUS, ReturnStrategy.BLOCKING_TASK_LIST); From cebf93eac91b33b4cd4e28ec7e7637ca24a6e208 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Wed, 9 Sep 2026 11:45:23 -0600 Subject: [PATCH 20/21] improve e2e test oss conductor version selection to have a fallback for fork repo PRs which won't get access to the variable. --- .github/workflows/integration-tests.yml | 35 +++++++++++++++++++++---- 1 file changed, 30 insertions(+), 5 deletions(-) diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index 04234cc85..b70098599 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -8,7 +8,7 @@ on: workflow_dispatch: inputs: oss_conductor_version: - description: 'OSS Conductor image tag (falls back to E2E_TEST_OSS_CONDUCTOR_VERSION org var)' + description: 'OSS Conductor image tag (falls back to E2E_TEST_OSS_CONDUCTOR_VERSION org var, then to a pinned default on fork PRs)' required: false type: string @@ -99,16 +99,41 @@ jobs: env: CONDUCTOR_SERVER_URL: http://localhost:8080/api CONDUCTOR_SERVER_TYPE: oss - OSS_CONDUCTOR_VERSION: ${{ inputs.oss_conductor_version || vars.E2E_TEST_OSS_CONDUCTOR_VERSION }} + # Used only when the org variable is unreachable because the run is a fork + # PR -- see the resolve step below. + FORK_PR_FALLBACK_VERSION: '3.32.3' steps: - - name: Verify OSS Conductor version is set + # OSS_CONDUCTOR_VERSION is resolved here rather than in the job `env` so + # that the two ways it can come back empty get different treatment: + # + # - Fork PR: GitHub withholds org/repo variables from pull_request runs + # on forks exactly as it withholds secrets, so vars.* is always "" for + # an outside contributor (observed in csharp-sdk#178). This job needs + # no secrets, only a tag, so pin one and keep running. Note that the + # workflow_run trigger above executes in the base repo's context and + # does get the org variable even for a fork PR; the fork branch here + # matters for the push/dispatch paths and for any future direct + # pull_request trigger. + # - Anything else: the org variable is genuinely missing or its + # repository access policy no longer covers this repo. Fail loudly + # rather than silently drifting onto the pin. + - name: Resolve OSS Conductor version + env: + REQUESTED_VERSION: ${{ inputs.oss_conductor_version || vars.E2E_TEST_OSS_CONDUCTOR_VERSION }} + IS_FORK_PR: ${{ (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) || (github.event_name == 'workflow_run' && github.event.workflow_run.head_repository.full_name != github.repository) }} run: | - if [ -z "$OSS_CONDUCTOR_VERSION" ]; then + if [ -n "$REQUESTED_VERSION" ]; then + resolved="$REQUESTED_VERSION" + elif [ "$IS_FORK_PR" = "true" ]; then + resolved="$FORK_PR_FALLBACK_VERSION" + echo "::notice::Fork PR: org variables are withheld, pinning conductoross/conductor:${resolved}" + else echo "::error::No Conductor OSS image tag resolved. Set the E2E_TEST_OSS_CONDUCTOR_VERSION organization variable (and ensure its repository access policy includes this repo), or pass the oss_conductor_version input via workflow_dispatch." exit 1 fi - echo "Using conductoross/conductor:$OSS_CONDUCTOR_VERSION" + echo "OSS_CONDUCTOR_VERSION=${resolved}" >> "$GITHUB_ENV" + echo "Using conductoross/conductor:${resolved}" - name: Checkout uses: actions/checkout@v6 From 776ed5a9d2ef294f43cf8e5515994f4ff181b7c9 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 11 Sep 2026 09:35:12 -0600 Subject: [PATCH 21/21] ci(oss): write the image tag once, pull it in CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports javascript-sdk#176's resolution of the same review feedback, so the OSS harness stays diffable across the SDKs. csharp-sdk#171 carries the identical change. One tag, one home. The tag was written twice: FORK_PR_FALLBACK_VERSION in the integration-tests-oss job, and `latest` as the local script's default. A local run therefore could not reproduce a CI failure, which is most of why the script exists. Rather than teach one of them to read the other, make the `image:` line of docker-compose-oss.yaml the only place it is written and let both fall through to it: the script applies no default of its own and exports OSS_CONDUCTOR_VERSION only when --version actually supplied one, and the fork-PR branch of the resolve step now leaves the variable unset instead of pinning its own copy. Fork CI and a plain local run reach the identical image by the identical path, with no YAML parsing on either side and one hardcode removed rather than a mechanism added. The non-fork empty case still fails loudly, and the workflow_run nuance is untouched -- that trigger runs in the base repo's context and still gets the org variable even for a fork PR. The script's "Using ..." and "Pulling ..." lines now come from `compose config --images conductor-server` instead of reconstructing the tag, so they stay honest whichever source supplied it. The compose header already noted that E2E_TEST_OSS_CONDUCTOR_VERSION is set to `latest` org-wide, so a normal CI run is not really pinned. That stays, with the addition that this default is now what actually pins fork PRs and local runs until someone sets the org variable to a real tag. Pull in CI. The script pulls before `up`, CI did not. On a GitHub-hosted runner the VM is ephemeral and starts with no cached copy, so `up` pulls anyway and the step is redundant today — kept regardless, because it costs no extra network pull (`up` then finds the image locally), it splits "couldn't pull the image" from "the stack didn't come up" into two distinct red steps, and it is what stops a mutable tag going stale the day this job moves to a self-hosted runner with a warm Docker daemon. It also prints the tag in use, which matters now that a fork PR's tag is not spelled out in the workflow. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/integration-tests.yml | 35 ++++++++++++++++-------- scripts/docker-compose-oss.yaml | 16 +++++++---- scripts/run-integration-oss.sh | 36 +++++++++++++++++-------- 3 files changed, 60 insertions(+), 27 deletions(-) diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index b70098599..2d82efae7 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -8,7 +8,7 @@ on: workflow_dispatch: inputs: oss_conductor_version: - description: 'OSS Conductor image tag (falls back to E2E_TEST_OSS_CONDUCTOR_VERSION org var, then to a pinned default on fork PRs)' + description: 'OSS Conductor image tag (falls back to the E2E_TEST_OSS_CONDUCTOR_VERSION org var, then to the default in scripts/docker-compose-oss.yaml on fork PRs)' required: false type: string @@ -99,9 +99,6 @@ jobs: env: CONDUCTOR_SERVER_URL: http://localhost:8080/api CONDUCTOR_SERVER_TYPE: oss - # Used only when the org variable is unreachable because the run is a fork - # PR -- see the resolve step below. - FORK_PR_FALLBACK_VERSION: '3.32.3' steps: # OSS_CONDUCTOR_VERSION is resolved here rather than in the job `env` so @@ -110,30 +107,31 @@ jobs: # - Fork PR: GitHub withholds org/repo variables from pull_request runs # on forks exactly as it withholds secrets, so vars.* is always "" for # an outside contributor (observed in csharp-sdk#178). This job needs - # no secrets, only a tag, so pin one and keep running. Note that the + # no secrets, only a tag, so leave the var unset and let the default + # baked into the `image:` line of scripts/docker-compose-oss.yaml + # apply. That is the same tag a plain local run of + # scripts/run-integration-oss.sh gets, and the one place it is + # written -- no second copy to drift out of sync here. Note that the # workflow_run trigger above executes in the base repo's context and # does get the org variable even for a fork PR; the fork branch here # matters for the push/dispatch paths and for any future direct # pull_request trigger. # - Anything else: the org variable is genuinely missing or its # repository access policy no longer covers this repo. Fail loudly - # rather than silently drifting onto the pin. + # rather than silently drifting onto the default. - name: Resolve OSS Conductor version env: REQUESTED_VERSION: ${{ inputs.oss_conductor_version || vars.E2E_TEST_OSS_CONDUCTOR_VERSION }} IS_FORK_PR: ${{ (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository) || (github.event_name == 'workflow_run' && github.event.workflow_run.head_repository.full_name != github.repository) }} run: | if [ -n "$REQUESTED_VERSION" ]; then - resolved="$REQUESTED_VERSION" + echo "OSS_CONDUCTOR_VERSION=${REQUESTED_VERSION}" >> "$GITHUB_ENV" elif [ "$IS_FORK_PR" = "true" ]; then - resolved="$FORK_PR_FALLBACK_VERSION" - echo "::notice::Fork PR: org variables are withheld, pinning conductoross/conductor:${resolved}" + echo "::notice::Fork PR: org variables are withheld, falling back to the default tag in scripts/docker-compose-oss.yaml" else echo "::error::No Conductor OSS image tag resolved. Set the E2E_TEST_OSS_CONDUCTOR_VERSION organization variable (and ensure its repository access policy includes this repo), or pass the oss_conductor_version input via workflow_dispatch." exit 1 fi - echo "OSS_CONDUCTOR_VERSION=${resolved}" >> "$GITHUB_ENV" - echo "Using conductoross/conductor:${resolved}" - name: Checkout uses: actions/checkout@v6 @@ -149,6 +147,21 @@ jobs: distribution: "zulu" java-version: "21" + # `docker compose up` only pulls an image when it is missing locally. On a + # GitHub-hosted runner the VM is ephemeral and starts with no cached copy + # of this image, so `up` would pull anyway and this step is redundant + # today. It is here deliberately: it costs no extra network pull (`up` + # then finds the image locally), it separates "couldn't pull the image" + # from "the stack didn't come up" into two distinct red steps, and it is + # what keeps a mutable tag from going stale if this job ever moves to a + # self-hosted runner with a warm Docker daemon -- the same reason + # scripts/run-integration-oss.sh pulls. It also prints the tag actually in + # use, which for a fork PR comes from the compose file's default. + - name: Pull Conductor OSS image + run: | + echo "Using $(docker compose -f scripts/docker-compose-oss.yaml config --images | grep -m1 '^conductoross/conductor:')" + docker compose -f scripts/docker-compose-oss.yaml pull conductor-server + - name: Start Conductor OSS stack run: docker compose -f scripts/docker-compose-oss.yaml up -d diff --git a/scripts/docker-compose-oss.yaml b/scripts/docker-compose-oss.yaml index cb1f36f9b..a30a91e7c 100644 --- a/scripts/docker-compose-oss.yaml +++ b/scripts/docker-compose-oss.yaml @@ -2,14 +2,20 @@ # Shared by scripts/run-integration-oss.sh and the integration-tests-oss job in # .github/workflows/integration-tests.yml. # -# OSS_CONDUCTOR_VERSION defaults to `latest` for local runs. CI resolves it from the -# E2E_TEST_OSS_CONDUCTOR_VERSION org variable (or a workflow_dispatch input); that variable is -# currently set to `latest` too, so CI tracks whatever `latest` resolves to at run time rather -# than a fixed version. Set the org variable to a real tag if the job needs to be deterministic. +# The `image:` default below is the SINGLE place the Conductor OSS image tag is written. +# Everything that does not override OSS_CONDUCTOR_VERSION lands on it: a plain +# scripts/run-integration-oss.sh run, and the integration-tests-oss job on a fork PR (where +# GitHub withholds org variables). Overrides are the script's --version flag and, in CI, the +# E2E_TEST_OSS_CONDUCTOR_VERSION org variable or a workflow_dispatch input. Bump the tag here +# and both follow. +# +# Note that the org variable is currently set to `latest`, so a normal CI run still tracks +# whatever `latest` resolves to at run time. Set it to a real tag if the job needs to be +# deterministic; until then this default is what actually pins fork PRs and local runs. services: conductor-server: - image: conductoross/conductor:${OSS_CONDUCTOR_VERSION:-latest} + image: conductoross/conductor:${OSS_CONDUCTOR_VERSION:-3.32.3} environment: - CONFIG_PROP=config-postgres.properties ports: diff --git a/scripts/run-integration-oss.sh b/scripts/run-integration-oss.sh index 8f44f8e8b..8b624b1d2 100755 --- a/scripts/run-integration-oss.sh +++ b/scripts/run-integration-oss.sh @@ -9,15 +9,17 @@ # the empirically-confirmed gaps). # # The stack (Conductor OSS + Postgres) is defined in -# scripts/docker-compose-oss.yaml and is torn down automatically on exit. The -# image is always pulled before starting, since `latest` (the local default) -# is a mutable tag and a cached copy would otherwise go stale silently. +# scripts/docker-compose-oss.yaml and is torn down automatically on exit. That +# file's `image:` line is also where the default tag lives -- this script +# applies no default of its own, so a plain run and a fork-PR CI run land on +# the identical image. The image is always pulled before starting, since a tag +# can be mutable and a cached copy would otherwise go stale silently. # # Usage: # scripts/run-integration-oss.sh [--keep-up] [--version ] [--include-gated] [-- gradle args] # Examples: -# scripts/run-integration-oss.sh -# scripts/run-integration-oss.sh --version 3.32.0-rc18 +# scripts/run-integration-oss.sh # default tag from the compose file +# scripts/run-integration-oss.sh --version 3.33.0-rc1 # scripts/run-integration-oss.sh --keep-up # scripts/run-integration-oss.sh --include-gated # also run tests normally skipped as Orkes-only # scripts/run-integration-oss.sh -- --tests "*WorkflowClientTests" @@ -41,7 +43,14 @@ while [[ $# -gt 0 ]]; do esac done -export OSS_CONDUCTOR_VERSION="${OSS_CONDUCTOR_VERSION:-latest}" +# No default is applied here on purpose. The default tag is written once, in the +# `image:` line of scripts/docker-compose-oss.yaml, so leaving OSS_CONDUCTOR_VERSION +# unset lets compose supply it -- the same path a fork PR takes in CI. Only export +# it when the caller actually asked for a specific tag, otherwise a value set but +# not exported in the caller's shell would never reach compose anyway. +if [[ -n "${OSS_CONDUCTOR_VERSION:-}" ]]; then + export OSS_CONDUCTOR_VERSION +fi SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" @@ -66,13 +75,18 @@ cleanup() { } trap cleanup EXIT -echo "Using conductoross/conductor:${OSS_CONDUCTOR_VERSION}" +# Ask compose what it resolved rather than reconstructing the tag here, so this +# stays correct whether the tag came from --version or from the compose default. +# `--images` lists every service's image and does not reliably honour a service +# filter, so select the server's by name rather than by position. +SERVER_IMAGE="$(compose config --images | grep -m1 '^conductoross/conductor:')" +echo "Using ${SERVER_IMAGE}" # `docker compose up` only pulls an image when it is missing locally, so a -# previously-cached `latest` (or any other mutable tag) would silently be -# reused instead of getting the current version. Pull unconditionally so the -# stack always reflects the tag we just printed. -echo "Pulling conductoross/conductor:${OSS_CONDUCTOR_VERSION} to ensure it's current..." +# previously-cached mutable tag (a re-pushed rc, or `latest` if that is what was +# asked for) would silently be reused instead of getting the current version. +# Pull unconditionally so the stack always reflects the tag we just printed. +echo "Pulling ${SERVER_IMAGE} to ensure it's current..." compose pull conductor-server echo "Starting Conductor OSS stack..."