apache/pulsar · error · IncompatibleSchemaException

To schema is not a KEY_VALUE schema.

Error message

To schema is not a KEY_VALUE schema.

What it means

KeyValueSchemaCompatibilityCheck only knows how to compare KEY_VALUE schemas; the incoming 'to' (new) schema must itself be a KEY_VALUE schema, otherwise the check cannot decompose it into key/value parts and throws immediately (unless strategy is ALWAYS_COMPATIBLE).

Source

Thrown at pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java:70

    @Override
    public SchemaType getSchemaType() {
        return SchemaType.KEY_VALUE;
    }

    @Override
    public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy)
            throws IncompatibleSchemaException {
        checkCompatible(Collections.singletonList(from), to, strategy);
    }

    @Override
    public void checkCompatible(Iterable<SchemaData> from, SchemaData to, SchemaCompatibilityStrategy strategy)
            throws IncompatibleSchemaException {
        if (strategy == SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE) {
            return;
        }
        if (to.getType() != SchemaType.KEY_VALUE) {
            throw new IncompatibleSchemaException("To schema is not a KEY_VALUE schema.");
        }
        LinkedList<SchemaData> fromKeyList = new LinkedList<>();
        LinkedList<SchemaData> fromValueList = new LinkedList<>();
        KeyValue<SchemaData, SchemaData> fromKeyValue;
        KeyValue<SchemaData, SchemaData> toKeyValue = decodeKeyValueSchemaData(to);
        SchemaType toKeyType = toKeyValue.getKey().getType();
        SchemaType toValueType = toKeyValue.getValue().getType();

        for (SchemaData schemaData : from) {
            if (schemaData.getType() != SchemaType.KEY_VALUE) {
                throw new IncompatibleSchemaException("From schema is not a KEY_VALUE schema.");
            }
            fromKeyValue = decodeKeyValueSchemaData(schemaData);
            if (fromKeyValue.getKey().getType() != toKeyType || fromKeyValue.getValue().getType() != toValueType) {
                throw new IncompatibleSchemaException(
                        String.format("Key schemas or Value schemas are different schema type, "
                                        + "from key schema type is %s and to key schema is %s,"
                                        + " from value schema is %s and to value schema is %s",

View on GitHub (pinned to 820761864e)

Solutions

1. Change the producer to use Schema.KeyValue(keySchema, valueSchema) matching the topic's schema type.
2. If the topic should no longer be KEY_VALUE, delete the schema (or the topic) and recreate with the desired schema type.
3. Check client configuration so all producers/consumers agree on the KeyValue schema.

Example fix

// before
Producer<MyVal> p = client.newProducer(Schema.AVRO(MyVal.class))...;
// after
Producer<KeyValue<MyKey, MyVal>> p = client.newProducer(Schema.KeyValue(Schema.AVRO(MyKey.class), Schema.AVRO(MyVal.class)))...;

SchemaInfo existing = admin.schemas().getSchemaInfo(topic);
if (existing.getType() == SchemaType.KEY_VALUE && schemaInfo.getType() != SchemaType.KEY_VALUE) {
    throw new IllegalArgumentException("Topic requires a KEY_VALUE schema");
}

boolean isKeyValueSchemaInfo(SchemaInfo info) {
    return info != null && info.getType() == SchemaType.KEY_VALUE;
}

try {
    admin.schemas().createSchema(topic, schemaInfo);
} catch (PulsarAdminException e) {
    if (e.getMessage().contains("To schema is not a KEY_VALUE schema")) {
        // wrap schema in Schema.KeyValue(...) or delete schema
    }
}

When it happens

Trigger: Updating a topic whose schema check routes to KeyValueSchemaCompatibilityCheck (existing schema is KEY_VALUE) with a new SchemaInfo whose type is AVRO/JSON/etc.

Common situations: A producer configured Schema.STRING or Schema.AVRO on a topic that was created with Schema.KeyValue(...); clients reconfigured without the KeyValue wrapper during refactors.

Related errors

• From schema is not a KEY_VALUE schema.
• Key schemas or Value schemas are different schema type, from
• RestException(conversionError)
• External schema is not compatible with the other schema type
• payload (${payloadSize} bytes) cannot be decoded with schema

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