apache/flink · error · UnsupportedOperationException

The transformation does not support async state, or you are

Error message

The transformation does not support async state, or you are enabling the async state without a keyed context (not behind a keyBy()).

What it means

Thrown by the base Transformation.enableAsyncState() when a subclass does not override it. Async (non-blocking) state processing is only implemented for transformations that operate in a keyed context (after keyBy) and whose runtime operators were written to use the async state API. Calling enableAsyncState() on any other transformation hits this default, which exists to fail loudly rather than silently ignore the request.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java:650

     * has the iteration head as a predecessor. This method is just a wrapper on top of {@code
     * getTransitivePredecessorsInternal} method with public access. It uses caching internally.
     *
     * @return The list of transitive predecessors.
     */
    public final List<Transformation<?>> getTransitivePredecessors() {
        return predecessorsCache.computeIfAbsent(this, key -> getTransitivePredecessorsInternal());
    }

    /**
     * Returns the {@link Transformation transformations} that are the immediate predecessors of the
     * current transformation in the transformation graph.
     */
    public abstract List<Transformation<?>> getInputs();

    /** Enabling the async state for this transformation. */
    public void enableAsyncState() {
        // Subclass should override this method if they support async state processing.
        throw new UnsupportedOperationException(
                "The transformation does not support async state, "
                        + "or you are enabling the async state without a keyed context "
                        + "(not behind a keyBy()).");
    }

    @Override
    public String toString() {
        return getClass().getSimpleName()
                + "{"
                + "id="
                + id
                + ", name='"
                + name
                + '\''
                + ", outputType="
                + outputType
                + ", parallelism="
                + parallelism

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Ensure a keyBy() precedes the operator so the transformation is keyed before enabling async state.
2. Only enable async state for the supported operators (windowed aggregates, window joins); remove enableAsyncState() from unsupported transformations.
3. Do not set the global async-state flag if your pipeline uses operators outside the supported set; enable it per-operator instead.
4. Check the operator's builder: only WindowAggOperatorBuilder/WindowJoinOperatorBuilder and other async-aware builders wire support through to their Transformation.

Example fix

// before
env.getConfig().enableAsyncState();
data.map(f).enableAsyncState(); // not keyed, not supported

// after
data.keyBy(k -> k.getId()).window(...).aggregate(...); // async state enabled only on supported keyed window operators

// Before enabling, confirm the transformation is keyed and supports async state.
if (!(transformation instanceof org.apache.flink.streaming.api.transformations.OneInputTransformation) || transformation.getInputs().isEmpty()) {
    throw new IllegalStateException("Async state requires a keyed, supported transformation.");
}
// Prefer enabling async state only via the dedicated builder API on supported operators.

// Only call on builders known to support async state
static boolean supportsAsyncState(Object builder) {
    return builder instanceof org.apache.flink.table.runtime.operators.aggregate.window.WindowAggOperatorBuilder
        || builder instanceof org.apache.flink.table.runtime.operators.join.window.WindowJoinOperatorBuilder;
}

try {
    transformation.enableAsyncState();
} catch (UnsupportedOperationException e) {
    // log and skip: this operator does not support async state
}

When it happens

Trigger: Calling transformation.enableAsyncState() on a Transformation whose concrete subclass did not override the method, OR enabling async state on a stream that has no keyed context (no preceding keyBy). Triggered by Table/SQL runtime builders (WindowAggOperatorBuilder, WindowJoinOperatorBuilder) that forward the flag when 'table.exec.async-state.enabled' or the async-state execution option is on, then call enableAsyncState() on the underlying OneInput/TwoInputTransformation that lacks support.

Common situations: Setting executionConfig.enableAsyncState() globally or table.exec.async-state.enabled=true while using operators that are not async-state-capable; applying async state to a non-keyed stream; upgrading Flink and assuming all operators now support async state when only a subset do.

Related errors

• Cannot have more than {streamingJobCountLimit} streaming job
• This state is only accessible by functions executed on a Key
• Only support seek at first.

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/41770e0a89be18c6. Report an issue: GitHub.