{"record":{"id":"440b6c9af7fcc52a","repo":"apache/pulsar","slug":"to-schema-is-not-a-key-value-schema","errorCode":null,"errorMessage":"To schema is not a KEY_VALUE schema.","messagePattern":"To schema is not a KEY_VALUE schema\\.","errorType":"exception","errorClass":"IncompatibleSchemaException","httpStatus":null,"severity":"error","filePath":"pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java","lineNumber":70,"sourceCode":"    @Override\n    public SchemaType getSchemaType() {\n        return SchemaType.KEY_VALUE;\n    }\n\n    @Override\n    public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy)\n            throws IncompatibleSchemaException {\n        checkCompatible(Collections.singletonList(from), to, strategy);\n    }\n\n    @Override\n    public void checkCompatible(Iterable<SchemaData> from, SchemaData to, SchemaCompatibilityStrategy strategy)\n            throws IncompatibleSchemaException {\n        if (strategy == SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE) {\n            return;\n        }\n        if (to.getType() != SchemaType.KEY_VALUE) {\n            throw new IncompatibleSchemaException(\"To schema is not a KEY_VALUE schema.\");\n        }\n        LinkedList<SchemaData> fromKeyList = new LinkedList<>();\n        LinkedList<SchemaData> fromValueList = new LinkedList<>();\n        KeyValue<SchemaData, SchemaData> fromKeyValue;\n        KeyValue<SchemaData, SchemaData> toKeyValue = decodeKeyValueSchemaData(to);\n        SchemaType toKeyType = toKeyValue.getKey().getType();\n        SchemaType toValueType = toKeyValue.getValue().getType();\n\n        for (SchemaData schemaData : from) {\n            if (schemaData.getType() != SchemaType.KEY_VALUE) {\n                throw new IncompatibleSchemaException(\"From schema is not a KEY_VALUE schema.\");\n            }\n            fromKeyValue = decodeKeyValueSchemaData(schemaData);\n            if (fromKeyValue.getKey().getType() != toKeyType || fromKeyValue.getValue().getType() != toValueType) {\n                throw new IncompatibleSchemaException(\n                        String.format(\"Key schemas or Value schemas are different schema type, \"\n                                        + \"from key schema type is %s and to key schema is %s,\"\n                                        + \" from value schema is %s and to value schema is %s\",","sourceCodeStart":52,"sourceCodeEnd":88,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java#L52-L88","documentation":"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).","triggerScenarios":"Updating a topic whose schema check routes to KeyValueSchemaCompatibilityCheck (existing schema is KEY_VALUE) with a new SchemaInfo whose type is AVRO/JSON/etc.","commonSituations":"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.","solutions":["Change the producer to use Schema.KeyValue(keySchema, valueSchema) matching the topic's schema type.","If the topic should no longer be KEY_VALUE, delete the schema (or the topic) and recreate with the desired schema type.","Check client configuration so all producers/consumers agree on the KeyValue schema."],"exampleFix":"// before\nProducer<MyVal> p = client.newProducer(Schema.AVRO(MyVal.class))...;\n// after\nProducer<KeyValue<MyKey, MyVal>> p = client.newProducer(Schema.KeyValue(Schema.AVRO(MyKey.class), Schema.AVRO(MyVal.class)))...;","handlingStrategy":"validation","validationCode":"SchemaInfo existing = admin.schemas().getSchemaInfo(topic);\nif (existing.getType() == SchemaType.KEY_VALUE && schemaInfo.getType() != SchemaType.KEY_VALUE) {\n    throw new IllegalArgumentException(\"Topic requires a KEY_VALUE schema\");\n}","typeGuard":"boolean isKeyValueSchemaInfo(SchemaInfo info) {\n    return info != null && info.getType() == SchemaType.KEY_VALUE;\n}","tryCatchPattern":"try {\n    admin.schemas().createSchema(topic, schemaInfo);\n} catch (PulsarAdminException e) {\n    if (e.getMessage().contains(\"To schema is not a KEY_VALUE schema\")) {\n        // wrap schema in Schema.KeyValue(...) or delete schema\n    }\n}","preventionTips":["Check topic schema type via admin.schemas() before configuring producers.","Centralize KeyValue schema definitions in a shared module.","Delete and recreate the schema when intentionally changing schema family.","Add integration tests asserting schema type compatibility."],"tags":["pulsar","key-value-schema","compatibility"],"backgroundTag":"incompatible-schema-type","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}