{"record":{"id":"140fda1ee9d5ec6f","repo":"apache/pulsar","slug":"serializer-type-mismatch-typearg-vs-serdetype","errorCode":null,"errorMessage":"Serializer type mismatch ${typeArg} vs ${serDeTypeArg}","messagePattern":"Serializer type mismatch (.+?) vs (.+?)","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java","lineNumber":168,"sourceCode":"            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);\n\n        if (input) {\n            if (!schemaTypeArg.asErasure().isAssignableTo(typeArg.asErasure())) {\n                throw new IllegalArgumentException(\n                        \"Schema type mismatch \" + typeArg.getActualName() + \" vs \" + schemaTypeArg.getActualName());","sourceCodeStart":150,"sourceCodeEnd":186,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java#L150-L186","documentation":"Thrown by ValidatorUtils.validateSerde in the serialization (deser=false) branch when the SerDe's declared type parameter is not a supertype of the function's output type argument. For serialization, the SerDe must be able to accept the function's output type, i.e. serDeTypeArg.asErasure().isAssignableFrom(typeArg.asErasure()) must hold.","triggerScenarios":"validateSerde(inputSerializer, typeArg, typePool, false) where the SerDe implements SerDe<T> and typeArg is not assignable to T — e.g. the function outputs SubEvent but the output SerDe is SerDe<UnrelatedType>; or a SerDe typed to a sibling class that the output does not extend.","commonSituations":"Output type changed in a function refactor while the output SerDe config was left pointing at the old type's SerDe; using a base-class SerDe where the function emits an unrelated type; copy-pasting input SerDe settings into the output SerDe field.","solutions":["Set the function's output type argument to a type assignable to the SerDe's type parameter.","Update the SerDe to implement SerDe<ActualOutputType> (or a supertype of the actual output).","Point the output serializer config at a SerDe written for the current output type."],"exampleFix":"// before\npublic class OutSerde implements SerDe<LegacyOut> { ... }\npublic class MyFunction implements Function<In, NewOut> { ... }\n// after\npublic class OutSerde implements SerDe<NewOut> { ... }","handlingStrategy":"type-guard","validationCode":"static boolean serdeMatchesOutput(String serdeClass, Class<?> outputType, 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<?> && outputType.isAssignableFrom((Class<?>) arg) == false\n                        ? outputType.equals(arg) || ((Class<?>) arg).isAssignableFrom(outputType)\n                        : true;\n            }\n        }\n        return false;\n    } catch (ClassNotFoundException e) { return false; }\n}","typeGuard":"static boolean isSerDeAccepting(Class<?> serdeElementType, Class<?> outputType) {\n    return serdeElementType.isAssignableFrom(outputType);\n}","tryCatchPattern":"try {\n    validateSerde(outputSerializer, typeArg, typePool, false);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage() != null && e.getMessage().startsWith(\"Serializer type mismatch\")) {\n        throw new IllegalStateException(\"Output SerDe \" + outputSerializer\n            + \" cannot serialize function output type; update its generic parameter\", e);\n    }\n    throw e;\n}","preventionTips":["Keep the output SerDe's type parameter a supertype of (or equal to) the function's declared output type.","Update output SerDe config whenever the function's output type changes.","Do not copy input SerDe settings into output serializer fields."],"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-14T05:17:10.506Z"}