{"record":{"id":"b2cf4c601b108ee2","repo":"apache/pulsar","slug":"serializer-type-mismatch","errorCode":null,"errorMessage":"Serializer type mismatch ","messagePattern":"Serializer type mismatch ","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java","lineNumber":163,"sourceCode":"        }\n        TypeDescription serdeClass;\n        try {\n            serdeClass = typePool.describe(inputSerializer).resolve();\n        } catch (TypePool.Resolution.NoSuchTypeException e) {\n            throw new IllegalArgumentException(\n                    String.format(\"The input serialization/deserialization class %s does not exist\",\n                            inputSerializer));\n        }\n        TypeDescription.Generic serDeTypeArg = serdeClass.getInterfaces().stream()\n                .filter(i -> i.asErasure().isAssignableTo(SerDe.class))\n                .findFirst()\n                .map(i -> i.getTypeArguments().get(0))\n                .orElseThrow(() -> new IllegalArgumentException(\n                        String.format(\"%s does not implement %s\", inputSerializer, SerDe.class.getName())));\n\n        if (deser) {\n            if (!serDeTypeArg.asErasure().isAssignableTo(typeArg.asErasure())) {\n                throw new IllegalArgumentException(\"Serializer type mismatch \" + typeArg.getActualName() + \" vs \"\n                        + serDeTypeArg.getActualName());\n            }\n        } else {\n            if (!serDeTypeArg.asErasure().isAssignableFrom(typeArg.asErasure())) {\n                throw new IllegalArgumentException(\"Serializer type mismatch \" + typeArg.getActualName() + \" vs \"\n                        + serDeTypeArg.getActualName());\n            }\n        }\n    }\n\n    private static void validateSchemaType(TypeDefinition schema, TypeDefinition typeArg, TypePool typePool,\n                                           boolean input) {\n\n        TypeDescription.Generic schemaTypeArg = schema.getInterfaces().stream()\n                .filter(i -> i.asErasure().isAssignableTo(Schema.class))\n                .findFirst()\n                .map(i -> i.getTypeArguments().get(0))\n                .orElse(null);","sourceCodeStart":145,"sourceCodeEnd":181,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java#L145-L181","documentation":"Thrown by ValidatorUtils.validateSerde when the SerDe's declared type parameter is not compatible with the function's input type argument (deser=true branch). For deserialization, the SerDe's type argument must be assignable TO the function's input type; otherwise the objects the SerDe produces would not match what the function expects.","triggerScenarios":"validateSerde(inputSerializer, typeArg, typePool, true) where the SerDe class implements SerDe<T> and T.asErasure().isAssignableTo(typeArg.asErasure()) is false — e.g. the function's input type is MyEvent but the SerDe is SerDe<OtherEvent> or SerDe<Object> paired with a narrower input type.","commonSituations":"Reusing a SerDe written for a different message class after changing the function's input type; generic SerDe (SerDe<Object> or SerDe<byte[]>) paired with a typed function input; schema/tenant config drift after evolving the function signature; copy-pasted function configs pointing to the previous version's SerDe.","solutions":["Make the function's input type argument match (or be a supertype of) the SerDe's type parameter — e.g. change the function signature to accept the SerDe's element type.","Change the SerDe's type parameter so it implements SerDe<YourInputType>.","Regenerate or update the SerDe when the message class changes so the generic type stays in sync."],"exampleFix":"// before\npublic class EventSerde implements SerDe<OldEvent> { ... }\npublic class MyFunction implements Function<NewEvent, Void> { ... }\n// after\npublic class EventSerde implements SerDe<NewEvent> { ... }","handlingStrategy":"type-guard","validationCode":"static boolean serdeMatchesInput(String serdeClass, Class<?> inputType, ClassLoader cl) {\n    try {\n        for (Type t : Class.forName(serdeClass, false, cl).getGenericInterfaces()) {\n            if (t instanceof ParameterizedType\n                    && ((ParameterizedType) t).getRawType() == SerDe.class) {\n                Type arg = ((ParameterizedType) t).getActualTypeArguments()[0];\n                return arg instanceof Class<?> && ((Class<?>) arg).isAssignableFrom(inputType);\n            }\n        }\n        return false;\n    } catch (ClassNotFoundException e) { return false; }\n}","typeGuard":"static <T> boolean isSerDeFor(Class<? extends SerDe<T>> serdeClass, Class<T> inputType) {\n    return SerDe.class.isAssignableFrom(serdeClass);\n}","tryCatchPattern":"try {\n    validateSerde(inputSerializer, typeArg, typePool, true);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage() != null && e.getMessage().startsWith(\"Serializer type mismatch\")) {\n        throw new IllegalStateException(\"Input SerDe \" + inputSerializer\n            + \" element type does not match function input type; align generics\", e);\n    }\n    throw e;\n}","preventionTips":["Declare the function's input type and the SerDe's generic from the same class constant to keep them in lockstep.","Regenerate both the POJO and its SerDe together whenever the message class changes.","Avoid generic SerDe<Object> when the function input is a concrete type."],"tags":["java","pulsar-functions","generics","serde","type-mismatch"],"backgroundTag":"type-mismatch","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"}