{"record":{"id":"58e69ed3c5ccfdf0","repo":"apache/pulsar","slug":"cannot-decode-a-message-without-schema","errorCode":null,"errorMessage":"Cannot decode a message without schema","messagePattern":"Cannot decode a message without schema","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java","lineNumber":301,"sourceCode":"        }\n        for (Map.Entry<SchemaVersion, Schema<?>> entry : schemaMap.entrySet()) {\n            schema.setSchema(entry.getKey(), entry.getValue());\n        }\n        return schema;\n    }\n\n    @Override\n    public boolean requireFetchingSchemaInfo() {\n        return true;\n    }\n\n    protected GenericRecord adapt(Object value, byte[] schemaVersion) {\n        if (value instanceof GenericRecord) {\n            return (GenericRecord) value;\n        }\n        SchemaVersion sv = getSchemaVersion(schemaVersion);\n        if (!schemaMap.containsKey(sv)) {\n            throw new IllegalStateException(\"Cannot decode a message without schema\");\n        }\n        return wrapPrimitiveObject(value, schemaMap.get(sv).getSchemaInfo().getType(), schemaVersion);\n    }\n\n    public static GenericRecord wrapPrimitiveObject(Object value, SchemaType type, byte[] schemaVersion) {\n        return GenericObjectWrapper.of(value, type, schemaVersion);\n    }\n\n    public Schema<?> getInternalSchema() {\n        return schemaMap.get(SchemaVersion.Latest);\n    }\n\n    public Schema<?> getInternalSchema(byte[] schemaVersion) {\n        return schemaMap.get(getSchemaVersion(schemaVersion));\n    }\n\n    /**\n     * Get a specific schema version, fetching from the Registry if it is not loaded yet.","sourceCodeStart":283,"sourceCodeEnd":319,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java#L283-L319","documentation":"AutoConsumeSchema.adapt() wraps decoded values into GenericRecords. If the raw value is not already a GenericRecord and the schema version's schema has not been fetched into schemaMap, it cannot know the value's type and throws IllegalStateException.","triggerScenarios":"Calling decode() (which calls adapt) with a schemaVersion for which fetchSchemaIfNeeded was never successfully run — e.g. decode called directly with a non-empty schema version bytes while schemaMap lacks that entry.","commonSituations":"Custom reader loops that call schema.decode(payload, schemaVersion) directly bypassing the normal message pipeline that pre-fetches schemas; schema fetch failed silently earlier so the map entry never got populated.","solutions":["Call fetchSchemaIfNeeded(schemaVersion) before decode for every new schema version","Let the consumer's normal message pipeline handle decoding instead of invoking decode() manually","Check for the IllegalStateException and lazily fetch the schema, then retry the decode"],"exampleFix":"// before\nGenericRecord r = autoSchema.decode(payload, schemaVersionBytes); // may throw\n// after\nautoSchema.fetchSchemaIfNeeded(BytesSchemaVersion.of(schemaVersionBytes));\nGenericRecord r = autoSchema.decode(payload, schemaVersionBytes);","handlingStrategy":"validation","validationCode":"SchemaVersion sv = BytesSchemaVersion.of(schemaVersionBytes);\nif (!autoSchema.schemaMap.containsKey(sv)) {\n    autoSchema.fetchSchemaIfNeeded(sv);\n}","typeGuard":"boolean hasSchemaForVersion(AutoConsumeSchema s, byte[] version) {\n    return s.schemaMap.containsKey(BytesSchemaVersion.of(version == null ? new byte[0] : version));\n}","tryCatchPattern":"try {\n    GenericRecord r = schema.decode(payload, version);\n} catch (IllegalStateException e) {\n    schema.fetchSchemaIfNeeded(BytesSchemaVersion.of(version));\n    GenericRecord r = schema.decode(payload, version);\n}","preventionTips":["Always run fetchSchemaIfNeeded before direct decode calls with explicit versions","Prefer the consumer's built-in message pipeline over calling decode() manually","Ensure schema fetch failures are surfaced rather than swallowed earlier in the pipeline"],"tags":["pulsar","schema","auto-consume","decode","illegal-state"],"backgroundTag":"missing-schema-for-decode","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}