apache/pulsar · error · UnsupportedOperationException

AutoConsumeSchema is not supported with schemaId

Error message

AutoConsumeSchema is not supported with schemaId

What it means

MessageImpl.getKeyValueBySchemaId explicitly rejects AutoConsumeSchema because auto-consumption has no way to split/decode a KeyValue payload against a schemaId. If you consume with AutoConsumeSchema and the message carries a schemaId (KeyValue schema path), calling getKeyValue()/getValue() throws UnsupportedOperationException. It is a deliberate 'not implemented' guard, not a data problem.

Source

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

        byte[] schemaVersion = getSchemaVersion();
        if (kvSchema.getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED) {
            org.apache.pulsar.common.schema.KeyValue<?, ?> keyValue =
                    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) {

View on GitHub (pinned to 820761864e)

Solutions

1. Use an explicit KeyValueSchema (Schema.KeyValue(KeySchema, ValueSchema, encoding)) instead of AUTO_CONSUME for this topic
2. Consume the value schema explicitly (e.g. Schema.AUTO_CONSUME on the value-only topic) rather than the KV topic
3. If you control the producer, stop attaching schemaIds or publish the key/value as separate fields so a plain schema can decode it
4. Catch UnsupportedOperationException and fall back to reading raw bytes via message.getData()

Example fix

// before
Consumer<KeyValue<MyKey,MyValue>> c = client.newConsumer(Schema.AUTO_CONSUME())
    .topic("kv-topic")...;
KeyValue<MyKey,MyValue> kv = c.receive().getValue(); // UnsupportedOperationException
// after
Consumer<KeyValue<MyKey,MyValue>> c = client.newConsumer(
    Schema.KeyValue(Schema.AVRO(MyKey.class), Schema.AVRO(MyValue.class), KeyValueEncodingType.SEPARATED))
    .topic("kv-topic")...;

// before consuming a KV topic, don't use AUTO_CONSUME
if (schema instanceof AutoConsumeSchema && topicSchemaType == SchemaType.KEY_VALUE) {
    throw new IllegalArgumentException("Use Schema.KeyValue(...) for KV topics, not AUTO_CONSUME");
}

boolean supportsKvDecode(Schema<?> schema) {
    return schema instanceof KeyValueSchemaImpl<?, ?>;
}

try {
    T value = message.getValue();
} catch (UnsupportedOperationException e) {
    // switch to explicit KeyValueSchema consumer or read raw bytes: message.getData()
}

When it happens

Trigger: Consumer created with Schema.AUTO_CONSUME() on a topic whose messages carry a schemaId, then calling getKeyValueBySchemaId (via getKeyValue()) on such a message.

Common situations: Using AUTO_CONSUME on a KeyValue-structured topic (e.g. Pulsar IO source/sink state topics or KV topics written by newer clients supporting schemaIds); upgrading clients where the KV-with-schemaId code path is newer than the consumer's expectation.

Related errors

• The schema is not a KeyValueSchema
• This schema is not meant to be used for encoding
• AutoConsumeSchema is not intended to be used for encoding
• Generic schema is not supported on schema type <schemaInfo.g
• Can not enable for all producers but denies for replicators,

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