apache/flink · error · InvalidTypesException

Input mismatch: {}

Error message

Input mismatch: {}

What it means

Thrown during input type validation when the declared TypeInformation for an input does not match the type extracted from the function's signature via reflection. This is a wrapper exception — it catches the detailed InvalidTypesException from `validateInfo` and re-throws it prefixed with 'Input mismatch:'. This happens when Flink's framework checks that the function's expected input type aligns with the actual stream type.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java:1482

    //  Validate input
    // --------------------------------------------------------------------------------------------

    private static void validateInputType(
            Class<?> baseClass, Class<?> clazz, int inputParamPos, TypeInformation<?> inTypeInfo) {
        List<Type> typeHierarchy = new ArrayList<>();

        // try to get generic parameter
        Type inType;
        try {
            inType = getParameterType(baseClass, typeHierarchy, clazz, inputParamPos);
        } catch (InvalidTypesException e) {
            return; // skip input validation e.g. for raw types
        }

        try {
            validateInfo(typeHierarchy, inType, inTypeInfo);
        } catch (InvalidTypesException e) {
            throw new InvalidTypesException("Input mismatch: " + e.getMessage(), e);
        }
    }

    @SuppressWarnings("unchecked")
    private static void validateInfo(
            List<Type> typeHierarchy, Type type, TypeInformation<?> typeInfo) {
        if (type == null) {
            throw new InvalidTypesException("Unknown Error. Type is null.");
        }

        if (typeInfo == null) {
            throw new InvalidTypesException("Unknown Error. TypeInformation is null.");
        }

        if (!(type instanceof TypeVariable<?>)) {
            // check for Java Basic Types
            if (typeInfo instanceof BasicTypeInfo) {

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Ensure the function's input type parameter matches the DataStream's element type
2. Add an explicit conversion map upstream to transform the data to the expected type
3. Fix the function signature to use the correct input type
4. If the mismatch is expected, use a cast or intermediate map to convert types

Example fix

// before — type mismatch
DataStream<String> stream = ...;
stream.map(new MapFunction<Integer, String>() { ... }); // wrong input type

// after — align types
DataStream<String> stream = ...;
stream.map(new MapFunction<String, MyEvent>() { ... });

// Pre-validate that function input type matches stream type
TypeInformation<?> functionInputType = TypeExtractor.getMapReturnTypes(
    myMapFunction, streamType);  // will throw if mismatch
// Or check manually:
TypeInformation<?> expectedInput = TypeInformation.of(String.class);
if (!stream.getType().equals(expectedInput)) {
    throw new IllegalStateException(
        "Stream type " + stream.getType() + " does not match function input " + expectedInput);
}

When it happens

Trigger: A function declares one input type via its generic signature (e.g. `MapFunction<String, X>`) but the DataStream it is applied to carries a different type (e.g. `DataStream<Integer>`). The validateInputType method extracts the function's declared input parameter type and compares it against the DataStream's TypeInformation.

Common situations: Connecting a `MapFunction<String, X>` to a `DataStream<Integer>` stream. Type mismatches after refactoring function signatures without updating the pipeline. Mismatched type parameters when composing operators from different libraries.

Related errors

• No lambda method found.
• Could not extract lambda method out of function: {} - {}
• Given class: {}is not a FunctionalInterface.
• Cannot convert type to class
• Tuple needs to be parameterized by using generics.

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