{"record":{"id":"e81187d860e7e2dc","repo":"apache/flink","slug":"reducefunction-of-reducingstate-can-not-be-a-richf","errorCode":null,"errorMessage":"ReduceFunction of ReducingState can not be a RichFunction.","messagePattern":"ReduceFunction of ReducingState can not be a RichFunction\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/state/ReducingStateDescriptor.java","lineNumber":60,"sourceCode":"\n    /**\n     * Creates a new {@code ReducingStateDescriptor} with the given name, type, and default value.\n     *\n     * <p>If this constructor fails (because it is not possible to describe the type via a class),\n     * consider using the {@link #ReducingStateDescriptor(String, ReduceFunction, TypeInformation)}\n     * constructor.\n     *\n     * @param name The (unique) name for the state.\n     * @param reduceFunction The {@code ReduceFunction} used to aggregate the state.\n     * @param typeClass The type of the values in the state.\n     */\n    public ReducingStateDescriptor(\n            String name, ReduceFunction<T> reduceFunction, Class<T> typeClass) {\n        super(name, typeClass, null);\n        this.reduceFunction = checkNotNull(reduceFunction);\n\n        if (reduceFunction instanceof RichFunction) {\n            throw new UnsupportedOperationException(\n                    \"ReduceFunction of ReducingState can not be a RichFunction.\");\n        }\n    }\n\n    /**\n     * Creates a new {@code ReducingStateDescriptor} with the given name and default value.\n     *\n     * @param name The (unique) name for the state.\n     * @param reduceFunction The {@code ReduceFunction} used to aggregate the state.\n     * @param typeInfo The type of the values in the state.\n     */\n    public ReducingStateDescriptor(\n            String name, ReduceFunction<T> reduceFunction, TypeInformation<T> typeInfo) {\n        super(name, typeInfo, null);\n        this.reduceFunction = checkNotNull(reduceFunction);\n    }\n\n    /**","sourceCodeStart":42,"sourceCodeEnd":78,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/state/ReducingStateDescriptor.java#L42-L78","documentation":"Thrown by the ReducingStateDescriptor constructor (Class-based overload) when the provided ReduceFunction is also a RichFunction. RichFunction implementations have a RuntimeContext with open()/close() lifecycle methods that the reducing state cannot invoke, because the reduce function is applied directly during state updates without function lifecycle management. The instanceof check fires immediately after the null check, before the descriptor is usable.","triggerScenarios":"Passing a ReduceFunction that also implements RichFunction (e.g. a class extending AbstractRichFunction and implementing ReduceFunction) to the ReducingStateDescriptor constructor. The check is on the Class-based overload; the TypeInformation and TypeSerializer overloads do not have this guard (though using a RichFunction there is equally unsupported).","commonSituations":"Reusing a RichFunction-based reduce function intended for DataStream.reduce() in a ReducingStateDescriptor; writing a ReduceFunction that needs RuntimeContext (e.g. for broadcasting state or metrics) and accidentally extending RichFunction.","solutions":["Separate the reduce logic into a plain ReduceFunction (no RichFunction) and pass that to the descriptor.","If you need RuntimeContext access (metrics, broadcast state), get it in the containing ProcessFunction and pass needed values to a non-rich ReduceFunction.","Do not make the ReduceFunction class extend AbstractRichFunction or implement RichFunction."],"exampleFix":"// before: RichFunction + ReduceFunction\npublic class MyReducer extends AbstractRichFunction implements ReduceFunction<Long> {\n    @Override\n    public Long reduce(Long a, Long b) { return a + b; }\n}\nnew ReducingStateDescriptor<>(\"sum\", new MyReducer(), Long.class); // throws\n\n// after: plain ReduceFunction\npublic class MyReducer implements ReduceFunction<Long> {\n    @Override\n    public Long reduce(Long a, Long b) { return a + b; }\n}\nnew ReducingStateDescriptor<>(\"sum\", new MyReducer(), Long.class);","handlingStrategy":"validation","validationCode":"ReduceFunction<T> fn = /* ... */;\nif (fn instanceof RichFunction) {\n    throw new IllegalArgumentException(\n        \"ReduceFunction for ReducingState must not be a RichFunction: \" + fn.getClass());\n}\nnew ReducingStateDescriptor<>(\"name\", fn, typeClass);","typeGuard":"public static boolean isPlainReduceFunction(Object fn) {\n    return fn instanceof ReduceFunction && !(fn instanceof RichFunction);\n}","tryCatchPattern":null,"preventionTips":["Implement ReduceFunction without extending RichFunction/AbstractRichFunction.","Access RuntimeContext in the enclosing ProcessFunction, not in the reduce function.","Unit-test the descriptor construction to catch RichFunction usage early."],"tags":["state","reducing-state","rich-function","validation"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}