apache/pulsar · error · IllegalStateException

The schema is not a KeyValueSchema

Error message

The schema is not a KeyValueSchema

What it means

Thrown by getKeyValueBySchemaId when the message's schema is neither AutoConsumeSchema nor a KeyValueSchemaImpl, i.e. the code path that decodes key+value with an explicit schemaId only works on KeyValue schemas. It indicates the consumer is using a plain (non-KV) schema but invoked the KV-with-schemaId decode path. This is a programming/schema-mismatch error.

Source

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

                    kvSchema.decode(getKeyBytes(), getData(), schemaVersion);
            if (schema instanceof AutoConsumeSchema) {
                return (T) AutoConsumeSchema.wrapPrimitiveObject(keyValue,
                        ((AutoConsumeSchema) schema).getSchemaInfo(schemaVersion).getType(), schemaVersion);
            } else {
                return (T) keyValue;
            }
        } else {
            return decode(schemaVersion);
        }
    }

    @SuppressWarnings("unchecked")
    private T getKeyValueBySchemaId(byte[] schemaId) {
        if (schema instanceof AutoConsumeSchema) {
            throw new UnsupportedOperationException("AutoConsumeSchema is not supported with schemaId");
        }
        if (!(schema instanceof KeyValueSchemaImpl<?, ?> kvSchema)) {
            throw new IllegalStateException("The schema is not a KeyValueSchema");
        }
        if (kvSchema.getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED) {
            return (T) kvSchema.decode(topic, getKeyBytes(), getData(), schemaId);
        } else {
            return decodeBySchemaId(schemaId);
        }
    }

    @SuppressWarnings("unchecked")
    private T getKeyValue() {
        KeyValueSchemaImpl<?, ?> kvSchema = getKeyValueSchema();
        if (kvSchema.getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED) {
            org.apache.pulsar.common.schema.KeyValue<?, ?> keyValue =
                    kvSchema.decode(getKeyBytes(), getData(), null);
            if (schema instanceof AutoConsumeSchema) {
                return (T) AutoConsumeSchema.wrapPrimitiveObject(keyValue,
                        ((AutoConsumeSchema) schema).getSchemaInfo(getSchemaVersion()).getType(), null);
            } else {

View on GitHub (pinned to 820761864e)

Solutions

1. Align the consumer's schema with the topic: use Schema.KeyValue(keySchema, valueSchema, encodingType) if the topic is KV-encoded
2. Check the topic's schema info (pulsar-admin schemas get) to confirm whether it is KEY_VALUE; adjust the consumer accordingly
3. If the topic should be value-only, fix the producer to stop writing KeyValue payloads
4. Catch IllegalStateException around getValue() and re-create the consumer with the correct schema

Example fix

// before
Consumer<MyValue> c = client.newConsumer(Schema.AVRO(MyValue.class)).topic("t")...;
c.receive().getValue(); // IllegalStateException: The schema is not a KeyValueSchema
// after
Consumer<KeyValue<MyKey,MyValue>> c = client.newConsumer(
    Schema.KeyValue(Schema.AVRO(MyKey.class), Schema.AVRO(MyValue.class), KeyValueEncodingType.INLINE))
    .topic("t")...;

SchemaInfo info = pulsarAdmin.schemas().getSchemaInfo(topic);
boolean isKv = info.getSchemaDefinition().contains("key") || info.getType() == SchemaType.KEY_VALUE;
// choose Schema.KeyValue(...) when isKv, otherwise a plain value schema

boolean isKeyValueSchema(Schema<?> s) {
    return s instanceof KeyValueSchemaImpl<?, ?>;
}

try {
    T value = message.getValue();
} catch (IllegalStateException e) {
    // recreate consumer with Schema.KeyValue(keySchema, valueSchema, encodingType)
}

When it happens

Trigger: Calling getKeyValue() on a message consumed with a non-KeyValue schema (e.g. Schema.AVRO, Schema.STRING) when the message carries a schemaId, routing into getKeyValueBySchemaId.

Common situations: Producer switched a topic to a KeyValue schema while the consumer still uses a plain value schema; mixing producers/consumers with different schema types on the same topic; misunderstanding that getKeyValue() requires Schema.KeyValue.

Related errors

• AutoConsumeSchema is not supported with schemaId
• Cannot decode a message without 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

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