{"record":{"id":"d26f7a889fdd1024","repo":"apache/pulsar","slug":"window-function-must-take-a-collection-as-input-d26f7a","errorCode":null,"errorMessage":"Window function must take a collection as input","messagePattern":"Window function must take a collection as input","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java","lineNumber":107,"sourceCode":"        return getFunctionTypes(functionConfig, typePool.describe(functionConfig.getClassName()).resolve());\n    }\n\n    public static TypeDefinition[] getFunctionTypes(FunctionConfig functionConfig, TypeDefinition functionClass) {\n        boolean isWindowConfigPresent = functionConfig.getWindowConfig() != null;\n        return getFunctionTypes(functionClass, isWindowConfigPresent);\n    }\n\n    public static TypeDefinition[] getFunctionTypes(TypeDefinition userClass, boolean isWindowConfigPresent) {\n        Class<?> classParent = getFunctionClassParent(userClass, isWindowConfigPresent);\n        TypeList.Generic typeArgsList = resolveInterfaceTypeArguments(userClass, classParent);\n        TypeDescription.Generic[] typeArgs = new TypeDescription.Generic[2];\n        typeArgs[0] = typeArgsList.get(0);\n        typeArgs[1] = typeArgsList.get(1);\n        // if window function\n        if (isWindowConfigPresent) {\n            if (classParent.equals(java.util.function.Function.class)) {\n                if (!typeArgs[0].asErasure().isAssignableTo(Collection.class)) {\n                    throw new IllegalArgumentException(\"Window function must take a collection as input\");\n                }\n                typeArgs[0] = typeArgs[0].getTypeArguments().get(0);\n            }\n        }\n        if (typeArgs[1].asErasure().isAssignableTo(Record.class)) {\n            typeArgs[1] = typeArgs[1].getTypeArguments().get(0);\n        }\n        if (typeArgs[1].asErasure().isAssignableTo(CompletableFuture.class)) {\n            typeArgs[1] = typeArgs[1].getTypeArguments().get(0);\n        }\n        return typeArgs;\n    }\n\n    private static TypeList.Generic resolveInterfaceTypeArguments(TypeDefinition userClass, Class<?> interfaceClass) {\n        if (!interfaceClass.isInterface()) {\n            throw new IllegalArgumentException(\"interfaceClass must be an interface\");\n        }\n        for (TypeDescription.Generic interfaze : userClass.getInterfaces()) {","sourceCodeStart":89,"sourceCodeEnd":125,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java#L89-L125","documentation":"FunctionCommon.getFunctionTypes inspects the function's implemented functional interface to determine input/output types. For windowed functions the input must be a Collection (the window of messages); if a window config is present but the function implements java.util.function.Function with an input type that is not assignable to java.util.Collection, this IllegalArgumentException is thrown. It enforces that window functions consume batches, not single messages.","triggerScenarios":"Submitting a function with windowConfig set whose class implements java.util.function.Function<NOT-A-COLLECTION, X> — e.g. Function<String, String> or Function<MyPojo, Void> instead of Function<Collection<MyMessage>, Result>.","commonSituations":"Developers converting an existing plain function to a windowed function without changing the signature; wrong generic parameters due to raw types; confusion between Function and WindowFunction-style interfaces.","solutions":["Change the windowed function's input type parameter to a Collection subtype, e.g. java.util.function.Function<Collection<T>, O>","Remove the windowConfig if per-message processing was intended","Ensure generics are concrete (not raw types) so type inspection can resolve the actual input type"],"exampleFix":"// before\npublic class MyWindowFn implements Function<String, Long> { ... }\n// after\npublic class MyWindowFn implements Function<Collection<String>, Long> { ... }","handlingStrategy":"validation","validationCode":"// Before submitting a windowed function, check the input generic\nType t = fn.getClass().getGenericInterfaces() // resolve java.util.function.Function<T, R>\n// and require:\n// Class<?> input = typeArg(fnClass, 0);\n// if (!Collection.class.isAssignableFrom(input))\n//     throw new IllegalArgumentException(\"Window functions must accept Collection<T>\");","typeGuard":"static boolean isValidWindowFunction(Class<?> fn) {\n    for (Type iface : fn.getGenericInterfaces()) {\n        if (iface instanceof ParameterizedType\n                && ((ParameterizedType) iface).getRawType() == java.util.function.Function.class) {\n            Type input = ((ParameterizedType) iface).getActualTypeArguments()[0];\n            if (input instanceof Class) {\n                return Collection.class.isAssignableFrom((Class<?>) input);\n            }\n        }\n    }\n    return false;\n}","tryCatchPattern":"try {\n    FunctionConfig validated = FunctionConfigUtils.validateUpdate(existing, proposed);\n    admin.functions().createFunction(validated);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().contains(\"collection\")) {\n        log.error(\"Window function signature invalid: input must be Collection<T>\", e);\n        // fix the class signature and resubmit\n    }\n    throw e;\n}","preventionTips":["Declare windowed functions as Function<Collection<T>, O> from the start","Do not convert a plain streaming function to windowed without changing its signature","Avoid raw types so generic inspection resolves real parameter types","Validate function signatures in CI before submitting to Pulsar"],"tags":["function-config","windowing","type-mismatch","validation"],"backgroundTag":"invalid-function-signature","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"}