{"record":{"id":"d2fd8338f08a16de","repo":"apache/pulsar","slug":"schema-type-mismatch-typearg-vs-schematypearg","errorCode":null,"errorMessage":"Schema type mismatch ${typeArg} vs ${schemaTypeArg}","messagePattern":"Schema 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":185,"sourceCode":"            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());\n            }\n        } else {\n            if (!schemaTypeArg.asErasure().isAssignableFrom(typeArg.asErasure())) {\n                throw new IllegalArgumentException(\n                        \"Schema type mismatch \" + typeArg.getActualName() + \" vs \" + schemaTypeArg.getActualName());\n            }\n        }\n    }\n}\n","sourceCodeStart":167,"sourceCodeEnd":196,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/ValidatorUtils.java#L167-L196","documentation":"Thrown by ValidatorUtils.validateSchemaType (invoked from validateSchema) when a custom input Schema's declared type argument is not assignable to the function's input type argument. For input schemas, the Schema<T>'s T must be assignable TO the function's input type, guaranteeing decoded records match the function's parameter type.","triggerScenarios":"validateSchema(schemaType, typeArg, typePool, true) with a custom (non-builtin) schemaType class that implements Schema<T>, where schemaTypeArg.asErasure().isAssignableTo(typeArg.asErasure()) is false — e.g. function input is Foo but the schema is SchemaImpl<Bar>.","commonSituations":"Custom Schema implementation written for a different POJO than the function consumes; function input type evolved (Avro class regenerated) while schema class was not; mixing a JSON schema of one type with a typed function parameter; configs copied between functions.","solutions":["Align the function's input type with the schema's type argument (make the function accept the schema's type).","Update the custom Schema class so its Schema<T> type argument is the function's input type (or a subtype of it).","Use a built-in schema type (e.g. JSON, AVRO) keyed to the correct class instead of the mismatched custom schema."],"exampleFix":"// before\npublic class FooSchema implements Schema<Bar> { ... }\npublic class MyFunction implements Function<Foo, Void> { ... }\n// after\npublic class FooSchema implements Schema<Foo> { ... }","handlingStrategy":"type-guard","validationCode":"static boolean inputSchemaMatches(Class<?> schemaClass, Class<?> inputType) {\n    for (Type t : schemaClass.getGenericInterfaces()) {\n        if (t instanceof ParameterizedType && ((ParameterizedType) t).getRawType() == Schema.class) {\n            Type arg = ((ParameterizedType) t).getActualTypeArguments()[0];\n            return arg instanceof Class<?> && ((Class<?>) arg).isAssignableFrom(inputType);\n        }\n    }\n    return false;\n}","typeGuard":"static <T> boolean isSchemaFor(Class<? extends Schema<T>> schemaClass, Class<T> type) {\n    return Schema.class.isAssignableFrom(schemaClass);\n}","tryCatchPattern":"try {\n    ValidatorUtils.validateSchema(schemaType, typeArg, typePool, true);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage() != null && e.getMessage().startsWith(\"Schema type mismatch\")) {\n        throw new IllegalStateException(\"Input schema \" + schemaType\n            + \" produces a type incompatible with the function's input parameter\", e);\n    }\n    throw e;\n}","preventionTips":["Define the custom Schema's generic parameter from the same POJO class used in the function signature.","Regenerate schema classes (e.g. Avro) and the function signature together.","Prefer built-in schema types (JSON/AVRO) named with the exact class to avoid hand-written generics drift."],"tags":["java","pulsar-functions","schema","type-mismatch"],"backgroundTag":"schema-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"}