apache/flink · error · UnsupportedOperationException

This state is only accessible by functions executed on a…

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.

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

• Cannot create accumulator
• The accumulator ' ' already exists and cannot be added.
• The broadcast variable with name
• The broadcast variable with name
• The runtime context has not been initialized.

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

Appendix: 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)