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

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.

Source

Thrown at pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java:107

        return getFunctionTypes(functionConfig, typePool.describe(functionConfig.getClassName()).resolve());
    }

    public static TypeDefinition[] getFunctionTypes(FunctionConfig functionConfig, TypeDefinition functionClass) {
        boolean isWindowConfigPresent = functionConfig.getWindowConfig() != null;
        return getFunctionTypes(functionClass, isWindowConfigPresent);
    }

    public static TypeDefinition[] getFunctionTypes(TypeDefinition userClass, boolean isWindowConfigPresent) {
        Class<?> classParent = getFunctionClassParent(userClass, isWindowConfigPresent);
        TypeList.Generic typeArgsList = resolveInterfaceTypeArguments(userClass, classParent);
        TypeDescription.Generic[] typeArgs = new TypeDescription.Generic[2];
        typeArgs[0] = typeArgsList.get(0);
        typeArgs[1] = typeArgsList.get(1);
        // if window function
        if (isWindowConfigPresent) {
            if (classParent.equals(java.util.function.Function.class)) {
                if (!typeArgs[0].asErasure().isAssignableTo(Collection.class)) {
                    throw new IllegalArgumentException("Window function must take a collection as input");
                }
                typeArgs[0] = typeArgs[0].getTypeArguments().get(0);
            }
        }
        if (typeArgs[1].asErasure().isAssignableTo(Record.class)) {
            typeArgs[1] = typeArgs[1].getTypeArguments().get(0);
        }
        if (typeArgs[1].asErasure().isAssignableTo(CompletableFuture.class)) {
            typeArgs[1] = typeArgs[1].getTypeArguments().get(0);
        }
        return typeArgs;
    }

    private static TypeList.Generic resolveInterfaceTypeArguments(TypeDefinition userClass, Class<?> interfaceClass) {
        if (!interfaceClass.isInterface()) {
            throw new IllegalArgumentException("interfaceClass must be an interface");
        }
        for (TypeDescription.Generic interfaze : userClass.getInterfaces()) {

View on GitHub (pinned to 820761864e)

Solutions

1. Change the windowed function's input type parameter to a Collection subtype, e.g. java.util.function.Function<Collection<T>, O>
2. Remove the windowConfig if per-message processing was intended
3. Ensure generics are concrete (not raw types) so type inspection can resolve the actual input type

Example fix

// before
public class MyWindowFn implements Function<String, Long> { ... }
// after
public class MyWindowFn implements Function<Collection<String>, Long> { ... }

// Before submitting a windowed function, check the input generic
Type t = fn.getClass().getGenericInterfaces() // resolve java.util.function.Function<T, R>
// and require:
// Class<?> input = typeArg(fnClass, 0);
// if (!Collection.class.isAssignableFrom(input))
//     throw new IllegalArgumentException("Window functions must accept Collection<T>");

static boolean isValidWindowFunction(Class<?> fn) {
    for (Type iface : fn.getGenericInterfaces()) {
        if (iface instanceof ParameterizedType
                && ((ParameterizedType) iface).getRawType() == java.util.function.Function.class) {
            Type input = ((ParameterizedType) iface).getActualTypeArguments()[0];
            if (input instanceof Class) {
                return Collection.class.isAssignableFrom((Class<?>) input);
            }
        }
    }
    return false;
}

try {
    FunctionConfig validated = FunctionConfigUtils.validateUpdate(existing, proposed);
    admin.functions().createFunction(validated);
} catch (IllegalArgumentException e) {
    if (e.getMessage().contains("collection")) {
        log.error("Window function signature invalid: input must be Collection<T>", e);
        // fix the class signature and resubmit
    }
    throw e;
}

When it happens

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

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

Related errors

• Invalid Java object for schema type + type + : + val.getCl
• Window function must take a collection as input
• Kubernetes Secret should be a Map containing id/key pairs
• Sliding interval must be positive [${slidingIntervalCount}]
• Sliding interval must be positive [${slidingIntervalDuration

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