apache/pulsar · error · IllegalArgumentException

Window function must take a collection as input

Error message

Window function must take a collection as input

What it means

During initialization of the WindowFunctionExecutor, if the user-provided function object implements java.util.function.Function rather than the Pulsar WindowFunction interface, the executor requires its first type argument to be Collection<T>. This error is thrown when a plain java.util.function.Function is supplied whose input type is not Collection, because windowing needs the entire window's records as a collection. It is a user-code contract violation detected at startup.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java:81

        this.windowManager = this.getWindowManager(this.windowConfig, context);
        this.initialized = true;
        this.start();
    }

    @SuppressWarnings("unchecked")
    private void initializeUserFunction(WindowConfig windowConfig) {
        String actualWindowFunctionClassName = windowConfig.getActualWindowFunctionClassName();
        ClassLoader clsLoader = Thread.currentThread().getContextClassLoader();
        Object userClassObject = Reflections.createInstance(
                actualWindowFunctionClassName,
                clsLoader);
        if (userClassObject instanceof java.util.function.Function) {
            Class<?>[] typeArgs = TypeResolver.resolveRawArguments(
                    java.util.function.Function.class, userClassObject.getClass());
            if (typeArgs[0].equals(Collection.class)) {
                bareWindowFunction = (java.util.function.Function<Collection<T>, X>) userClassObject;
            } else {
                throw new IllegalArgumentException("Window function must take a collection as input");
            }
        } else if (userClassObject instanceof WindowFunction) {
            windowFunction = (WindowFunction<T, X>) userClassObject;
        } else {
            throw new IllegalArgumentException("Window function does not implement the correct interface");
        }
    }

    private WindowConfig getWindowConfigs(Context context) {

        if (!context.getUserConfigValue(WindowConfig.WINDOW_CONFIG_KEY).isPresent()) {
            throw new IllegalArgumentException("Window Configs cannot be found");
        }
        WindowConfig windowConfig = new Gson().fromJson(
                (new Gson().toJson(context.getUserConfigValue(WindowConfig.WINDOW_CONFIG_KEY).get())),
                WindowConfig.class);

        return windowConfig;

View on GitHub (pinned to 820761864e)

Solutions

1. Make the function implement org.apache.pulsar.functions.api.windowing.WindowFunction<T,X> instead of java.util.function.Function
2. Or, if using java.util.function.Function, change its input type parameter to Collection<T> (e.g. java.util.function.Function<Collection<String>, X>)
3. Ensure the class uses concrete generic parameters so TypeResolver can resolve Collection as the first argument
4. Redeploy/restart the function with the corrected class

Example fix

// before
class MyFunc implements java.util.function.Function<String, String> { ... }
// after
class MyFunc implements WindowFunction<String, String> {
    public String process(Collection<String> input, Window window) { ... }
}

boolean ok = java.util.function.Function.class.isAssignableFrom(fnClass)
    && TypeResolver.resolveRawArguments(java.util.function.Function.class, fnClass)[0] == Collection.class;
if (!ok) throw new IllegalArgumentException("window function input must be Collection<T>");

static boolean isCollectionWindowFunction(Class<?> c) {
    return java.util.function.Function.class.isAssignableFrom(c)
        && TypeResolver.resolveRawArguments(java.util.function.Function.class, c)[0].equals(Collection.class);
}

try {
    executor.initialize();
} catch (IllegalArgumentException e) {
    // fix the function's generic signature to accept Collection<T>
    throw new IllegalStateException("bad window function signature: " + e.getMessage(), e);
}

When it happens

Trigger: FunctionConfig supplies a class implementing java.util.function.Function with input type T (not Collection<T>) to a windowed function; initializeUserFunction resolves the raw type arguments via TypeResolver and the first argument fails the equals(Collection.class) check.

Common situations: Developers reusing an existing non-windowed function class in a windowed topology; generics erased or raw types so resolveRawArguments returns the wrong argument; copy-paste from a non-windowing example.

Related errors

• Window function does not implement the correct interface
• Window Configs cannot be found
• Late data topic can be defined only when specifying a timest
• Timestamp extractor class %s must be in class path
• User class must be concrete

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/7ae52fd17f1a2c30. Report an issue: GitHub.