apache/pulsar · error · IllegalArgumentException

Sink transform function output must be of type Record

Error message

Sink transform function output must be of type Record

What it means

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.

Source

Thrown at pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java:509

                if (functionDefinition == null) {
                    throw new IllegalArgumentException(
                            "Function package doesn't contain the META-INF/services/pulsar-io.yaml file.");
                }
                functionClassName = functionDefinition.getFunctionClass();
                if (functionClassName == null) {
                    throw new IllegalArgumentException("Transform function class name must be set");
                }
            }
            TypeDefinition functionClass;
            try {
                functionClass = transformFunction.resolveType(functionClassName);
            } catch (TypePool.Resolution.NoSuchTypeException e) {
                throw new IllegalArgumentException(
                        String.format("Function class %s not found", functionClassName), e);
            }
            // extract type from transform function class
            if (!getRawFunctionTypes(functionClass, false)[1].asErasure().isAssignableTo(Record.class)) {
                throw new IllegalArgumentException("Sink transform function output must be of type Record");
            }
            typeArg = getFunctionTypes(functionClass, false)[0];
            inputFunction = transformFunction;
        } else {
            // extract type from sink class
            typeArg = getSinkType(sinkClass);
            inputFunction = sinkFunction;
        }

        if (sinkConfig.getTopicToSerdeClassName() != null) {
            for (String serdeClassName : sinkConfig.getTopicToSerdeClassName().values()) {
                ValidatorUtils.validateSerde(serdeClassName, typeArg, inputFunction.getTypePool(), true);
            }
        }

        if (sinkConfig.getTopicToSchemaType() != null) {
            for (String schemaType : sinkConfig.getTopicToSchemaType().values()) {
                ValidatorUtils.validateSchema(schemaType, typeArg, inputFunction.getTypePool(), true);

View on GitHub (pinned to 820761864e)

Solutions

1. Change the transform function's output type parameter to extend Record, e.g. class MyTransform implements Function<String, Record<byte[]>>
2. Return Record implementations (e.g. Record.build(...)) from the transform's process method
3. Pick a transform function designed for sinks if the output must stay non-Record

Example fix

// before
class MyTransform implements Function<String, String> { ... }
// after
class MyTransform implements Function<String, Record<String>> {
  public Record<String> process(String input, Context ctx) { return Record.build(input); }
}

// statically: ensure output type parameter extends Record
class MyTransform implements org.apache.pulsar.functions.api.Function<String, Record<String>> {}

static boolean isRecordOutput(Class<?> fn) {
  for (java.lang.reflect.Type t : fn.getGenericInterfaces()) {
    if (t instanceof java.lang.reflect.ParameterizedType) {
      java.lang.reflect.Type out = ((java.lang.reflect.ParameterizedType) t).getActualTypeArguments()[1];
      return out.getTypeName().startsWith("org.apache.pulsar.functions.api.Record");
    }
  }
  return false;
}

try {
  admin.sinks().createSink(config, pkgPath);
} catch (IllegalArgumentException e) {
  if (e.getMessage() != null && e.getMessage().contains("must be of type Record")) {
    log.error("Change transform function output type parameter to Record<T>", e);
  } else { throw e; }
}

When it happens

Trigger: 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).

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• Serializer type mismatch
• Serializer type mismatch ${typeArg} vs ${serDeTypeArg}
• SourceRecord class type must be PulsarRecord
• Sink does not implement correct interface
• PartitionId needs to be specified for every record while in

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/ca5ad2a497d418c5. Report an issue: GitHub.