{"record":{"id":"52dabfee20675123","repo":"apache/pulsar","slug":"failed-to-decode-message-from-topic-topic-with","errorCode":null,"errorMessage":"Failed to decode message from topic ${topic} with schemaId ${schemaId}","messagePattern":"Failed to decode message from topic (.+?) with schemaId (.+?)","errorType":"exception","errorClass":"SchemaSerializationException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java","lineNumber":547,"sourceCode":"\n    private T decodeBySchema(byte[] schemaVersion) {\n        T value = poolMessage ? schema.decode(payload.nioBuffer(), schemaVersion) : null;\n        if (value != null) {\n            return value;\n        }\n\n        if (null == schemaVersion) {\n            return schema.decode(getByteBuffer());\n        } else {\n            return schema.decode(getByteBuffer(), schemaVersion);\n        }\n    }\n\n    private T decodeBySchemaId(byte[] schemaId) {\n        try {\n            return schema.decode(topic, getByteBuffer(), schemaId);\n        } catch (Exception e) {\n            throw new SchemaSerializationException(\"Failed to decode message from topic \" + topic\n                    + \" with schemaId \" + Base64.getEncoder().encodeToString(schemaId), e);\n        }\n    }\n\n    private ByteBuffer getByteBuffer() {\n        if (msgMetadata.isNullValue()) {\n            return null;\n        }\n        return this.payload.nioBuffer();\n    }\n\n    @SuppressWarnings(\"unchecked\")\n    private T getKeyValueBySchemaVersion() {\n        KeyValueSchemaImpl<?, ?> kvSchema = getKeyValueSchema();\n        byte[] schemaVersion = getSchemaVersion();\n        if (kvSchema.getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED) {\n            org.apache.pulsar.common.schema.KeyValue<?, ?> keyValue =\n                    kvSchema.decode(getKeyBytes(), getData(), schemaVersion);","sourceCodeStart":529,"sourceCodeEnd":565,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L529-L565","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["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","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","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","Catch SchemaSerializationException around getValue() and dead-letter or skip poison messages instead of crashing the consumer loop"],"exampleFix":"// before\nMessage<T> msg = consumer.receive();\nT value = msg.getValue(); // throws SchemaSerializationException on poison payload\n// after\nMessage<T> msg = consumer.receive();\nT value;\ntry {\n    value = msg.getValue();\n} catch (SchemaSerializationException e) {\n    log.warn(\"Skipping undecodable message {} on {}\", msg.getMessageId(), msg.getTopicName(), e);\n    consumer.acknowledge(msg); // or negativeAckreeived / dead-letter\n    return;\n}","handlingStrategy":"try-catch","validationCode":"// verify topic schema before consuming\nSchemaInfo info = pulsarAdmin.schemas().getSchemaInfo(topic); // throws if none\n// ensure consumer schema matches; for auto: consumer = client.newConsumer(Schema.AUTO_CONSUME()).topic(topic)...;","typeGuard":"boolean isDecodable(Message<T> msg) {\n    try { msg.getValue(); return true; } catch (SchemaSerializationException e) { return false; }\n}","tryCatchPattern":"try {\n    T value = message.getValue();\n} catch (SchemaSerializationException e) {\n    // log message id + topic, then ack/nack or dead-letter the poison message\n}","preventionTips":["Enable schema validation enforcement on the namespace so incompatible publishes are rejected at the producer","Use Schema.AUTO_CONSUME() when the topic's schema may evolve","Keep consumer and producer client versions aligned with the topic's schema versions","Wrap getValue() in the consume loop and dead-letter undecodable messages"],"tags":["pulsar","schema","deserialization","client"],"backgroundTag":"schema-deserialization-failed","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"}