apache/pulsar · error · SchemaSerializationException

Failed to decode message from topic ${topic} with schemaId $

Error message

Failed to decode message from topic ${topic} with schemaId ${schemaId}

What it means

MessageImpl.decodeBySchemaId wraps any exception thrown by schema.decode(topic, buffer, schemaId) in a SchemaSerializationException. It means the message payload could not be decoded with the schema identified by the given schemaId — either the bytes don't match the schema, the schema version is unknown to the client, or the payload is corrupt. This is thrown on the consumer side when calling getValue()/getKeyValue() on a message carrying an explicit schemaId.

Source

Thrown at pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java:547

    private T decodeBySchema(byte[] schemaVersion) {
        T value = poolMessage ? schema.decode(payload.nioBuffer(), schemaVersion) : null;
        if (value != null) {
            return value;
        }

        if (null == schemaVersion) {
            return schema.decode(getByteBuffer());
        } else {
            return schema.decode(getByteBuffer(), schemaVersion);
        }
    }

    private T decodeBySchemaId(byte[] schemaId) {
        try {
            return schema.decode(topic, getByteBuffer(), schemaId);
        } catch (Exception e) {
            throw new SchemaSerializationException("Failed to decode message from topic " + topic
                    + " with schemaId " + Base64.getEncoder().encodeToString(schemaId), e);
        }
    }

    private ByteBuffer getByteBuffer() {
        if (msgMetadata.isNullValue()) {
            return null;
        }
        return this.payload.nioBuffer();
    }

    @SuppressWarnings("unchecked")
    private T getKeyValueBySchemaVersion() {
        KeyValueSchemaImpl<?, ?> kvSchema = getKeyValueSchema();
        byte[] schemaVersion = getSchemaVersion();
        if (kvSchema.getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED) {
            org.apache.pulsar.common.schema.KeyValue<?, ?> keyValue =
                    kvSchema.decode(getKeyBytes(), getData(), schemaVersion);

View on GitHub (pinned to 820761864e)

Solutions

1. Check that the topic's current schema version matches what your consumer expects (pulsar-admin schemas get) and that the schemaId in the error decodes to a schema the client can fetch
2. Ensure all producers on the topic write with a schema compatible with the consumer's schema; enable schema validation enforcement (isValidationEnforced) on the namespace to reject incompatible writes at publish time
3. If the schema evolved, upgrade the consumer's schema/reader or use AutoConsumeSchema so the client fetches the schema version the message was written with
4. Catch SchemaSerializationException around getValue() and dead-letter or skip poison messages instead of crashing the consumer loop

Example fix

// before
Message<T> msg = consumer.receive();
T value = msg.getValue(); // throws SchemaSerializationException on poison payload
// after
Message<T> msg = consumer.receive();
T value;
try {
    value = msg.getValue();
} catch (SchemaSerializationException e) {
    log.warn("Skipping undecodable message {} on {}", msg.getMessageId(), msg.getTopicName(), e);
    consumer.acknowledge(msg); // or negativeAckreeived / dead-letter
    return;
}

// verify topic schema before consuming
SchemaInfo info = pulsarAdmin.schemas().getSchemaInfo(topic); // throws if none
// ensure consumer schema matches; for auto: consumer = client.newConsumer(Schema.AUTO_CONSUME()).topic(topic)...;

boolean isDecodable(Message<T> msg) {
    try { msg.getValue(); return true; } catch (SchemaSerializationException e) { return false; }
}

try {
    T value = message.getValue();
} catch (SchemaSerializationException e) {
    // log message id + topic, then ack/nack or dead-letter the poison message
}

When it happens

Trigger: Calling message.getValue() (or getKeyValue()) on a message whose schemaId cannot be resolved or whose payload fails to decode: producer wrote bytes that don't conform to the topic schema, the schema was deleted/changed and the client's cached SchemaReader can't parse the payload, or schema validation is disabled allowing incompatible writes.

Common situations: Schema evolution mistakes (changing a topic's schema to an incompatible version while old messages are still in the backlog); producers writing with schema validation off; consumers with a stale or mismatched schema; corruption after topic compaction or manual data migration.

Understand the failure class

• Parsing and encoding errors: unexpected token, malformed input — why parsers reject input and how to find the real culprit.

Related errors

• payload (${payloadSize} bytes) cannot be decoded with schema
• Can not enable for all producers but denies for replicators,
• Failed to add schema to an active topic with empty(BYTES) sc
• External schema is not compatible with the other schema type
• Try to reserve/release memory failed, the param memorySize i

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