wehub/conductor-oss--conductor
1
0
Fork 0
Code Issues Pull Requests Actions Deployments Models Agent Notes Packages Projects Releases Wiki Activity

chore: import upstream snapshot with attribution
CI / unit-test (push) Has been cancelled
Details
CI / detect-changes (push) Has been cancelled
Details
CI / build (push) Has been cancelled
Details
Publish docs via GitHub Pages / Deploy docs (push) Has been cancelled
Details
CI / test-harness (push) Has been cancelled
Details
CI / generate-e2e-matrix (push) Has been cancelled
Details
CI / e2e (push) Has been cancelled
Details
CI / build-ui (push) Has been cancelled
Details
Release Drafter / update_release_draft (push) Has been cancelled
Details
UI v2 Integration CI / E2E (Integration) (push) Has been cancelled
Details
UI v2 CI / Lint, Format & Test (push) Has been cancelled
Details
UI v2 CI / E2E (Mocked) (push) Has been cancelled
Details

Browse Source
This commit is contained in:
wehub-resource-sync
2026-07-13 12:37:56 +08:00
commit a9cd7750f4
3727 changed files with 627706 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/documentation/api/bulk.md
+184
View File
@@ -0,0 +1,184 @@
---
description: "Conductor Bulk Operations API — pause, resume, restart, retry, terminate, remove, and search workflows in batch."
---
# Bulk Operations API
The Bulk Operations API lets you perform workflow management operations on multiple workflows in a single request. All endpoints use the base path `/api/workflow/bulk`.
Every endpoint accepts a list of workflow IDs in the request body and returns a `BulkResponse`:
```json
{
"bulkSuccessfulResults": ["workflow-id-1", "workflow-id-2"],
"bulkErrorResults": {
"workflow-id-3": "Workflow is not in a running state"
}
}
```
Operations are **best-effort** — each workflow is processed independently. If one fails, the rest still proceed.
## Endpoints
| Endpoint | Method | Description |
|---|---|---|
| `/bulk/pause` | `PUT` | Pause multiple workflows |
| `/bulk/resume` | `PUT` | Resume multiple paused workflows |
| `/bulk/restart` | `POST` | Restart multiple completed workflows |
| `/bulk/retry` | `POST` | Retry the last failed task in multiple workflows |
| `/bulk/terminate` | `POST` | Terminate multiple running workflows |
| `/bulk/remove` | `DELETE` | Remove multiple workflows from the system |
| `/bulk/terminate-remove` | `DELETE` | Terminate and remove multiple workflows |
| `/bulk/search` | `POST` | Search/fetch multiple workflows by ID |
### Bulk Pause
```
PUT /api/workflow/bulk/pause
```
```shell
curl -X PUT 'http://localhost:8080/api/workflow/bulk/pause' \
-H 'Content-Type: application/json' \
-d '["workflow-id-1", "workflow-id-2", "workflow-id-3"]'
```
**Response** `200 OK`
```json
{
"bulkSuccessfulResults": ["workflow-id-1", "workflow-id-2"],
"bulkErrorResults": {
"workflow-id-3": "Workflow is already paused"
}
}
```
### Bulk Resume
```
PUT /api/workflow/bulk/resume
```
```shell
curl -X PUT 'http://localhost:8080/api/workflow/bulk/resume' \
-H 'Content-Type: application/json' \
-d '["workflow-id-1", "workflow-id-2"]'
```
**Response** `200 OK` — returns a `BulkResponse`.
### Bulk Restart
```
POST /api/workflow/bulk/restart?useLatestDefinitions=false
```
| Parameter | Description | Default |
|---|---|---|
| `useLatestDefinitions` | Use latest workflow and task definitions | `false` |
```shell
curl -X POST 'http://localhost:8080/api/workflow/bulk/restart?useLatestDefinitions=true' \
-H 'Content-Type: application/json' \
-d '["workflow-id-1", "workflow-id-2"]'
```
**Response** `200 OK` — returns a `BulkResponse`.
### Bulk Retry
```
POST /api/workflow/bulk/retry
```
Retries the last failed task for each workflow.
```shell
curl -X POST 'http://localhost:8080/api/workflow/bulk/retry' \
-H 'Content-Type: application/json' \
-d '["workflow-id-1", "workflow-id-2"]'
```
**Response** `200 OK` — returns a `BulkResponse`.
### Bulk Terminate
```
POST /api/workflow/bulk/terminate?reason=
```
| Parameter | Description | Required |
|---|---|---|
| `reason` | Reason for termination | No |
```shell
curl -X POST 'http://localhost:8080/api/workflow/bulk/terminate?reason=batch+cleanup' \
-H 'Content-Type: application/json' \
-d '["workflow-id-1", "workflow-id-2", "workflow-id-3"]'
```
**Response** `200 OK` — returns a `BulkResponse`.
### Bulk Remove
```
DELETE /api/workflow/bulk/remove?archiveWorkflow=true
```
| Parameter | Description | Default |
|---|---|---|
| `archiveWorkflow` | Archive before removing | `true` |
```shell
curl -X DELETE 'http://localhost:8080/api/workflow/bulk/remove' \
-H 'Content-Type: application/json' \
-d '["workflow-id-1", "workflow-id-2"]'
```
!!! warning
This permanently removes workflow execution data.
**Response** `200 OK` — returns a `BulkResponse`.
### Bulk Terminate and Remove
```
DELETE /api/workflow/bulk/terminate-remove?reason=&archiveWorkflow=true
```
Terminates running workflows and removes them in one call.
| Parameter | Description | Default |
|---|---|---|
| `reason` | Reason for termination | — |
| `archiveWorkflow` | Archive before removing | `true` |
```shell
curl -X DELETE 'http://localhost:8080/api/workflow/bulk/terminate-remove?reason=decommissioned' \
-H 'Content-Type: application/json' \
-d '["workflow-id-1", "workflow-id-2"]'
```
**Response** `200 OK` — returns a `BulkResponse`.
### Bulk Search
```
POST /api/workflow/bulk/search?includeTasks=true
```
Fetches multiple workflows by their IDs in a single call. Unlike the other bulk endpoints, this returns workflow objects rather than a `BulkResponse`.
| Parameter | Description | Default |
|---|---|---|
| `includeTasks` | Include task details | `true` |
```shell
curl -X POST 'http://localhost:8080/api/workflow/bulk/search?includeTasks=false' \
-H 'Content-Type: application/json' \
-d '["workflow-id-1", "workflow-id-2"]'
```
**Response** `200 OK` — returns a `BulkResponse` where `bulkSuccessfulResults` contains the full workflow objects.
docs/documentation/api/eventhandlers.md
+212
View File
@@ -0,0 +1,212 @@
---
description: "Conductor Event Handlers API — create, update, delete, and list event handlers for event-driven workflow orchestration."
---
# Event Handlers API
The Event Handlers API manages event handler definitions — rules that start workflows or complete tasks in response to events from message brokers (Kafka, NATS, SQS, AMQP). All endpoints use the base path `/api/event`.
For details on configuring event handlers, see [Event Handler Configuration](../configuration/eventhandlers.md). For configuring message broker connections, see the [Event Bus Orchestration](../../devguide/how-tos/event-bus.md) guide.
## Endpoints
| Endpoint | Method | Description |
|---|---|---|
| `/event` | `POST` | Create a new event handler |
| `/event` | `PUT` | Update an existing event handler |
| `/event` | `GET` | Get all event handlers |
| `/event/{name}` | `DELETE` | Delete an event handler |
| `/event/{event}` | `GET` | Get event handlers for a specific event |
### Create an Event Handler
```
POST /api/event
```
```shell
curl -X POST 'http://localhost:8080/api/event' \
-H 'Content-Type: application/json' \
-d '{
"name": "order_event_handler",
"event": "kafka:orders_topic:new_order",
"active": true,
"actions": [
{
"action": "start_workflow",
"start_workflow": {
"name": "order_processing",
"version": 1,
"input": {
"orderId": "${eventPayload.orderId}",
"customerId": "${eventPayload.customerId}",
"payload": "${eventPayload}"
}
}
}
]
}'
```
**Response** `200 OK` — no response body.
#### Event Handler Fields
| Field | Description | Required |
|---|---|---|
| `name` | Unique name for the event handler | Yes |
| `event` | Event identifier in format `type:queue:subject` (e.g., `kafka:my_topic:my_event`) | Yes |
| `active` | Whether the handler is active | Yes |
| `actions` | List of actions to execute when the event is received | Yes |
| `condition` | Optional JavaScript expression to filter events | No |
| `evaluatorType` | Expression evaluator type (`javascript` or `graaljs`) | No |
#### Action Types
| Action | Description |
|---|---|
| `start_workflow` | Start a new workflow execution |
| `complete_task` | Complete a pending task (e.g., a WAIT task) |
| `fail_task` | Fail a pending task |
#### Complete Task Action Example
```json
{
"name": "approval_handler",
"event": "kafka:approvals_topic:approved",
"active": true,
"actions": [
{
"action": "complete_task",
"complete_task": {
"workflowId": "${eventPayload.workflowId}",
"taskRefName": "wait_for_approval",
"output": {
"approved": true,
"approvedBy": "${eventPayload.approver}"
}
}
}
]
}
```
### Update an Event Handler
```
PUT /api/event
```
Updates an existing event handler. The request body is the full event handler definition (same format as create).
```shell
curl -X PUT 'http://localhost:8080/api/event' \
-H 'Content-Type: application/json' \
-d '{
"name": "order_event_handler",
"event": "kafka:orders_topic:new_order",
"active": false,
"actions": [
{
"action": "start_workflow",
"start_workflow": {
"name": "order_processing",
"version": 2,
"input": {
"payload": "${eventPayload}"
}
}
}
]
}'
```
**Response** `200 OK` — no response body.
### Get All Event Handlers
```
GET /api/event
```
Returns a list of all registered event handlers.
```shell
curl 'http://localhost:8080/api/event'
```
**Response** `200 OK`
```json
[
{
"name": "order_event_handler",
"event": "kafka:orders_topic:new_order",
"active": true,
"actions": [
{
"action": "start_workflow",
"start_workflow": {
"name": "order_processing",
"version": 1,
"input": {
"payload": "${eventPayload}"
}
}
}
]
}
]
```
### Delete an Event Handler
```
DELETE /api/event/{name}
```
Removes an event handler by name.
```shell
curl -X DELETE 'http://localhost:8080/api/event/order_event_handler'
```
**Response** `200 OK` — no response body.
### Get Event Handlers for an Event
```
GET /api/event/{event}?activeOnly=true
```
Returns event handlers configured for a specific event.
| Parameter | Description | Default |
|---|---|---|
| `event` | Event identifier (e.g., `kafka:orders_topic:new_order`) | — |
| `activeOnly` | Only return active handlers | `true` |
```shell
curl 'http://localhost:8080/api/event/kafka:orders_topic:new_order?activeOnly=true'
```
**Response** `200 OK` — returns a list of matching event handler definitions.
---
## Event Identifier Format
Event identifiers follow the pattern:
```
{type}:{queue/topic}:{subject}
```
| Type | Example | Description |
|---|---|---|
| `kafka` | `kafka:my_topic:my_event` | Apache Kafka topic |
| `nats` | `nats:my_subject:my_event` | NATS subject |
| `sqs` | `sqs:my_queue:my_event` | Amazon SQS queue |
| `amqp_exchange` | `amqp_exchange:my_exchange:my_event` | RabbitMQ exchange |
| `conductor` | `conductor:my_event:my_event` | Conductor internal event queue |
docs/documentation/api/files.md
+256
View File
@@ -0,0 +1,256 @@
---
description: "Conductor File API — create, upload, download, and inspect file payloads. Includes single-shot and multipart presigned URL flows."
---
# File API
The File API manages binary file payloads associated with workflow executions. All endpoints use the base path `/api/files` and are gated by `conductor.file-storage.enabled=true` — when the feature is disabled, every endpoint returns `404`. See [File Storage](../advanced/file-storage.md) for backend setup.
Path variables carry the bare `fileId` (a UUID assigned at creation). Request and response bodies carry the prefixed handle as `fileHandleId` (`conductor://file/<fileId>`); pass either form back in to subsequent calls — the server normalizes them.
Download access is *workflow-family scoped*: callers must supply a `workflowId` that belongs to the same workflow family (self, ancestors, or descendants) as the file's owning workflow. Cross-family access returns `403 Forbidden`.
## Create a File
```
POST /api/files
```
Reserves a `fileId`, persists the metadata record, and returns a presigned upload URL. The file is created in `UPLOADING` status; the client must confirm completion via [Confirm Upload](#confirm-upload) once the bytes are uploaded.
**Request body:**
| Field | Description | Required |
|---|---|---|
| `workflowId` | Workflow execution that owns this file. Used for download-access scoping. | Yes |
| `fileName` | Original file name. | No |
| `contentType` | MIME type. | No |
| `taskId` | Task that produced the file, if applicable. | No |
```shell
curl -X POST 'http://localhost:8080/api/files' \
-H 'Content-Type: application/json' \
-d '{
"workflowId": "3a5b8c2d-1234-5678-9abc-def012345678",
"fileName": "input.mp4",
"contentType": "video/mp4"
}'
```
**Response** `201 Created`
```json
{
"fileHandleId": "conductor://file/a1b2c3d4-5678-90ab-cdef-111111111111",
"fileName": "input.mp4",
"contentType": "video/mp4",
"storageType": "S3",
"uploadStatus": "UPLOADING",
"uploadUrl": "https://bucket.s3.amazonaws.com/conductor/3a5b8c2d.../a1b2c3d4...?X-Amz-Signature=...",
"uploadUrlExpiresAt": 1700000060000,
"createdAt": 1700000000000
}
```
The client `PUT`s file bytes directly to `uploadUrl`, then calls [Confirm Upload](#confirm-upload).
---
## Get a Fresh Upload URL
```
GET /api/files/{fileId}/upload-url
```
Issues a new presigned upload URL — used to retry an upload after the original URL expired.
```shell
curl 'http://localhost:8080/api/files/a1b2c3d4-5678-90ab-cdef-111111111111/upload-url'
```
**Response** `200 OK`
```json
{
"fileHandleId": "conductor://file/a1b2c3d4-5678-90ab-cdef-111111111111",
"uploadUrl": "https://bucket.s3.amazonaws.com/conductor/3a5b8c2d.../a1b2c3d4...?X-Amz-Signature=...",
"expiresAt": 1700000060000
}
```
---
## Confirm Upload
```
POST /api/files/{fileId}/upload-complete
```
Marks the upload as complete. The server probes the storage backend to verify the object exists, then transitions the record to `UPLOADED` and records the backend-reported content hash and size.
```shell
curl -X POST 'http://localhost:8080/api/files/a1b2c3d4-5678-90ab-cdef-111111111111/upload-complete'
```
**Response** `200 OK`
```json
{
"fileHandleId": "conductor://file/a1b2c3d4-5678-90ab-cdef-111111111111",
"uploadStatus": "UPLOADED",
"contentHash": "d41d8cd98f00b204e9800998ecf8427e"
}
```
`409 Conflict` if the file is already in `UPLOADED` status; `500 Internal Server Error` if the backend reports the object is missing.
---
## Get a Download URL
```
GET /api/files/{workflowId}/{fileId}/download-url
```
Issues a presigned download URL. The caller's `workflowId` must be in the same workflow family as the file's owning workflow.
| Parameter | Description |
|---|---|
| `workflowId` | The caller's workflow ID, used for family-scope check. |
| `fileId` | The file to download. |
```shell
curl 'http://localhost:8080/api/files/3a5b8c2d-1234-5678-9abc-def012345678/a1b2c3d4-5678-90ab-cdef-111111111111/download-url'
```
**Response** `200 OK`
```json
{
"fileHandleId": "conductor://file/a1b2c3d4-5678-90ab-cdef-111111111111",
"downloadUrl": "https://bucket.s3.amazonaws.com/conductor/3a5b8c2d.../a1b2c3d4...?X-Amz-Signature=...",
"expiresAt": 1700000060000
}
```
`403 Forbidden` if the caller's workflow is not in the file's family. `400 Bad Request` if the file is not yet `UPLOADED`.
---
## Get File Metadata
```
GET /api/files/{fileId}
```
Returns the file metadata record. Does not expose the server-internal storage path.
```shell
curl 'http://localhost:8080/api/files/a1b2c3d4-5678-90ab-cdef-111111111111'
```
**Response** `200 OK`
```json
{
"fileHandleId": "conductor://file/a1b2c3d4-5678-90ab-cdef-111111111111",
"fileName": "input.mp4",
"contentType": "video/mp4",
"contentHash": "d41d8cd98f00b204e9800998ecf8427e",
"storageType": "S3",
"uploadStatus": "UPLOADED",
"workflowId": "3a5b8c2d-1234-5678-9abc-def012345678",
"taskId": "task-uuid-1",
"createdAt": 1700000000000,
"updatedAt": 1700000005000
}
```
---
## Multipart Upload
Multipart is opt-in (typically for files above ~100 MB) and supported on `s3`, `azure-blob`, and `gcs` backends. The `local` backend does not support multipart.
### Initiate Multipart
```
POST /api/files/{fileId}/multipart
```
Begins a multipart upload session. Returns a backend-specific `uploadId`. For GCS/Azure, `uploadUrl` is the resumable session URL clients upload parts to directly; for S3 it is `null` and clients fetch a per-part URL via [Get Part Upload URL](#get-part-upload-url).
```shell
curl -X POST 'http://localhost:8080/api/files/a1b2c3d4-5678-90ab-cdef-111111111111/multipart'
```
**Response** `200 OK`
```json
{
"fileHandleId": "conductor://file/a1b2c3d4-5678-90ab-cdef-111111111111",
"uploadId": "S3-multipart-upload-id-string",
"uploadUrl": null
}
```
### Get Part Upload URL
```
GET /api/files/{fileId}/multipart/{uploadId}/part/{partNumber}
```
S3 only. Returns a presigned URL for uploading a single part. Part numbers start at `1`.
```shell
curl 'http://localhost:8080/api/files/a1b2c3d4-5678-90ab-cdef-111111111111/multipart/S3-multipart-upload-id-string/part/1'
```
**Response** `200 OK`
```json
{
"fileHandleId": "conductor://file/a1b2c3d4-5678-90ab-cdef-111111111111",
"uploadUrl": "https://bucket.s3.amazonaws.com/conductor/.../?partNumber=1&uploadId=...&X-Amz-Signature=...",
"expiresAt": 1700000060000
}
```
### Complete Multipart
```
POST /api/files/{fileId}/multipart/{uploadId}/complete
```
Finalizes the multipart upload and transitions the file to `UPLOADED`. The request body carries the ordered list of part ETags (or backend equivalents) from the part uploads.
```shell
curl -X POST 'http://localhost:8080/api/files/a1b2c3d4-5678-90ab-cdef-111111111111/multipart/S3-multipart-upload-id-string/complete' \
-H 'Content-Type: application/json' \
-d '{
"partETags": ["\"etag-of-part-1\"", "\"etag-of-part-2\"", "\"etag-of-part-3\""]
}'
```
**Response** `200 OK`
```json
{
"fileHandleId": "conductor://file/a1b2c3d4-5678-90ab-cdef-111111111111",
"uploadStatus": "UPLOADED",
"contentHash": "d41d8cd98f00b204e9800998ecf8427e"
}
```
---
## Errors
| Status | Cause |
|---|---|
| `400 Bad Request` | Missing `workflowId` on create; download requested before file is `UPLOADED`. |
| `403 Forbidden` | Caller's `workflowId` is not in the file's workflow family. |
| `404 Not Found` | Unknown `fileId`, or `conductor.file-storage.enabled=false`. |
| `409 Conflict` | Confirm-upload called on a file already in `UPLOADED` status. |
| `413 Payload Too Large` | `FileStorageException` raised by a backend (e.g., upstream size enforcement). |
| `500 Internal Server Error` | Backend reports object missing on confirm/complete; other transient/non-transient backend errors. |
docs/documentation/api/index.md
+116
View File
@@ -0,0 +1,116 @@
---
description: "Conductor REST API reference — complete endpoint documentation for workflow orchestration including metadata, execution management, task polling, bulk operations, and event handlers."
---
# API Reference
Conductor exposes a full REST API for managing workflow definitions, executions, tasks, and events.
## Base URL
All API endpoints are relative to your Conductor server's base URL:
```
http://localhost:8080/api/
```
For example, to list all workflow definitions:
```shell
curl http://localhost:8080/api/metadata/workflow
```
If your Conductor server runs on a different host or port, replace `localhost:8080` accordingly.
## Authentication
Conductor OSS does not require authentication by default. All API endpoints are open. If you need to secure your Conductor instance, you can add authentication via a reverse proxy (e.g., Nginx, Envoy) or by implementing a custom security filter in Spring Boot.
## Content Type
All request and response bodies use JSON. Set the following headers on requests with a body:
```
Content-Type: application/json
```
A few endpoints return plain text (e.g., workflow ID on start). These are noted in their documentation.
## Common Response Codes
| Status Code | Description |
|---|---|
| `200 OK` | Request succeeded. Response body contains the result. |
| `204 No Content` | Request succeeded but there is no response body (e.g., poll with no tasks available). |
| `400 Bad Request` | Invalid request — check your request body or parameters. |
| `404 Not Found` | The requested resource (workflow, task, definition) does not exist. |
| `409 Conflict` | Conflict with current state (e.g., trying to resume a workflow that is not paused). |
| `500 Internal Server Error` | Server-side error. Check Conductor server logs. |
### Error Response Format
When an error occurs, the response body contains:
```json
{
"status": 400,
"message": "Workflow definition is not valid",
"instance": "conductor-server",
"retryable": false
}
```
## Quick Start
Register a workflow definition, start it, and check its status — all in three commands:
```shell
# 1. Register a workflow definition
curl -X POST 'http://localhost:8080/api/metadata/workflow' \
-H 'Content-Type: application/json' \
-d '{
"name": "hello_workflow",
"version": 1,
"tasks": [
{
"name": "hello_task",
"taskReferenceName": "hello_ref",
"type": "HTTP",
"inputParameters": {
"uri": "https://jsonplaceholder.typicode.com/posts/1",
"method": "GET"
}
}
],
"schemaVersion": 2
}'
# 2. Start a workflow execution
WORKFLOW_ID=$(curl -s -X POST 'http://localhost:8080/api/workflow/hello_workflow' \
-H 'Content-Type: application/json' \
-d '{}')
echo "Started workflow: $WORKFLOW_ID"
# 3. Check workflow status
curl "http://localhost:8080/api/workflow/$WORKFLOW_ID"
```
## API Sections
| Section | Base Path | Description |
|---|---|---|
| **[Metadata](metadata.md)** | `/api/metadata` | Register, update, validate, and delete workflow and task definitions |
| **[Start Workflow](startworkflow.md)** | `/api/workflow` | Start workflows asynchronously, synchronously, or with dynamic definitions |
| **[Workflow](workflow.md)** | `/api/workflow` | Manage executions: get status, pause, resume, retry, restart, terminate, search |
| **[Task](task.md)** | `/api/tasks` | Poll for tasks, update results, manage queues, view logs, search |
| **[Bulk Operations](bulk.md)** | `/api/workflow/bulk` | Pause, resume, restart, retry, terminate, or remove workflows in batch |
| **[Event Handlers](eventhandlers.md)** | `/api/event` | Create and manage event-driven workflow triggers |
| **[Task Domains](taskdomains.md)** | — | Route tasks to specific worker pools at runtime |
## Swagger UI
The Swagger UI at `http://localhost:8080/swagger-ui/index.html` provides an interactive API explorer where you can try endpoints directly from your browser.
## SDKs
For programmatic access, use one of the official [Conductor SDKs](../clientsdks/index.md) which wrap these REST APIs with language-native interfaces for Java, Python, Go, JavaScript, C#, Ruby, and Rust.
docs/documentation/api/metadata.md
+317
View File
@@ -0,0 +1,317 @@
---
description: "Conductor Metadata API — register, update, validate, and delete workflow and task definitions. Manage your orchestration blueprints via REST."
---
# Metadata API
The Metadata API manages workflow and task definitions — the blueprints that Conductor uses to orchestrate executions. All endpoints use the base path `/api/metadata`.
## Workflow Definitions
| Endpoint | Method | Description |
|---|---|---|
| `/metadata/workflow` | `GET` | Get all workflow definitions |
| `/metadata/workflow` | `POST` | Create a new workflow definition |
| `/metadata/workflow` | `PUT` | Create or update workflow definitions (batch) |
| `/metadata/workflow/{name}` | `GET` | Get a workflow definition by name |
| `/metadata/workflow/{name}/{version}` | `DELETE` | Delete a workflow definition by name and version |
| `/metadata/workflow/validate` | `POST` | Validate a workflow definition without saving |
| `/metadata/workflow/names-and-versions` | `GET` | Get all workflow names and versions (no definition bodies) |
| `/metadata/workflow/latest-versions` | `GET` | Get only the latest version of each workflow definition |
### Get All Workflow Definitions
```
GET /api/metadata/workflow
```
Returns a list of all registered workflow definitions.
```shell
curl http://localhost:8080/api/metadata/workflow
```
**Response** `200 OK`
```json
[
{
"name": "order_processing",
"version": 1,
"tasks": [...],
"inputParameters": [],
"outputParameters": {},
"schemaVersion": 2
}
]
```
### Create a Workflow Definition
```
POST /api/metadata/workflow
```
Registers a new workflow definition. Request body is a [Workflow Definition](../configuration/workflowdef/index.md).
```shell
curl -X POST 'http://localhost:8080/api/metadata/workflow' \
-H 'Content-Type: application/json' \
-d '{
"name": "my_workflow",
"version": 1,
"tasks": [
{
"name": "my_task",
"taskReferenceName": "my_task_ref",
"type": "SIMPLE"
}
],
"schemaVersion": 2,
"ownerEmail": "dev@example.com"
}'
```
**Response** `200 OK` — no response body.
### Create or Update Workflow Definitions
```
PUT /api/metadata/workflow
```
Creates or updates workflow definitions in bulk. Request body is a list of [Workflow Definitions](../configuration/workflowdef/index.md). Returns a `BulkResponse` indicating success and failure for each definition.
```shell
curl -X PUT 'http://localhost:8080/api/metadata/workflow' \
-H 'Content-Type: application/json' \
-d '[
{"name": "workflow_a", "version": 1, "tasks": [...], "schemaVersion": 2},
{"name": "workflow_b", "version": 1, "tasks": [...], "schemaVersion": 2}
]'
```
**Response** `200 OK`
```json
{
"bulkSuccessfulResults": ["workflow_a", "workflow_b"],
"bulkErrorResults": {}
}
```
### Get Workflow Definition by Name
```
GET /api/metadata/workflow/{name}?version={version}
```
| Parameter | Description | Required |
|---|---|---|
| `name` | Workflow name | Yes (path) |
| `version` | Workflow version | No (defaults to latest) |
```shell
curl 'http://localhost:8080/api/metadata/workflow/my_workflow?version=1'
```
**Response** `200 OK` — returns the full workflow definition JSON.
### Delete a Workflow Definition
```
DELETE /api/metadata/workflow/{name}/{version}
```
Removes a workflow definition by name and version. Does **not** remove workflow executions associated with the definition.
| Parameter | Description | Required |
|---|---|---|
| `name` | Workflow name | Yes (path) |
| `version` | Workflow version | Yes (path) |
```shell
curl -X DELETE 'http://localhost:8080/api/metadata/workflow/my_workflow/1'
```
**Response** `200 OK` — no response body.
### Validate a Workflow Definition
```
POST /api/metadata/workflow/validate
```
Validates a workflow definition without registering it. Useful for CI/CD pipelines or pre-deployment checks.
```shell
curl -X POST 'http://localhost:8080/api/metadata/workflow/validate' \
-H 'Content-Type: application/json' \
-d '{
"name": "my_workflow",
"version": 1,
"tasks": [
{
"name": "my_task",
"taskReferenceName": "my_task_ref",
"type": "SIMPLE"
}
],
"schemaVersion": 2
}'
```
**Response** `200 OK` if valid. `400 Bad Request` with error details if invalid.
### Get Workflow Names and Versions
```
GET /api/metadata/workflow/names-and-versions
```
Returns a lightweight map of workflow names to their available versions (no definition bodies). Useful for building UIs or listing available workflows.
```shell
curl http://localhost:8080/api/metadata/workflow/names-and-versions
```
**Response** `200 OK`
```json
{
"order_processing": [
{"name": "order_processing", "version": 1},
{"name": "order_processing", "version": 2}
],
"user_onboarding": [
{"name": "user_onboarding", "version": 1}
]
}
```
### Get Latest Versions Only
```
GET /api/metadata/workflow/latest-versions
```
Returns only the latest version of each workflow definition.
```shell
curl http://localhost:8080/api/metadata/workflow/latest-versions
```
**Response** `200 OK` — returns a list of workflow definitions (one per workflow name, latest version only).
---
## Task Definitions
| Endpoint | Method | Description |
|---|---|---|
| `/metadata/taskdefs` | `GET` | Get all task definitions |
| `/metadata/taskdefs` | `POST` | Create new task definitions |
| `/metadata/taskdefs` | `PUT` | Update a task definition |
| `/metadata/taskdefs/{taskType}` | `GET` | Get a task definition by name |
| `/metadata/taskdefs/{taskType}` | `DELETE` | Delete a task definition |
### Get All Task Definitions
```
GET /api/metadata/taskdefs
```
```shell
curl http://localhost:8080/api/metadata/taskdefs
```
**Response** `200 OK`
```json
[
{
"name": "my_task",
"retryCount": 3,
"retryLogic": "FIXED",
"retryDelaySeconds": 10,
"timeoutSeconds": 300,
"timeoutPolicy": "TIME_OUT_WF",
"responseTimeoutSeconds": 180
}
]
```
### Create Task Definitions
```
POST /api/metadata/taskdefs
```
Registers new task definitions. Request body is a list of [Task Definitions](../configuration/taskdef.md).
```shell
curl -X POST 'http://localhost:8080/api/metadata/taskdefs' \
-H 'Content-Type: application/json' \
-d '[
{
"name": "my_task",
"retryCount": 3,
"retryLogic": "FIXED",
"retryDelaySeconds": 10,
"timeoutSeconds": 300,
"timeoutPolicy": "TIME_OUT_WF",
"responseTimeoutSeconds": 180,
"ownerEmail": "dev@example.com"
}
]'
```
**Response** `200 OK` — no response body.
### Update a Task Definition
```
PUT /api/metadata/taskdefs
```
Updates an existing task definition. Request body is a single [Task Definition](../configuration/taskdef.md).
```shell
curl -X PUT 'http://localhost:8080/api/metadata/taskdefs' \
-H 'Content-Type: application/json' \
-d '{
"name": "my_task",
"retryCount": 5,
"retryLogic": "EXPONENTIAL_BACKOFF",
"retryDelaySeconds": 5,
"timeoutSeconds": 600,
"timeoutPolicy": "TIME_OUT_WF",
"responseTimeoutSeconds": 300
}'
```
**Response** `200 OK` — no response body.
### Get Task Definition by Name
```
GET /api/metadata/taskdefs/{taskType}
```
```shell
curl http://localhost:8080/api/metadata/taskdefs/my_task
```
**Response** `200 OK` — returns the task definition JSON.
### Delete a Task Definition
```
DELETE /api/metadata/taskdefs/{taskType}
```
```shell
curl -X DELETE http://localhost:8080/api/metadata/taskdefs/my_task
```
**Response** `200 OK` — no response body.
docs/documentation/api/scheduler.md
+261
View File
@@ -0,0 +1,261 @@
---
description: "REST API reference for Conductor's workflow scheduler — create, list, search, pause, resume, delete schedules, preview cron execution times, and search execution history."
---
# Scheduler API
All scheduler endpoints are relative to `/api/scheduler`.
## Create or update a schedule
```
POST /api/scheduler/schedules
```
Creates a new schedule or updates an existing one (matched by `name`).
**Request body:**
```json
{
"name": "daily-report-schedule",
"cronExpression": "0 0 9 * * MON-FRI",
"zoneId": "America/New_York",
"startWorkflowRequest": {
"name": "daily_report_workflow",
"version": 1,
"input": {},
"correlationId": "daily-report-${scheduledTime}"
},
"runCatchupScheduleInstances": false,
"paused": false,
"scheduleStartTime": 0,
"scheduleEndTime": 0,
"description": "Triggers the daily report workflow on weekday mornings"
}
```
**Schedule fields:**
| Field | Type | Required | Default | Description |
|---|---|---|---|---|
| `name` | string | Yes | — | Unique schedule identifier |
| `cronExpression` | string | Yes | — | 6-field Spring cron expression (second precision) |
| `zoneId` | string | No | `UTC` | IANA timezone for cron evaluation |
| `startWorkflowRequest` | object | Yes | — | Workflow trigger configuration (see below) |
| `runCatchupScheduleInstances` | boolean | No | `false` | Fire missed slots on scheduler restart |
| `paused` | boolean | No | `false` | Create in paused state |
| `scheduleStartTime` | long | No | — | Earliest fire time (epoch ms) |
| `scheduleEndTime` | long | No | — | Latest fire time (epoch ms) |
| `description` | string | No | — | Free-text description |
**startWorkflowRequest fields:**
| Field | Type | Required | Description |
|---|---|---|---|
| `name` | string | Yes | Workflow name |
| `version` | integer | No | Workflow version (latest if omitted) |
| `input` | object | No | Static input merged with auto-injected `_scheduledTime` and `_executedTime` |
| `correlationId` | string | No | Supports `${scheduledTime}` template variable |
| `taskToDomain` | object | No | Task-to-domain mapping |
| `priority` | integer | No | Execution priority (0-99) |
**Response:** `200 OK` — returns the saved `WorkflowSchedule` object.
??? note "Example using cURL"
```shell
curl -X POST 'http://localhost:8080/api/scheduler/schedules' \
-H 'Content-Type: application/json' \
-d '{
"name": "daily-report-schedule",
"cronExpression": "0 0 9 * * MON-FRI",
"zoneId": "America/New_York",
"startWorkflowRequest": {
"name": "daily_report_workflow",
"version": 1,
"correlationId": "daily-report-${scheduledTime}"
}
}'
```
---
## List all schedules
```
GET /api/scheduler/schedules
```
**Query parameters:**
| Parameter | Type | Required | Description |
|---|---|---|---|
| `workflowName` | string | No | Filter by workflow name |
**Response:** `200 OK` — array of `WorkflowSchedule` objects.
??? note "Example using cURL"
```shell
# List all
curl 'http://localhost:8080/api/scheduler/schedules'
# Filter by workflow
curl 'http://localhost:8080/api/scheduler/schedules?workflowName=daily_report_workflow'
```
---
## Search schedules
```
GET /api/scheduler/schedules/search
```
**Query parameters:**
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
| `workflowName` | string | No | — | Filter by workflow name |
| `scheduleName` | string | No | — | Filter by schedule name |
| `paused` | boolean | No | — | Filter by paused state |
| `freeText` | string | No | `*` | Free-text search |
| `start` | integer | No | `0` | Pagination offset |
| `size` | integer | No | `100` | Page size |
| `sort` | string | No | — | Sort fields |
**Response:** `200 OK` — `SearchResult` with `totalHits` and `results` array.
??? note "Example using cURL"
```shell
curl 'http://localhost:8080/api/scheduler/schedules/search?workflowName=daily_report_workflow&size=10'
```
---
## Get a schedule by name
```
GET /api/scheduler/schedules/{name}
```
**Response:** `200 OK` — `WorkflowSchedule` object.
??? note "Example using cURL"
```shell
curl 'http://localhost:8080/api/scheduler/schedules/daily-report-schedule'
```
---
## Delete a schedule
```
DELETE /api/scheduler/schedules/{name}
```
**Response:** `204 No Content`
??? note "Example using cURL"
```shell
curl -X DELETE 'http://localhost:8080/api/scheduler/schedules/daily-report-schedule'
```
---
## Pause a schedule
```
PUT /api/scheduler/schedules/{name}/pause
```
**Query parameters:**
| Parameter | Type | Required | Description |
|---|---|---|---|
| `reason` | string | No | Reason for pausing |
**Response:** `200 OK`
??? note "Example using cURL"
```shell
curl -X PUT 'http://localhost:8080/api/scheduler/schedules/daily-report-schedule/pause?reason=maintenance+window'
```
---
## Resume a schedule
```
PUT /api/scheduler/schedules/{name}/resume
```
**Response:** `200 OK`
??? note "Example using cURL"
```shell
curl -X PUT 'http://localhost:8080/api/scheduler/schedules/daily-report-schedule/resume'
```
---
## Preview next execution times
```
GET /api/scheduler/nextFewSchedules
```
Preview when a cron expression will fire next, without creating a schedule.
**Query parameters:**
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
| `cronExpression` | string | Yes | — | Cron expression to evaluate |
| `scheduleStartTime` | long | No | — | Window start (epoch ms) |
| `scheduleEndTime` | long | No | — | Window end (epoch ms) |
| `limit` | integer | No | `5` | Number of times to return |
**Response:** `200 OK` — array of epoch-millisecond timestamps.
??? note "Example using cURL"
```shell
curl 'http://localhost:8080/api/scheduler/nextFewSchedules?cronExpression=0+0+9+*+*+MON-FRI&limit=5'
```
---
## Search execution history
```
GET /api/scheduler/search/executions
```
Search past scheduled workflow executions.
**Query parameters:**
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
| `query` | string | No | — | Structured query |
| `freeText` | string | No | `*` | Free-text search (matches schedule name, workflow name) |
| `start` | integer | No | `0` | Pagination offset |
| `size` | integer | No | `100` | Page size |
| `sort` | string | No | — | Sort fields |
**Response:** `200 OK` — `SearchResult` with execution records:
| Field | Description |
|---|---|
| `executionId` | Unique execution record ID |
| `scheduleName` | Parent schedule name |
| `scheduledTime` | Cron slot time (epoch ms) |
| `executionTime` | Actual dispatch time (epoch ms) |
| `workflowName` | Triggered workflow name |
| `workflowId` | Triggered workflow instance ID |
| `state` | `POLLED`, `EXECUTED`, or `FAILED` |
| `reason` | Failure reason (if `FAILED`) |
??? note "Example using cURL"
```shell
curl 'http://localhost:8080/api/scheduler/search/executions?freeText=daily-report-schedule&size=20'
```
docs/documentation/api/startworkflow.md
+178
View File
@@ -0,0 +1,178 @@
---
description: "Start Conductor workflow executions — asynchronous, synchronous, and dynamic workflow execution via REST API with curl examples."
---
# Start Workflow API
## Start a Workflow (Asynchronous)
```
POST /api/workflow
```
Starts a new workflow execution asynchronously. Returns the workflow ID immediately.
### Request Body
| Field | Description | Required |
|---|---|---|
| `name` | Workflow name (must be registered) | Yes |
| `version` | Workflow version | No (defaults to latest) |
| `input` | JSON object with input parameters for the workflow | No |
| `correlationId` | Unique ID to correlate multiple workflow executions | No |
| `taskToDomain` | Task-to-domain mapping. See [Task Domains](taskdomains.md). | No |
| `workflowDef` | Inline [Workflow Definition](../configuration/workflowdef/index.md) for dynamic workflows. See [Dynamic Workflows](#dynamic-workflows). | No |
| `externalInputPayloadStoragePath` | Path to external payload storage. See [External Payload Storage](../advanced/externalpayloadstorage.md). | No |
| `priority` | Priority level (0–99) for tasks within this workflow | No |
### Example
```shell
curl -X POST 'http://localhost:8080/api/workflow' \
-H 'Content-Type: application/json' \
-d '{
"name": "myWorkflow",
"version": 1,
"correlationId": "order-123",
"priority": 1,
"input": {
"customerId": "CUST-456",
"amount": 99.99
},
"taskToDomain": {
"*": "mydomain"
}
}'
```
**Response** `200 OK` — returns the workflow ID as plain text:
```
3a5b8c2d-1234-5678-9abc-def012345678
```
### Start with Path Parameters
```
POST /api/workflow/{name}
```
Alternative way to start a workflow — specify the name in the path and pass input as the request body.
| Parameter | Type | Description | Required |
|---|---|---|---|
| `name` | Path | Workflow name | Yes |
| `version` | Query | Workflow version | No |
| `correlationId` | Query | Correlation ID | No |
| `priority` | Query | Priority 0–99 (default: 0) | No |
```shell
curl -X POST 'http://localhost:8080/api/workflow/myWorkflow?version=1&correlationId=order-123' \
-H 'Content-Type: application/json' \
-d '{"customerId": "CUST-456", "amount": 99.99}'
```
**Response** `200 OK` — returns the workflow ID as plain text.
---
## Execute a Workflow (Synchronous)
```
POST /api/workflow/execute/{name}/{version}
```
Starts a workflow and **waits for completion** (or a specified condition) before returning the result. This eliminates the need to poll for workflow status.
| Parameter | Type | Description | Required |
|---|---|---|---|
| `name` | Path | Workflow name | Yes |
| `version` | Path | Workflow version (use `0` for latest) | Yes |
| `requestId` | Query | Idempotency key | No (auto-generated) |
| `waitUntilTaskRef` | Query | Comma-separated task reference names to wait for | No |
| `waitForSeconds` | Query | Maximum wait time in seconds | No (default: 10) |
| `consistency` | Query | `DURABLE` or `EVENTUAL` | No (default: `DURABLE`) |
| `returnStrategy` | Query | Controls which workflow state is returned | No (default: `TARGET_WORKFLOW`) |
Request body: a StartWorkflowRequest object (same format as the [async start](#start-a-workflow-asynchronous)).
### Example
```shell
curl -X POST 'http://localhost:8080/api/workflow/execute/my_workflow/1?waitForSeconds=30' \
-H 'Content-Type: application/json' \
-d '{
"name": "my_workflow",
"version": 1,
"input": {
"url": "https://api.example.com/data"
}
}'
```
**Response** `200 OK` — returns the workflow execution result:
```json
{
"workflowId": "3a5b8c2d-1234-5678-9abc-def012345678",
"requestId": "req-uuid",
"status": "COMPLETED",
"output": {
"response": {...}
},
"tasks": [...]
}
```
### Wait Behavior
- If `waitUntilTaskRef` is specified, the API returns when any listed task reaches a terminal state (or a WAIT task is encountered)
- If the workflow completes before the timeout, the result is returned immediately
- If the timeout is reached, the current workflow state is returned — the workflow continues running in the background
- Sub-workflow WAIT tasks are detected recursively
---
## Dynamic Workflows
Start a one-time workflow without pre-registering its definition. Provide the full workflow definition inline via the `workflowDef` field.
```shell
curl -X POST 'http://localhost:8080/api/workflow' \
-H 'Content-Type: application/json' \
-d '{
"name": "my_adhoc_workflow",
"workflowDef": {
"ownerApp": "my_app",
"ownerEmail": "owner@example.com",
"name": "my_adhoc_workflow",
"version": 1,
"tasks": [
{
"name": "fetch_data",
"type": "HTTP",
"taskReferenceName": "fetch_data",
"inputParameters": {
"uri": "${workflow.input.uri}",
"method": "GET"
},
"taskDefinition": {
"name": "fetch_data",
"retryCount": 0,
"timeoutSeconds": 3600,
"timeoutPolicy": "TIME_OUT_WF",
"responseTimeoutSeconds": 3000
}
}
]
},
"input": {
"uri": "https://api.example.com/data"
}
}'
```
**Response** `200 OK` — returns the workflow ID as plain text.
!!! note
If a `taskDefinition` is already registered via the Metadata API, it does not need to be included inline in the dynamic workflow definition.
docs/documentation/api/task.md
+470
View File
@@ -0,0 +1,470 @@
---
description: "Conductor Task API — poll, update, search, and manage tasks. Includes batch polling, task logs, queue management, and poll data."
---
# Task API
The Task API manages task execution — polling, updating, logging, and queue management. All endpoints use the base path `/api/tasks`.
## Get Task
```
GET /api/tasks/{taskId}
```
Returns the task details for a given task ID.
```shell
curl 'http://localhost:8080/api/tasks/a1b2c3d4-5678-90ab-cdef-111111111111'
```
**Response** `200 OK`
```json
{
"taskType": "my_task",
"status": "COMPLETED",
"referenceTaskName": "my_task_ref",
"retryCount": 0,
"seq": 1,
"startTime": 1700000001000,
"endTime": 1700000003000,
"updateTime": 1700000003000,
"pollCount": 1,
"taskId": "a1b2c3d4-5678-90ab-cdef-111111111111",
"workflowInstanceId": "3a5b8c2d-1234-5678-9abc-def012345678",
"inputData": {"key": "value"},
"outputData": {"result": "success"},
"workerId": "worker-host-1"
}
```
---
## Poll and Update Tasks
These endpoints are used by workers to poll for tasks and update their results. They are typically called by the [SDK](../clientsdks/index.md), not manually.
### Poll for a Task
```
GET /api/tasks/poll/{taskType}?workerid=&domain=
```
Polls for a single task of the given type. Returns `204 No Content` if no task is available.
| Parameter | Description | Required |
|---|---|---|
| `taskType` | Task type to poll for | Yes |
| `workerid` | Identifier for the worker polling | No |
| `domain` | Task domain. See [Task Domains](taskdomains.md). | No |
```shell
curl 'http://localhost:8080/api/tasks/poll/my_task?workerid=worker-1'
```
**Response** `200 OK` — returns a task object (same format as Get Task above), or `204 No Content` if no tasks are queued.
### Batch Poll
```
GET /api/tasks/poll/batch/{taskType}?count=1&timeout=100&workerid=&domain=
```
Polls for multiple tasks in a single request. This is a **long poll** — the connection waits until `timeout` or at least 1 task is available.
| Parameter | Description | Default |
|---|---|---|
| `taskType` | Task type to poll for | — |
| `count` | Maximum number of tasks to return | `1` |
| `timeout` | Long poll timeout in milliseconds | `100` |
| `workerid` | Worker identifier | — |
| `domain` | Task domain | — |
```shell
# Poll for up to 5 tasks, wait up to 1 second
curl 'http://localhost:8080/api/tasks/poll/batch/my_task?count=5&timeout=1000&workerid=worker-1'
```
**Response** `200 OK` — returns a list of task objects, or an empty list if no tasks are available.
```json
[
{
"taskType": "my_task",
"status": "IN_PROGRESS",
"taskId": "task-uuid-1",
"workflowInstanceId": "workflow-uuid-1",
"inputData": {"key": "value1"}
},
{
"taskType": "my_task",
"status": "IN_PROGRESS",
"taskId": "task-uuid-2",
"workflowInstanceId": "workflow-uuid-2",
"inputData": {"key": "value2"}
}
]
```
### Update Task
```
POST /api/tasks
```
Updates the result of a task execution. Returns the task ID.
```shell
curl -X POST 'http://localhost:8080/api/tasks' \
-H 'Content-Type: application/json' \
-d '{
"workflowInstanceId": "3a5b8c2d-1234-5678-9abc-def012345678",
"taskId": "a1b2c3d4-5678-90ab-cdef-111111111111",
"status": "COMPLETED",
"outputData": {
"result": "processed successfully",
"recordCount": 42
}
}'
```
**Request body fields:**
| Field | Description | Required |
|---|---|---|
| `workflowInstanceId` | Workflow execution ID | Yes |
| `taskId` | Task ID | Yes |
| `status` | `IN_PROGRESS`, `COMPLETED`, `FAILED`, or `FAILED_WITH_TERMINAL_ERROR` | Yes |
| `outputData` | JSON map of output data | No |
| `reasonForIncompletion` | Reason for failure (when status is `FAILED`) | No |
| `callbackAfterSeconds` | Callback delay — task will be put back in queue after this time | No |
| `logs` | List of log entries to append | No |
**Response** `200 OK` — returns the task ID as plain text.
### Update Task V2
```
POST /api/tasks/update-v2
```
Updates a task and returns the **next available task** to be processed — combining an update and poll in one call. Returns `204 No Content` if no next task is available.
```shell
curl -X POST 'http://localhost:8080/api/tasks/update-v2' \
-H 'Content-Type: application/json' \
-d '{
"workflowInstanceId": "3a5b8c2d-1234-5678-9abc-def012345678",
"taskId": "a1b2c3d4-5678-90ab-cdef-111111111111",
"status": "COMPLETED",
"outputData": {"result": "done"}
}'
```
**Response** `200 OK` — returns the next task object, or `204 No Content` if no tasks are queued.
### Update Task by Reference Name
```
POST /api/tasks/{workflowId}/{taskRefName}/{status}?workerid=
```
Updates a task using the workflow ID and task reference name instead of the task ID. This is useful for completing WAIT or HUMAN tasks from external systems.
| Parameter | Description | Required |
|---|---|---|
| `workflowId` | Workflow execution ID | Yes |
| `taskRefName` | Task reference name in the workflow | Yes |
| `status` | `IN_PROGRESS`, `COMPLETED`, `FAILED`, or `FAILED_WITH_TERMINAL_ERROR` | Yes |
| `workerid` | Worker identifier | No |
Request body: JSON map of output data.
```shell
# Complete a WAIT task with output data
curl -X POST 'http://localhost:8080/api/tasks/3a5b8c2d.../wait_for_approval/COMPLETED' \
-H 'Content-Type: application/json' \
-d '{"approved": true, "approver": "jane@example.com"}'
```
**Response** `200 OK` — no response body.
### Update Task by Reference Name (Synchronous)
```
POST /api/tasks/{workflowId}/{taskRefName}/{status}/sync?workerid=
```
Same as above, but returns the **updated workflow** after the task update is processed. Useful for synchronous execution patterns where you need the workflow state immediately after updating a task.
```shell
curl -X POST 'http://localhost:8080/api/tasks/3a5b8c2d.../wait_for_signal/COMPLETED/sync' \
-H 'Content-Type: application/json' \
-d '{"signal": "proceed"}'
```
**Response** `200 OK` — returns the full workflow execution object.
---
## Task Logs
### Add a Task Log
```
POST /api/tasks/{taskId}/log
```
Adds an execution log entry to a task. Request body: log message as a plain string.
```shell
curl -X POST 'http://localhost:8080/api/tasks/a1b2c3d4.../log' \
-H 'Content-Type: text/plain' \
-d 'Processing started for batch #42'
```
**Response** `200 OK` — no response body.
### Get Task Logs
```
GET /api/tasks/{taskId}/log
```
Returns execution logs for a task. Returns `204 No Content` if no logs exist.
```shell
curl 'http://localhost:8080/api/tasks/a1b2c3d4.../log'
```
**Response** `200 OK`
```json
[
{
"log": "Processing started for batch #42",
"taskId": "a1b2c3d4-5678-90ab-cdef-111111111111",
"createdTime": 1700000001000
},
{
"log": "Batch #42 completed: 100 records processed",
"taskId": "a1b2c3d4-5678-90ab-cdef-111111111111",
"createdTime": 1700000003000
}
]
```
---
## Queue Management
| Endpoint | Method | Description |
|---|---|---|
| `/queue/all` | `GET` | Get pending task counts for all queues |
| `/queue/all/verbose` | `GET` | Get detailed queue info including per-shard counts |
| `/queue/size` | `GET` | Get queue size for a specific task type |
| `/queue/sizes` | `GET` | *(Deprecated)* Get queue sizes for task types. Use `/queue/size` instead. |
| `/queue/requeue/{taskType}` | `POST` | Requeue pending tasks of a given type |
### Get Queue Size
```
GET /api/tasks/queue/size?taskType=&domain=&isolationGroupId=&executionNamespace=
```
Returns the queue depth for a specific task type, optionally filtered by domain and isolation group.
```shell
curl 'http://localhost:8080/api/tasks/queue/size?taskType=my_task'
```
**Response** `200 OK`
```json
5
```
### Get All Queue Sizes
```
GET /api/tasks/queue/all
```
Returns a map of task type to pending count for all queues.
```shell
curl 'http://localhost:8080/api/tasks/queue/all'
```
**Response** `200 OK`
```json
{
"my_task": 5,
"http_task": 0,
"email_task": 12
}
```
### Get All Queue Details (Verbose)
```
GET /api/tasks/queue/all/verbose
```
Returns detailed queue information including per-shard counts.
```shell
curl 'http://localhost:8080/api/tasks/queue/all/verbose'
```
**Response** `200 OK`
```json
{
"my_task": {
"size": 5,
"shards": {"0": 3, "1": 2}
}
}
```
### Requeue Pending Tasks
```
POST /api/tasks/queue/requeue/{taskType}
```
Requeues all pending tasks of the specified type. Useful for recovery after worker issues.
```shell
curl -X POST 'http://localhost:8080/api/tasks/queue/requeue/my_task'
```
**Response** `200 OK` — returns the number of tasks requeued.
---
## Poll Data
### Get Poll Data for a Task Type
```
GET /api/tasks/queue/polldata?taskType=
```
Returns the last poll data for a given task type — useful for monitoring worker health and activity.
```shell
curl 'http://localhost:8080/api/tasks/queue/polldata?taskType=my_task'
```
**Response** `200 OK`
```json
[
{
"queueName": "my_task",
"domain": null,
"workerId": "worker-host-1",
"lastPollTime": 1700000005000
}
]
```
### Get Poll Data for All Task Types
```
GET /api/tasks/queue/polldata/all
```
Returns the last poll data for all task types.
```shell
curl 'http://localhost:8080/api/tasks/queue/polldata/all'
```
**Response** `200 OK` — returns a list of poll data objects (same format as above) for all task types.
---
## Search Tasks
All search endpoints support the same query parameters:
| Parameter | Description | Default |
|---|---|---|
| `start` | Page offset | `0` |
| `size` | Number of results | `100` |
| `sort` | Sort order: `<field>:ASC` or `<field>:DESC` | — |
| `freeText` | Full-text search query | `*` |
| `query` | SQL-like where clause | — |
### Search (Summary)
```
GET /api/tasks/search?start=0&size=100&sort=&freeText=&query=
```
Returns `SearchResult<TaskSummary>` — lightweight results.
```shell
# Find failed tasks for a specific workflow type
curl 'http://localhost:8080/api/tasks/search?query=workflowType%3D%27order_processing%27+AND+status%3D%27FAILED%27&size=10'
# Free-text search
curl 'http://localhost:8080/api/tasks/search?freeText=timeout'
```
**Response** `200 OK`
```json
{
"totalHits": 3,
"results": [
{
"taskId": "task-uuid",
"taskType": "my_task",
"referenceTaskName": "my_task_ref",
"workflowId": "workflow-uuid",
"workflowType": "order_processing",
"status": "FAILED",
"startTime": "2024-01-15T10:30:00Z",
"updateTime": "2024-01-15T10:30:05Z",
"executionTime": 5000
}
]
}
```
### Search V2 (Full)
```
GET /api/tasks/search-v2?start=0&size=100&sort=&freeText=&query=
```
Returns `SearchResult<Task>` — full task objects including input/output data.
---
## External Storage
```
GET /api/tasks/externalstoragelocation?path=&operation=&payloadType=
```
Get the URI for external task payload storage. See [External Payload Storage](../advanced/externalpayloadstorage.md).
```shell
curl 'http://localhost:8080/api/tasks/externalstoragelocation?path=task/output&operation=WRITE&payloadType=TASK_OUTPUT'
```
**Response** `200 OK`
```json
{
"uri": "s3://conductor-payloads/task/output/...",
"path": "task/output/..."
}
```
docs/documentation/api/taskdomains.md
+95
View File
@@ -0,0 +1,95 @@
---
description: "Task Domains — route Conductor tasks to specific worker groups for isolated development, testing, and canary deployments."
---
# Task Domains
Task domains helps support task development. The idea is same "task definition" can be implemented in different "domains". A domain is some arbitrary name that the developer controls. So when the workflow is started, the caller can specify, out of all the tasks in the workflow, which tasks need to run in a specific domain, this domain is then used to poll for task on the client side to execute it.
As an example if a workflow (WF1) has 3 tasks T1, T2, T3. The workflow is deployed and working fine, which means there are T2 workers polling and executing. If you modify T2 and run it locally there is no guarantee that your modified T2 worker will get the task that you are looking for as it coming from the general T2 queue. "Task Domain" feature solves this problem by splitting the T2 queue by domains, so when the app polls for task T2 in a specific domain, it get the correct task.
When starting a workflow multiple domains can be specified as a fall backs, for example "domain1,domain2". Conductor keeps track of last polling time for each task, so in this case it checks if the there are any active workers (workers polled at least once in a 10 second window) for "domain1" then the task is put in "domain1", if not then the same check is done for the next domain in sequence "domain2" and so on.
If no workers are active for the domains provided:
- If `NO_DOMAIN` is provided as last token in list of domains, then no domain is set.
- Else, task will be added to last inactive domain in list of domains, hoping that workers would soon be available for that domain.
Also, a `*` token can be used to apply domains for all tasks. This can be overridden by providing task specific mappings along with `*`.
For example, the below configuration:
```json
"taskToDomain": {
"*": "mydomain",
"some_task_x":"NO_DOMAIN",
"some_task_y": "someDomain, NO_DOMAIN",
"some_task_z": "someInactiveDomain1, someInactiveDomain2"
}
```
- puts `some_task_x` in default queue (no domain).
- puts `some_task_y` in `someDomain` domain, if available or in default otherwise.
- puts `some_task_z` in `someInactiveDomain2`, even though workers are not available yet.
- and puts all other tasks in `mydomain` (even if workers are not available).
<b>Note</b> that this "fall back" type domain strings can only be used when starting the workflow, when polling from the client only one domain is used. Also, `NO_DOMAIN` token should be used last.
## How to use Task Domains
### Change the poll call
The poll call must now specify the domain.
#### Java Client
If you are using the java client then a simple property change will force TaskRunnerConfigurer to pass the domain to the poller.
```
conductor.worker.T2.domain=mydomain //Task T2 needs to poll for domain "mydomain"
```
#### REST call
`GET {{ api_prefix }}/tasks/poll/batch/T2?workerid=myworker&domain=mydomain`
`GET {{ api_prefix }}/tasks/poll/T2?workerid=myworker&domain=mydomain`
### Change the start workflow call
When starting the workflow, make sure the task to domain mapping is passes
#### Java Client
```java
{Map<String, Object> input = new HashMap<>();
input.put("wf_input1", "one");
Map<String, String> taskToDomain = new HashMap<>();
taskToDomain.put("T2", "mydomain");
// Other options ...
// taskToDomain.put("*", "mydomain, NO_DOMAIN")
// taskToDomain.put("T2", "mydomain, fallbackDomain1, fallbackDomain2")
StartWorkflowRequest swr = new StartWorkflowRequest();
swr.withName("myWorkflow")
.withCorrelationId("corr1")
.withVersion(1)
.withInput(input)
.withTaskToDomain(taskToDomain);
wfclient.startWorkflow(swr);
```
#### REST call
`POST {{ api_prefix }}/workflow`
```json
{
"name": "myWorkflow",
"version": 1,
"correlatonId": "corr1"
"input": {
"wf_input1": "one"
},
"taskToDomain": {
"*": "mydomain",
"some_task_x":"NO_DOMAIN",
"some_task_y": "someDomain, NO_DOMAIN"
}
}
```
docs/documentation/api/workflow.md
+440
View File
@@ -0,0 +1,440 @@
---
description: "Conductor Workflow API — manage workflow executions including pause, resume, retry, restart, rerun, terminate, search, and test workflows via REST."
---
# Workflow API
The Workflow API manages workflow executions. All endpoints use the base path `/api/workflow`.
For starting workflows, see [Start Workflow API](startworkflow.md).
## Retrieve Workflows
| Endpoint | Method | Description |
|---|---|---|
| `/{workflowId}` | `GET` | Get workflow execution by ID |
| `/{workflowId}/tasks` | `GET` | Get tasks for a workflow execution (paginated) |
| `/running/{name}` | `GET` | Get running workflow IDs by type |
| `/{name}/correlated/{correlationId}` | `GET` | Get workflows by correlation ID |
| `/{name}/correlated` | `POST` | Get workflows for multiple correlation IDs |
### Get Workflow by ID
```
GET /api/workflow/{workflowId}?includeTasks=true
```
| Parameter | Description | Default |
|---|---|---|
| `workflowId` | Workflow execution ID | — |
| `includeTasks` | Include task details in response | `true` |
```shell
curl 'http://localhost:8080/api/workflow/3a5b8c2d-1234-5678-9abc-def012345678'
```
**Response** `200 OK`
```json
{
"workflowId": "3a5b8c2d-1234-5678-9abc-def012345678",
"workflowName": "order_processing",
"workflowVersion": 1,
"status": "COMPLETED",
"startTime": 1700000000000,
"endTime": 1700000005000,
"input": {"orderId": "ORD-123"},
"output": {"paymentId": "PAY-456"},
"tasks": [
{
"taskId": "task-uuid",
"taskType": "HTTP",
"referenceTaskName": "validate",
"status": "COMPLETED",
"outputData": {"response": {"statusCode": 200}}
}
],
"correlationId": "order-123"
}
```
### Get Tasks for a Workflow
```
GET /api/workflow/{workflowId}/tasks?start=0&count=15&status=
```
Returns a paginated list of tasks for a workflow execution.
| Parameter | Description | Default |
|---|---|---|
| `start` | Page offset | `0` |
| `count` | Number of results | `15` |
| `status` | Filter by task status (can specify multiple) | All statuses |
```shell
# Get first 10 tasks
curl 'http://localhost:8080/api/workflow/3a5b8c2d.../tasks?count=10'
# Get only failed tasks
curl 'http://localhost:8080/api/workflow/3a5b8c2d.../tasks?status=FAILED'
```
**Response** `200 OK`
```json
{
"totalHits": 5,
"results": [
{
"taskId": "task-uuid",
"taskType": "HTTP",
"referenceTaskName": "validate",
"status": "COMPLETED"
}
]
}
```
### Get Running Workflows
```
GET /api/workflow/running/{name}?version=1&startTime=&endTime=
```
Returns a list of workflow IDs for running workflows of the given type.
| Parameter | Description | Default |
|---|---|---|
| `name` | Workflow name | — |
| `version` | Workflow version | `1` |
| `startTime` | Filter by start time (epoch ms) | — |
| `endTime` | Filter by end time (epoch ms) | — |
```shell
curl 'http://localhost:8080/api/workflow/running/order_processing?version=1'
```
**Response** `200 OK`
```json
["3a5b8c2d-1234-...", "7f8e9d0c-5678-..."]
```
### Get Workflows by Correlation ID
```
GET /api/workflow/{name}/correlated/{correlationId}?includeClosed=false&includeTasks=false
```
| Parameter | Description | Default |
|---|---|---|
| `includeClosed` | Include completed/terminated workflows | `false` |
| `includeTasks` | Include task details | `false` |
```shell
curl 'http://localhost:8080/api/workflow/order_processing/correlated/order-123?includeClosed=true'
```
### Get Workflows for Multiple Correlation IDs
```
POST /api/workflow/{name}/correlated?includeClosed=false&includeTasks=false
```
```shell
curl -X POST 'http://localhost:8080/api/workflow/order_processing/correlated?includeClosed=true' \
-H 'Content-Type: application/json' \
-d '["order-123", "order-456", "order-789"]'
```
**Response** `200 OK` — a map of correlation ID to list of workflows.
---
## Manage Workflows
| Endpoint | Method | Description |
|---|---|---|
| `/{workflowId}/pause` | `PUT` | Pause a workflow |
| `/{workflowId}/resume` | `PUT` | Resume a paused workflow |
| `/{workflowId}/restart` | `POST` | Restart a completed workflow from the beginning |
| `/{workflowId}/retry` | `POST` | Retry the last failed task |
| `/{workflowId}/rerun` | `POST` | Rerun from a specific task |
| `/{workflowId}/skiptask/{taskReferenceName}` | `PUT` | Skip a task in a running workflow |
| `/{workflowId}/resetcallbacks` | `POST` | Reset callback times for SIMPLE tasks |
| `/decide/{workflowId}` | `PUT` | Trigger the decider for a workflow |
| `/{workflowId}` | `DELETE` | Terminate a running workflow |
| `/{workflowId}/remove` | `DELETE` | Remove a workflow from the system |
| `/{workflowId}/terminate-remove` | `DELETE` | Terminate and remove in one call |
### Pause
```
PUT /api/workflow/{workflowId}/pause
```
Pauses the workflow. No further tasks will be scheduled until resumed. Currently running tasks are **not** affected.
```shell
curl -X PUT 'http://localhost:8080/api/workflow/3a5b8c2d.../pause'
```
### Resume
```
PUT /api/workflow/{workflowId}/resume
```
```shell
curl -X PUT 'http://localhost:8080/api/workflow/3a5b8c2d.../resume'
```
### Restart
```
POST /api/workflow/{workflowId}/restart?useLatestDefinitions=false
```
Restarts a completed workflow from the beginning. Current execution history is wiped out.
| Parameter | Description | Default |
|---|---|---|
| `useLatestDefinitions` | Use latest workflow and task definitions | `false` |
```shell
curl -X POST 'http://localhost:8080/api/workflow/3a5b8c2d.../restart'
```
### Retry
```
POST /api/workflow/{workflowId}/retry?resumeSubworkflowTasks=false
```
Retries the last failed task in the workflow.
| Parameter | Description | Default |
|---|---|---|
| `resumeSubworkflowTasks` | Also resume failed sub-workflow tasks | `false` |
```shell
curl -X POST 'http://localhost:8080/api/workflow/3a5b8c2d.../retry'
```
### Rerun
```
POST /api/workflow/{workflowId}/rerun
```
Re-runs a completed workflow from a specific task.
```shell
curl -X POST 'http://localhost:8080/api/workflow/3a5b8c2d.../rerun' \
-H 'Content-Type: application/json' \
-d '{
"reRunFromWorkflowId": "3a5b8c2d...",
"workflowInput": {"orderId": "ORD-999"},
"reRunFromTaskId": "task-uuid",
"taskInput": {"override": true}
}'
```
### Skip Task
```
PUT /api/workflow/{workflowId}/skiptask/{taskReferenceName}
```
Skips a task in a running workflow and continues forward. Optionally provide updated input/output:
```shell
curl -X PUT 'http://localhost:8080/api/workflow/3a5b8c2d.../skiptask/validate_ref' \
-H 'Content-Type: application/json' \
-d '{
"taskInput": {},
"taskOutput": {"skipped": true, "reason": "manual override"}
}'
```
### Reset Callbacks
```
POST /api/workflow/{workflowId}/resetcallbacks
```
Resets callback times of all non-terminal SIMPLE tasks to 0, causing them to be re-evaluated immediately.
### Decide
```
PUT /api/workflow/decide/{workflowId}
```
Manually triggers the decider for a workflow. The decider evaluates workflow state and schedules the next tasks. Normally automatic — use this for debugging.
### Terminate
```
DELETE /api/workflow/{workflowId}?reason=
```
| Parameter | Description | Required |
|---|---|---|
| `reason` | Reason for termination | No |
```shell
curl -X DELETE 'http://localhost:8080/api/workflow/3a5b8c2d...?reason=cancelled+by+user'
```
### Remove
```
DELETE /api/workflow/{workflowId}/remove?archiveWorkflow=true
```
| Parameter | Description | Default |
|---|---|---|
| `archiveWorkflow` | Archive before removing | `true` |
!!! warning
This permanently removes the workflow execution data. Use with caution.
### Terminate and Remove
```
DELETE /api/workflow/{workflowId}/terminate-remove?reason=&archiveWorkflow=true
```
Terminates a running workflow and removes it from the system in one call.
---
## Search Workflows
All search endpoints support the same query parameters:
| Parameter | Description | Default |
|---|---|---|
| `start` | Page offset | `0` |
| `size` | Number of results | `100` |
| `sort` | Sort order: `<field>:ASC` or `<field>:DESC` | — |
| `freeText` | Full-text search query | `*` |
| `query` | SQL-like where clause | — |
### Search (Summary)
```
GET /api/workflow/search?start=0&size=100&sort=&freeText=&query=
```
Returns `SearchResult<WorkflowSummary>` — lightweight results without full workflow details.
```shell
# Find completed workflows of a specific type
curl 'http://localhost:8080/api/workflow/search?query=workflowType%3D%27order_processing%27+AND+status%3D%27COMPLETED%27&size=10'
# Free-text search
curl 'http://localhost:8080/api/workflow/search?freeText=order-123'
```
**Response** `200 OK`
```json
{
"totalHits": 42,
"results": [
{
"workflowType": "order_processing",
"version": 1,
"workflowId": "3a5b8c2d...",
"correlationId": "order-123",
"startTime": "2024-01-15T10:30:00Z",
"updateTime": "2024-01-15T10:30:05Z",
"endTime": "2024-01-15T10:30:05Z",
"status": "COMPLETED",
"executionTime": 5000
}
]
}
```
### Search V2 (Full)
```
GET /api/workflow/search-v2
```
Same parameters as search, but returns `SearchResult<Workflow>` — full workflow objects including task details.
### Search by Tasks
```
GET /api/workflow/search-by-tasks
```
Search for workflows based on task-level parameters. Returns `SearchResult<WorkflowSummary>`.
### Search by Tasks V2
```
GET /api/workflow/search-by-tasks-v2
```
Returns `SearchResult<Workflow>` with full workflow objects.
### Query Syntax
The `query` parameter supports SQL-like expressions:
| Example | Description |
|---|---|
| `workflowType = 'order_processing'` | Filter by workflow type |
| `status = 'FAILED'` | Filter by status |
| `startTime > 1700000000000` | Filter by start time (epoch ms) |
| `workflowType = 'order_processing' AND status = 'COMPLETED'` | Combine conditions |
The `freeText` parameter supports Elasticsearch query syntax:
| Example | Description |
|---|---|
| `workflowType:"order_processing"` | Match workflow type |
| `order-123` | Match any field |
---
## Test Workflow
```
POST /api/workflow/test
```
Test a workflow execution using mock data without actually running it. Useful for validating workflow definitions and task wiring before deployment.
```shell
curl -X POST 'http://localhost:8080/api/workflow/test' \
-H 'Content-Type: application/json' \
-d '{
"name": "my_workflow",
"version": 1,
"workflowDef": {...},
"taskRefToMockOutput": {
"my_task_ref": {
"key": "mocked_value"
}
}
}'
```
**Response** `200 OK` — returns the simulated workflow execution with mocked task outputs.
---
## External Storage
```
GET /api/workflow/externalstoragelocation?path=&operation=&payloadType=
```
Get the URI for external payload storage. See [External Payload Storage](../advanced/externalpayloadstorage.md).