{"record":{"id":"4a19fa807254b300","repo":"apache/flink","slug":"this-state-is-only-accessible-by-functions-execute","errorCode":null,"errorMessage":"This state is only accessible by functions executed on a KeyedStream","messagePattern":"This state is only accessible by functions executed on a KeyedStream","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java","lineNumber":202,"sourceCode":"            AccumulatorHelper.compareAccumulatorTypes(\n                    name, accumulator.getClass(), accumulatorClass);\n        } else {\n            // Create new accumulator\n            try {\n                accumulator = accumulatorClass.newInstance();\n            } catch (Exception e) {\n                throw new RuntimeException(\n                        \"Cannot create accumulator \" + accumulatorClass.getName());\n            }\n            accumulators.put(name, accumulator);\n        }\n        return (Accumulator<V, A>) accumulator;\n    }\n\n    @Override\n    @PublicEvolving\n    public <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties) {\n        throw new UnsupportedOperationException(\n                \"This state is only accessible by functions executed on a KeyedStream\");\n    }\n\n    @Override\n    @PublicEvolving\n    public <T> ListState<T> getListState(ListStateDescriptor<T> stateProperties) {\n        throw new UnsupportedOperationException(\n                \"This state is only accessible by functions executed on a KeyedStream\");\n    }\n\n    @Override\n    @PublicEvolving\n    public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) {\n        throw new UnsupportedOperationException(\n                \"This state is only accessible by functions executed on a KeyedStream\");\n    }\n\n    @Override","sourceCodeStart":184,"sourceCodeEnd":220,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java#L184-L220","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Call .keyBy(keySelector) on the DataStream before the operator that uses keyed state.","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).","Move the stateful operator downstream of the keyBy() call."],"exampleFix":"// before — keyed state on non-keyed stream\nstream.map(new MyStatefulMapper()); // calls getState() -> throws\n\n// after — key first\nstream\n    .keyBy(event -> event.getUserId())\n    .map(new MyStatefulMapper()); // now keyed, getState() works","handlingStrategy":"validation","validationCode":"// Ensure the function is placed after keyBy() in the pipeline\n// Verify at design time: the operator using ValueState must be downstream of keyBy()\n// Example pipeline:\nstream.keyBy(e -> e.getKey()).map(new MyStatefulMapper());","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always call keyBy() before any operator that uses keyed state.","Review the Flink job graph to confirm a hash/key partition exists before stateful operators.","Initialize state descriptors in open(), and if you see this error, add keyBy() upstream."],"tags":["keyed-state","state","valuestate","runtime-context","user-error"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}