conductor-oss/conductor · error · IllegalStateException

Workflow message queue for workflowId={} has reached the max

Error message

Workflow message queue for workflowId={} has reached the maximum size of {}

What it means

Thrown by InMemoryWorkflowMessageQueueDAO.push when the per-workflowId queue has already reached maxQueueSize. This DAO backs the per-workflow message queue with an in-memory HashMap of LinkedLists (not durable, not clustered). The bound prevents unbounded memory growth from a single workflow.

Source

Thrown at core/src/main/java/com/netflix/conductor/core/dao/InMemoryWorkflowMessageQueueDAO.java:48

 *
 * <p>Used as the default DAO when no Redis-backed implementation is available (e.g. when {@code
 * conductor.db.type} is not a Redis variant). Not durable across server restarts.
 */
public class InMemoryWorkflowMessageQueueDAO implements WorkflowMessageQueueDAO {

    private final Map<String, Queue<WorkflowMessage>> queues = new HashMap<>();

    private final int maxQueueSize;

    public InMemoryWorkflowMessageQueueDAO(WorkflowMessageQueueProperties properties) {
        this.maxQueueSize = properties.getMaxQueueSize();
    }

    @Override
    public synchronized void push(String workflowId, WorkflowMessage message) {
        Queue<WorkflowMessage> queue = queues.computeIfAbsent(workflowId, k -> new LinkedList<>());
        if (queue.size() >= maxQueueSize) {
            throw new IllegalStateException(
                    "Workflow message queue for workflowId="
                            + workflowId
                            + " has reached the maximum size of "
                            + maxQueueSize);
        }
        queue.add(message);
    }

    @Override
    public synchronized List<WorkflowMessage> pop(String workflowId, int maxCount) {
        Queue<WorkflowMessage> queue = queues.get(workflowId);
        if (queue == null || queue.isEmpty()) {
            return Collections.emptyList();
        }
        List<WorkflowMessage> result = new ArrayList<>(maxCount);
        for (int i = 0; i < maxCount && !queue.isEmpty(); i++) {
            result.add(queue.poll());
        }

View on GitHub (pinned to cf7c3e4a8a)

Solutions

Drain the queue by consuming/popping messages for that workflowId before pushing more.
Raise conductor workflow-message-queue max-queue-size if the burst is legitimate.
Switch to the Redis-backed WorkflowMessageQueueDAO for production — the in-memory impl is single-node and non-durable.
Fix a stuck consumer so the queue drains naturally.
Delete the workflowId queue (delete(workflowId)) if it is orphaned.

Defensive patterns

Strategy: validation

Validate before calling

// Check queue size before pushing
long size = messageQueueDAO.size(workflowId);
if (size >= maxQueueSize) {
    // drain or back off instead of letting push throw
    handleBackpressure(workflowId);
}

Try / catch

try {
    messageQueueDAO.push(workflowId, message);
} catch (IllegalStateException e) {
    // queue full -> apply back-pressure: drain, raise size, or switch to durable DAO
}

Prevention

Do not use the in-memory DAO for production-scale queuing; switch to the Redis-backed implementation.
Size maxQueueSize to your worst-case burst and monitor queue depth.
Ensure consumers drain queues so producers never hit the cap.

When it happens

Trigger: Calling push(workflowId, message) when queues.get(workflowId).size() >= properties.getMaxQueueSize(). Fires before queue.add, so the message is rejected and the queue stays full.

Common situations: A workflow that consumes messages slower than they arrive (back-pressure not honored). A consumer stuck or dead, so messages accumulate. Using the in-memory DAO (default when no Redis variant) in a scenario that needs durable/scalable queuing — the bound trips quickly under load. A producer loop pushing without draining.

Related errors

AI-assisted analysis of conductor-oss/conductor@cf7c3e4a8a (2026-08-14). Data as JSON: /api/errors/41ba0fe507f77b24. Report an issue: GitHub.