apache/pulsar · error · RestException

e.getMessage()

Error message

e.getMessage()

What it means

During a function trigger (test message) the worker serializes your input into the input topic's schema using an AUTO_PRODUCE_BYTES producer. If the payload cannot be encoded with that schema, a SchemaSerializationException is wrapped into a 400 RestException with the schema error appended. An IOException on the underlying producer/reader operations is instead converted to a 500 RestException carrying the raw exception message.

Source

Thrown at pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/ComponentImpl.java:1225

                    MessageId newMsgId = MessageId.fromByteArray(
                            Base64.getDecoder().decode((String) msg.getProperties().get("__pfn_input_msg_id__")));

                    if (msgId.equals(newMsgId)
                            && msg.getProperties().get("__pfn_input_topic__")
                            .equals(TopicName.get(inputTopicToWrite).toString())) {
                        return new String(msg.getData());
                    }
                }
                curTime = System.currentTimeMillis();
            }
            throw new RestException(Status.REQUEST_TIMEOUT, "Request Timed Out");
        } catch (SchemaSerializationException e) {
            throw new RestException(Status.BAD_REQUEST, String.format(
                    "Failed to serialize input with error: %s. Please check"
                            + "if input data conforms with the schema of the input topic.",
                    e.getMessage()));
        } catch (IOException e) {
            throw new RestException(Status.INTERNAL_SERVER_ERROR, e.getMessage());
        } finally {
            if (reader != null) {
                reader.closeAsync();
            }
            if (producer != null) {
                producer.closeAsync();
            }
        }
    }

    @Override
    public FunctionState getFunctionState(final String tenant,
                                          final String namespace,
                                          final String functionName,
                                          final String key,
                                          final AuthenticationParameters authParams) {

        if (!isWorkerServiceAvailable()) {

View on GitHub (pinned to 820761864e)

Solutions

1. Match the trigger payload to the input topic's declared schema (e.g. send valid Avro/JSON-encoded bytes for an AvroSchema topic).
2. Inspect the appended e.getMessage() to see which field or encoding the schema rejected.
3. For 500/IOException variants, check worker-to-broker connectivity and topic existence, then retry the trigger.
4. Verify the function's input schema config (or use AUTO_CONSUME / correct schema type) if the schema was set incorrectly at registration.

Example fix

// before: trigger with raw JSON string against an Avro-schema topic
curl -X PUT .../trigger --data '{"id":"not-a-long"}'

// after: send schema-conformant payload (or base64 Avro-encoded bytes)
curl -X PUT .../trigger --data '{"id":42}'

// Validate payload against the topic schema before triggering (pseudocode using Pulsar client)
Schema<T> schema = ...; // same schema as the input topic
try {
    byte[] encoded = schema.encode(payload); // throws SchemaSerializationException if incompatible
} catch (SchemaSerializationException e) {
    // fix payload before calling the trigger endpoint
}

try {
    Response r = triggerFunction(tenant, ns, fn, payload);
} catch (RestApiException e) {
    if (e.getResponse().getStatus() == 400 && e.getMessage().contains("Failed to serialize input")) {
        // adjust payload to schema
    } else if (e.getResponse().getStatus() == 500) {
        // IOException: check broker connectivity, retry
    }
}

When it happens

Trigger: Calling PUT /admin/v3/functions/{tenant}/{namespace}/{functionName}/trigger with input bytes that do not conform to the declared schema of the function's input topic (at line 1225 specifically: any IOException from producer.send/reader operations in triggerFunction).

Common situations: Sending JSON text to a topic with an Avro/Protobuf schema; sending a string to an int64-schema topic; schema evolved on the topic but the client payload was not updated; broker/bookkeeper connectivity problems producing IOExceptions during send.

Related errors

• This method is not supported
• Not implemented for ${this.getClass}
• Unknown version ${BytesSchemaVersion.of(schemaVersion)}
• Can't get accurate schema information for topic ${topicName}
• Size of data received by BooleanSchema is not 1

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