apache/flink · error · UnsupportedOperationException

This state is only accessible by functions executed on a Key

Error message

This state is only accessible by functions executed on a KeyedStream

What it means

Thrown by AbstractRuntimeUDFContext.getState(ValueStateDescriptor) when the function's RuntimeContext is not a keyed streaming context. Keyed state (ValueState, ListState, etc.) requires the function to operate on a KeyedStream — i.e. the upstream was keyBy()'d. The base RuntimeUDFContext throws because it does not provide keyed state.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java:202

            AccumulatorHelper.compareAccumulatorTypes(
                    name, accumulator.getClass(), accumulatorClass);
        } else {
            // Create new accumulator
            try {
                accumulator = accumulatorClass.newInstance();
            } catch (Exception e) {
                throw new RuntimeException(
                        "Cannot create accumulator " + accumulatorClass.getName());
            }
            accumulators.put(name, accumulator);
        }
        return (Accumulator<V, A>) accumulator;
    }

    @Override
    @PublicEvolving
    public <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties) {
        throw new UnsupportedOperationException(
                "This state is only accessible by functions executed on a KeyedStream");
    }

    @Override
    @PublicEvolving
    public <T> ListState<T> getListState(ListStateDescriptor<T> stateProperties) {
        throw new UnsupportedOperationException(
                "This state is only accessible by functions executed on a KeyedStream");
    }

    @Override
    @PublicEvolving
    public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) {
        throw new UnsupportedOperationException(
                "This state is only accessible by functions executed on a KeyedStream");
    }

    @Override

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Call .keyBy(keySelector) on the DataStream before the operator that uses keyed state.
2. If global (non-keyed) state is needed, use a KeyedStream with a constant key or a broadcast state pattern, or use OperatorState (getListState on OperatorStateStore).
3. Move the stateful operator downstream of the keyBy() call.

Example fix

// before — keyed state on non-keyed stream
stream.map(new MyStatefulMapper()); // calls getState() -> throws

// after — key first
stream
    .keyBy(event -> event.getUserId())
    .map(new MyStatefulMapper()); // now keyed, getState() works

// Ensure the function is placed after keyBy() in the pipeline
// Verify at design time: the operator using ValueState must be downstream of keyBy()
// Example pipeline:
stream.keyBy(e -> e.getKey()).map(new MyStatefulMapper());

When it happens

Trigger: A RichFunction calls getRuntimeContext().getState(new ValueStateDescriptor<>(...)) without a preceding keyBy() on the input DataStream. The context is a plain RuntimeUDFContext rather than a StreamingRuntimeContext backed by a keyed operator.

Common situations: Using keyed state in a map/flatMap/filter before calling keyBy(). Applying a ProcessFunction with state on a non-keyed stream. Forgetting to keyBy() when migrating from global-state patterns.

Related errors

• The runtime context has not been initialized.
• The accumulator '${name}' already exists and cannot be added
• Cannot create accumulator ${accumulatorClass.getName()}
• The broadcast variable with name '${name}' is not a List. A
• The broadcast variable with name '${name}' has not been set.

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