{"record":{"id":"a1f75e0a8b1482ec","repo":"apache/flink","slug":"input-mismatch","errorCode":null,"errorMessage":"Input mismatch: {}","messagePattern":"Input mismatch: (.+?)","errorType":"exception","errorClass":"InvalidTypesException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java","lineNumber":1482,"sourceCode":"    //  Validate input\n    // --------------------------------------------------------------------------------------------\n\n    private static void validateInputType(\n            Class<?> baseClass, Class<?> clazz, int inputParamPos, TypeInformation<?> inTypeInfo) {\n        List<Type> typeHierarchy = new ArrayList<>();\n\n        // try to get generic parameter\n        Type inType;\n        try {\n            inType = getParameterType(baseClass, typeHierarchy, clazz, inputParamPos);\n        } catch (InvalidTypesException e) {\n            return; // skip input validation e.g. for raw types\n        }\n\n        try {\n            validateInfo(typeHierarchy, inType, inTypeInfo);\n        } catch (InvalidTypesException e) {\n            throw new InvalidTypesException(\"Input mismatch: \" + e.getMessage(), e);\n        }\n    }\n\n    @SuppressWarnings(\"unchecked\")\n    private static void validateInfo(\n            List<Type> typeHierarchy, Type type, TypeInformation<?> typeInfo) {\n        if (type == null) {\n            throw new InvalidTypesException(\"Unknown Error. Type is null.\");\n        }\n\n        if (typeInfo == null) {\n            throw new InvalidTypesException(\"Unknown Error. TypeInformation is null.\");\n        }\n\n        if (!(type instanceof TypeVariable<?>)) {\n            // check for Java Basic Types\n            if (typeInfo instanceof BasicTypeInfo) {\n","sourceCodeStart":1464,"sourceCodeEnd":1500,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractor.java#L1464-L1500","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure the function's input type parameter matches the DataStream's element type","Add an explicit conversion map upstream to transform the data to the expected type","Fix the function signature to use the correct input type","If the mismatch is expected, use a cast or intermediate map to convert types"],"exampleFix":"// before — type mismatch\nDataStream<String> stream = ...;\nstream.map(new MapFunction<Integer, String>() { ... }); // wrong input type\n\n// after — align types\nDataStream<String> stream = ...;\nstream.map(new MapFunction<String, MyEvent>() { ... });","handlingStrategy":"validation","validationCode":"// Pre-validate that function input type matches stream type\nTypeInformation<?> functionInputType = TypeExtractor.getMapReturnTypes(\n    myMapFunction, streamType);  // will throw if mismatch\n// Or check manually:\nTypeInformation<?> expectedInput = TypeInformation.of(String.class);\nif (!stream.getType().equals(expectedInput)) {\n    throw new IllegalStateException(\n        \"Stream type \" + stream.getType() + \" does not match function input \" + expectedInput);\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always match function input type parameters to the upstream stream's element type","Use `.map(...)` with functions whose IN matches the DataStream<T> type parameter T","Add intermediate map operators to convert types explicitly","Run type checking early in the pipeline to catch mismatches at composition time"],"tags":["type-validation","input-mismatch","type-extraction","reflection"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}