apache/flink · error · UnsupportedOperationException

ReduceFunction of ReducingState can not be a RichFunction.

Error message

ReduceFunction of ReducingState can not be a RichFunction.

What it means

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.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/state/ReducingStateDescriptor.java:60

    /**
     * Creates a new {@code ReducingStateDescriptor} with the given name, type, and default value.
     *
     * <p>If this constructor fails (because it is not possible to describe the type via a class),
     * consider using the {@link #ReducingStateDescriptor(String, ReduceFunction, TypeInformation)}
     * constructor.
     *
     * @param name The (unique) name for the state.
     * @param reduceFunction The {@code ReduceFunction} used to aggregate the state.
     * @param typeClass The type of the values in the state.
     */
    public ReducingStateDescriptor(
            String name, ReduceFunction<T> reduceFunction, Class<T> typeClass) {
        super(name, typeClass, null);
        this.reduceFunction = checkNotNull(reduceFunction);

        if (reduceFunction instanceof RichFunction) {
            throw new UnsupportedOperationException(
                    "ReduceFunction of ReducingState can not be a RichFunction.");
        }
    }

    /**
     * Creates a new {@code ReducingStateDescriptor} with the given name and default value.
     *
     * @param name The (unique) name for the state.
     * @param reduceFunction The {@code ReduceFunction} used to aggregate the state.
     * @param typeInfo The type of the values in the state.
     */
    public ReducingStateDescriptor(
            String name, ReduceFunction<T> reduceFunction, TypeInformation<T> typeInfo) {
        super(name, typeInfo, null);
        this.reduceFunction = checkNotNull(reduceFunction);
    }

    /**

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Separate the reduce logic into a plain ReduceFunction (no RichFunction) and pass that to the descriptor.
2. If you need RuntimeContext access (metrics, broadcast state), get it in the containing ProcessFunction and pass needed values to a non-rich ReduceFunction.
3. Do not make the ReduceFunction class extend AbstractRichFunction or implement RichFunction.

Example fix

// before: RichFunction + ReduceFunction
public class MyReducer extends AbstractRichFunction implements ReduceFunction<Long> {
    @Override
    public Long reduce(Long a, Long b) { return a + b; }
}
new ReducingStateDescriptor<>("sum", new MyReducer(), Long.class); // throws

// after: plain ReduceFunction
public class MyReducer implements ReduceFunction<Long> {
    @Override
    public Long reduce(Long a, Long b) { return a + b; }
}
new ReducingStateDescriptor<>("sum", new MyReducer(), Long.class);

ReduceFunction<T> fn = /* ... */;
if (fn instanceof RichFunction) {
    throw new IllegalArgumentException(
        "ReduceFunction for ReducingState must not be a RichFunction: " + fn.getClass());
}
new ReducingStateDescriptor<>("name", fn, typeClass);

public static boolean isPlainReduceFunction(Object fn) {
    return fn instanceof ReduceFunction && !(fn instanceof RichFunction);
}

When it happens

Trigger: 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).

Common situations: 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.

Related errors

• {e.getMessage()}
• Cannot access jar file{t.getMessage() == null ? "." : ": " +
• The jarFile and entryPointClassName can not be null at the s
• Not allowed configuration change(s) were detected:\n - {erro
• Cannot create empty classifier chain.

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