From c5d487af8aa52d9c5ffce1c30c8c310a768e7eb1 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Thu, 13 Aug 2026 12:37:33 -0600 Subject: [PATCH 01/17] 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 4efb61f59..ef13c06d6 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); @@ -331,6 +334,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); @@ -345,6 +350,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); @@ -359,6 +366,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); @@ -373,6 +382,8 @@ void testSyncBlockingTask() 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); @@ -390,7 +401,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); @@ -405,7 +419,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); @@ -420,6 +437,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); @@ -434,7 +453,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); @@ -449,6 +471,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); @@ -727,6 +751,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); @@ -763,6 +789,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 37da691ef7453d16d75c1e691a1d932a52d99a9c Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Thu, 13 Aug 2026 12:50:00 -0600 Subject: [PATCH 02/17] 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 d8fca483bf96c84097e8746d79a6b94044bad80a Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 09:34:09 -0600 Subject: [PATCH 03/17] 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 a8f1409801d492f0c0e05f368f27da71ffdab3de Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 10:03:32 -0600 Subject: [PATCH 04/17] 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 56cb1df2189b3ca5ae69405be1d660d3d6f007a3 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 10:21:08 -0600 Subject: [PATCH 05/17] 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 427d5d11b2706c402af1db56b14921fb47dfb9c6 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 12:23:58 -0600 Subject: [PATCH 06/17] 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 22fb9d23ed8015a995dabac3ae5952630b7adae5 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 12:41:39 -0600 Subject: [PATCH 07/17] 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 14cc5a8f18a636f5ab3d8b371229671724abd40d Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 14 Aug 2026 12:48:44 -0600 Subject: [PATCH 08/17] 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 9b662598b7a3bd244328ce42deeacbcc725611b0 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Mon, 17 Aug 2026 09:38:20 -0600 Subject: [PATCH 09/17] 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 c49f5477fb9fcd0c17f7dec2feed312eef59a963 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Mon, 17 Aug 2026 10:10:46 -0600 Subject: [PATCH 10/17] 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 df2275e361b0db7a6c5debec09f180ea13d1d877 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 11:41:07 -0600 Subject: [PATCH 11/17] 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 fb2b3f5fc81f47bdf1d8b7bfeecf0e1736f7a523 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 11:41:28 -0600 Subject: [PATCH 12/17] 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 7727fcde82a282a2322e8f07e9e5a20ee061d451 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 11:41:28 -0600 Subject: [PATCH 13/17] 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 0d8bfe7a5cf67acaef72706946bf415b2f9f6952 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 12:40:17 -0600 Subject: [PATCH 14/17] 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 bd59e292d07a71923bf9443efbc895f0c20a9505 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 13:31:20 -0600 Subject: [PATCH 15/17] 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 f1fcf0f5f647dc7c06c207d272c1e40d0d80dddf Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 28 Aug 2026 13:34:31 -0600 Subject: [PATCH 16/17] 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 0d55ef496dedda2f2dc4f5f625860da8cd8bbefb Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Mon, 31 Aug 2026 10:20:46 -0600 Subject: [PATCH 17/17] 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; }