{"record":{"id":"ca5ad2a497d418c5","repo":"apache/pulsar","slug":"sink-transform-function-output-must-be-of-type-rec","errorCode":null,"errorMessage":"Sink transform function output must be of type Record","messagePattern":"Sink transform function output must be of type Record","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java","lineNumber":509,"sourceCode":"                if (functionDefinition == null) {\n                    throw new IllegalArgumentException(\n                            \"Function package doesn't contain the META-INF/services/pulsar-io.yaml file.\");\n                }\n                functionClassName = functionDefinition.getFunctionClass();\n                if (functionClassName == null) {\n                    throw new IllegalArgumentException(\"Transform function class name must be set\");\n                }\n            }\n            TypeDefinition functionClass;\n            try {\n                functionClass = transformFunction.resolveType(functionClassName);\n            } catch (TypePool.Resolution.NoSuchTypeException e) {\n                throw new IllegalArgumentException(\n                        String.format(\"Function class %s not found\", functionClassName), e);\n            }\n            // extract type from transform function class\n            if (!getRawFunctionTypes(functionClass, false)[1].asErasure().isAssignableTo(Record.class)) {\n                throw new IllegalArgumentException(\"Sink transform function output must be of type Record\");\n            }\n            typeArg = getFunctionTypes(functionClass, false)[0];\n            inputFunction = transformFunction;\n        } else {\n            // extract type from sink class\n            typeArg = getSinkType(sinkClass);\n            inputFunction = sinkFunction;\n        }\n\n        if (sinkConfig.getTopicToSerdeClassName() != null) {\n            for (String serdeClassName : sinkConfig.getTopicToSerdeClassName().values()) {\n                ValidatorUtils.validateSerde(serdeClassName, typeArg, inputFunction.getTypePool(), true);\n            }\n        }\n\n        if (sinkConfig.getTopicToSchemaType() != null) {\n            for (String schemaType : sinkConfig.getTopicToSchemaType().values()) {\n                ValidatorUtils.validateSchema(schemaType, typeArg, inputFunction.getTypePool(), true);","sourceCodeStart":491,"sourceCodeEnd":527,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java#L491-L527","documentation":"During sink validation, the transform function's output type (element 1 of its raw generic types) must be assignable to org.apache.pulsar.functions.api.Record. Thrown when the transform function's Output type argument is not a Record subtype, so the sink cannot consume its output.","triggerScenarios":"Declaring a transform function for a sink whose output generic parameter is a plain type (e.g. String, POJO) instead of Record<T> (or a subtype).","commonSituations":"Reusing a plain Function<T,R> style class as a sink transform; generics erasure hiding the Record bound; migrating a function written for pipeline use into a sink transform slot.","solutions":["Change the transform function's output type parameter to extend Record, e.g. class MyTransform implements Function<String, Record<byte[]>>","Return Record implementations (e.g. Record.build(...)) from the transform's process method","Pick a transform function designed for sinks if the output must stay non-Record"],"exampleFix":"// before\nclass MyTransform implements Function<String, String> { ... }\n// after\nclass MyTransform implements Function<String, Record<String>> {\n  public Record<String> process(String input, Context ctx) { return Record.build(input); }\n}","handlingStrategy":"type-guard","validationCode":"// statically: ensure output type parameter extends Record\nclass MyTransform implements org.apache.pulsar.functions.api.Function<String, Record<String>> {}","typeGuard":"static boolean isRecordOutput(Class<?> fn) {\n  for (java.lang.reflect.Type t : fn.getGenericInterfaces()) {\n    if (t instanceof java.lang.reflect.ParameterizedType) {\n      java.lang.reflect.Type out = ((java.lang.reflect.ParameterizedType) t).getActualTypeArguments()[1];\n      return out.getTypeName().startsWith(\"org.apache.pulsar.functions.api.Record\");\n    }\n  }\n  return false;\n}","tryCatchPattern":"try {\n  admin.sinks().createSink(config, pkgPath);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage() != null && e.getMessage().contains(\"must be of type Record\")) {\n    log.error(\"Change transform function output type parameter to Record<T>\", e);\n  } else { throw e; }\n}","preventionTips":["Always declare transform output as Record<T> when used in sink pipelines","Compile-time check: Function<I, Record<O>> generics"],"tags":["pulsar-functions","sink","generics","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"}