CCOR-13193 - run integration tests against Conductor OSS, and fix two client bugs it surfaced

.github/workflows/integration-tests.yml

@@ -5,14 +5,36 @@ 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
+
+  # ---------------------------------------------------------------------------
+  # 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:
@@ -44,7 +66,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 +86,90 @@ 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
+    # 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
+      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 }}'
+            });
 

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 `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.
+
+```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

conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/WorkflowExecutor.java

@@ -174,10 +174,21 @@ private void initMonitor() {
                     for (Map.Entry<String, CompletableFuture<Workflow>> entry : runningWorkflowFutures.entrySet()) {
                         String workflowId = entry.getKey();
                         CompletableFuture<Workflow> future = entry.getValue();
-                        Workflow workflow = workflowClient.getWorkflow(workflowId, true);
-                        if (workflow.getStatus().isTerminal()) {
-                            future.complete(workflow);
+                        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 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);
                         }
                     }
                 },

conductor-client/src/main/java/com/netflix/conductor/sdk/workflow/executor/task/AnnotatedWorkerExecutor.java

@@ -52,6 +52,12 @@ public class AnnotatedWorkerExecutor {
 
     private final Set<String> 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) {
@@ -77,6 +83,9 @@ public AnnotatedWorkerExecutor(TaskClient taskClient, WorkerConfiguration worker
      */
     public synchronized void initWorkers(String... basePackages) {
         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();
     }
 
@@ -109,10 +118,18 @@ public synchronized void initWorkersFromClasses(List<? extends Class<?>> classes
     }
 
 
-    /** Shuts down the workers */
-    public void shutdown() {
+    /**
+     * Shuts down the workers.
+     *
+     * <p>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;
         }
     }
 
@@ -168,7 +185,15 @@ private boolean classBelongsToPackage(List<String> packagesToScan, String classN
         return false;
     }
 
-    public void addBean(Object bean) {
+    /**
+     * Registers every {@link WorkerTask}-annotated method on the bean as a worker.
+     *
+     * <p>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);
@@ -222,6 +247,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",
@@ -231,11 +257,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.
+     *
+     * <p>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);
@@ -253,6 +296,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();