{"record":{"id":"7ae52fd17f1a2c30","repo":"apache/pulsar","slug":"window-function-must-take-a-collection-as-input","errorCode":null,"errorMessage":"Window function must take a collection as input","messagePattern":"Window function must take a collection as input","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java","lineNumber":81,"sourceCode":"        this.windowManager = this.getWindowManager(this.windowConfig, context);\n        this.initialized = true;\n        this.start();\n    }\n\n    @SuppressWarnings(\"unchecked\")\n    private void initializeUserFunction(WindowConfig windowConfig) {\n        String actualWindowFunctionClassName = windowConfig.getActualWindowFunctionClassName();\n        ClassLoader clsLoader = Thread.currentThread().getContextClassLoader();\n        Object userClassObject = Reflections.createInstance(\n                actualWindowFunctionClassName,\n                clsLoader);\n        if (userClassObject instanceof java.util.function.Function) {\n            Class<?>[] typeArgs = TypeResolver.resolveRawArguments(\n                    java.util.function.Function.class, userClassObject.getClass());\n            if (typeArgs[0].equals(Collection.class)) {\n                bareWindowFunction = (java.util.function.Function<Collection<T>, X>) userClassObject;\n            } else {\n                throw new IllegalArgumentException(\"Window function must take a collection as input\");\n            }\n        } else if (userClassObject instanceof WindowFunction) {\n            windowFunction = (WindowFunction<T, X>) userClassObject;\n        } else {\n            throw new IllegalArgumentException(\"Window function does not implement the correct interface\");\n        }\n    }\n\n    private WindowConfig getWindowConfigs(Context context) {\n\n        if (!context.getUserConfigValue(WindowConfig.WINDOW_CONFIG_KEY).isPresent()) {\n            throw new IllegalArgumentException(\"Window Configs cannot be found\");\n        }\n        WindowConfig windowConfig = new Gson().fromJson(\n                (new Gson().toJson(context.getUserConfigValue(WindowConfig.WINDOW_CONFIG_KEY).get())),\n                WindowConfig.class);\n\n        return windowConfig;","sourceCodeStart":63,"sourceCodeEnd":99,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java#L63-L99","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Make the function implement org.apache.pulsar.functions.api.windowing.WindowFunction<T,X> instead of java.util.function.Function","Or, if using java.util.function.Function, change its input type parameter to Collection<T> (e.g. java.util.function.Function<Collection<String>, X>)","Ensure the class uses concrete generic parameters so TypeResolver can resolve Collection as the first argument","Redeploy/restart the function with the corrected class"],"exampleFix":"// before\nclass MyFunc implements java.util.function.Function<String, String> { ... }\n// after\nclass MyFunc implements WindowFunction<String, String> {\n    public String process(Collection<String> input, Window window) { ... }\n}","handlingStrategy":"type-guard","validationCode":"boolean ok = java.util.function.Function.class.isAssignableFrom(fnClass)\n    && TypeResolver.resolveRawArguments(java.util.function.Function.class, fnClass)[0] == Collection.class;\nif (!ok) throw new IllegalArgumentException(\"window function input must be Collection<T>\");","typeGuard":"static boolean isCollectionWindowFunction(Class<?> c) {\n    return java.util.function.Function.class.isAssignableFrom(c)\n        && TypeResolver.resolveRawArguments(java.util.function.Function.class, c)[0].equals(Collection.class);\n}","tryCatchPattern":"try {\n    executor.initialize();\n} catch (IllegalArgumentException e) {\n    // fix the function's generic signature to accept Collection<T>\n    throw new IllegalStateException(\"bad window function signature: \" + e.getMessage(), e);\n}","preventionTips":["Implement WindowFunction<T,X> for windowed functions rather than raw java.util.function.Function","Always use concrete generic parameters so TypeResolver can resolve them","Unit-test function initialization before deploying"],"tags":["pulsar-functions","windowing","type-mismatch"],"backgroundTag":"wrong-function-interface","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}