{"record":{"id":"0bd6b1c1e4c95e8c","repo":"apache/flink","slug":"cannot-set-a-udf-as-combinable-if-it-does-not-impl","errorCode":null,"errorMessage":"Cannot set a UDF as combinable if it does not implement the interface {}","messagePattern":"Cannot set a UDF as combinable if it does not implement the interface (.+?)","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java","lineNumber":139,"sourceCode":"     * @return The secondary order.\n     */\n    public Ordering getGroupOrder() {\n        return this.groupOrder;\n    }\n\n    /**\n     * Marks the group reduce operation as combinable. Combinable operations may pre-reduce the data\n     * before the actual group reduce operations. Combinable user-defined functions must implement\n     * the interface {@link GroupCombineFunction}.\n     *\n     * @param combinable Flag to mark the group reduce operation as combinable.\n     */\n    public void setCombinable(boolean combinable) {\n        // sanity check\n        if (combinable\n                && !GroupCombineFunction.class.isAssignableFrom(\n                        this.userFunction.getUserCodeClass())) {\n            throw new IllegalArgumentException(\n                    \"Cannot set a UDF as combinable if it does not implement the interface \"\n                            + GroupCombineFunction.class.getName());\n        } else {\n            this.combinable = combinable;\n        }\n    }\n\n    /**\n     * Checks whether the operation is combinable.\n     *\n     * @return True, if the UDF is combinable, false if not.\n     * @see #setCombinable(boolean)\n     */\n    public boolean isCombinable() {\n        return this.combinable;\n    }\n\n    public void setCustomPartitioner(Partitioner<?> customPartitioner) {","sourceCodeStart":121,"sourceCodeEnd":157,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorBase.java#L121-L157","documentation":"Thrown by GroupReduceOperatorBase.setCombinable(boolean) when combinable is true but the user function class does not implement the GroupCombineFunction interface. A combinable group reduce pre-reduces data using the combine phase; this requires the UDF to implement GroupCombineFunction.combine(Iterable, Collector) so the combiner can call it.","triggerScenarios":"Calling reduceGroup with a function that only extends GroupReduceFunction, then calling setCombinable(true) on the operator. Annotating a non-GroupCombineFunction UDF with @Combinable. The UDF class implements GroupReduceFunction but not GroupCombineFunction.","commonSituations":"Writing a GroupReduceFunction and wanting combiner optimization without implementing the additional GroupCombineFunction interface. Forgetting that combinable requires implementing two interfaces. Copying a reduce function that was not designed for combining and enabling combine on it.","solutions":["Make the UDF implement both GroupReduceFunction and GroupCombineFunction (or extend CombineFunction which bridges them).","If the UDF cannot be made combinable (e.g., it needs full group context), call setCombinable(false) or do not call it.","Use @Combinable annotation only on classes that implement GroupCombineFunction."],"exampleFix":"// before\npublic static class MyReducer extends GroupReduceFunction<MyType, MyType> {\n    public void reduce(Iterable<MyType> values, Collector<MyType> out) { ... }\n}\n// elsewhere\nreduceOp.setCombinable(true); // throws: MyReducer does not implement GroupCombineFunction\n\n// after\npublic static class MyReducer extends RichGroupReduceFunction<MyType, MyType>\n        implements GroupCombineFunction<MyType, MyType> {\n    public void reduce(Iterable<MyType> values, Collector<MyType> out) { ... }\n    public void combine(Iterable<MyType> values, Collector<MyType> out) { reduce(values, out); }\n}\n// elsewhere\nreduceOp.setCombinable(true); // ok","handlingStrategy":"type-guard","validationCode":"void safeSetCombinable(GroupReduceOperatorBase<?, ?, ?> op, boolean combinable) {\n    if (combinable) {\n    Class<?> udfClass = op.getUserFunctionWrapper().getUserCodeClass();\n        if (!GroupCombineFunction.class.isAssignableFrom(udfClass)) {\n            throw new IllegalStateException(\n                \"UDF \" + udfClass.getName() + \" must implement GroupCombineFunction to be combinable\");\n        }\n    }\n    op.setCombinable(combinable);\n}","typeGuard":"static boolean isCombinableCapable(Class<?> udfClass) {\n    return GroupCombineFunction.class.isAssignableFrom(udfClass);\n}","tryCatchPattern":"try {\n    reduceOp.setCombinable(true);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"GroupCombineFunction\")) {\n        // UDF does not implement GroupCombineFunction; either implement it or disable combine\n        log.warn(\"Cannot enable combiner: {}\", e.getMessage());\n    }\n}","preventionTips":["Implement GroupCombineFunction on any UDF you intend to mark combinable.","Use @Combinable annotation only on classes that implement GroupCombineFunction.","Write a unit test asserting the UDF class is assignable from GroupCombineFunction before enabling combine."],"tags":["group-reduce","combiner","interface-check","dataset-api","flink-core"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}