What causes "Cannot decode a message without schema" in apache/pulsar?

Call fetchSchemaIfNeeded(schemaVersion) before decode for every new schema version

apache/pulsar · error · IllegalStateException

Cannot decode a message without schema

Error message

Cannot decode a message without schema

What it means

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.

Source

Thrown at pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java:301

        }
        for (Map.Entry<SchemaVersion, Schema<?>> entry : schemaMap.entrySet()) {
            schema.setSchema(entry.getKey(), entry.getValue());
        }
        return schema;
    }

    @Override
    public boolean requireFetchingSchemaInfo() {
        return true;
    }

    protected GenericRecord adapt(Object value, byte[] schemaVersion) {
        if (value instanceof GenericRecord) {
            return (GenericRecord) value;
        }
        SchemaVersion sv = getSchemaVersion(schemaVersion);
        if (!schemaMap.containsKey(sv)) {
            throw new IllegalStateException("Cannot decode a message without schema");
        }
        return wrapPrimitiveObject(value, schemaMap.get(sv).getSchemaInfo().getType(), schemaVersion);
    }

    public static GenericRecord wrapPrimitiveObject(Object value, SchemaType type, byte[] schemaVersion) {
        return GenericObjectWrapper.of(value, type, schemaVersion);
    }

    public Schema<?> getInternalSchema() {
        return schemaMap.get(SchemaVersion.Latest);
    }

    public Schema<?> getInternalSchema(byte[] schemaVersion) {
        return schemaMap.get(getSchemaVersion(schemaVersion));
    }

    /**
     * Get a specific schema version, fetching from the Registry if it is not loaded yet.

View on GitHub (pinned to 820761864e)

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

Example fix

// before
GenericRecord r = autoSchema.decode(payload, schemaVersionBytes); // may throw
// after
autoSchema.fetchSchemaIfNeeded(BytesSchemaVersion.of(schemaVersionBytes));
GenericRecord r = autoSchema.decode(payload, schemaVersionBytes);

Defensive patterns

Strategy: validation

Validate before calling

SchemaVersion sv = BytesSchemaVersion.of(schemaVersionBytes);
if (!autoSchema.schemaMap.containsKey(sv)) {
    autoSchema.fetchSchemaIfNeeded(sv);
}

Type guard

boolean hasSchemaForVersion(AutoConsumeSchema s, byte[] version) {
    return s.schemaMap.containsKey(BytesSchemaVersion.of(version == null ? new byte[0] : version));
}

Try / catch

try {
    GenericRecord r = schema.decode(payload, version);
} catch (IllegalStateException e) {
    schema.fetchSchemaIfNeeded(BytesSchemaVersion.of(version));
    GenericRecord r = schema.decode(payload, version);
}

Prevention

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

When it happens

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

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

Related errors

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