{"record":{"id":"3f2cadf2751e10a2","repo":"apache/pulsar","slug":"from-schema-is-not-a-key-value-schema","errorCode":null,"errorMessage":"From schema is not a KEY_VALUE schema.","messagePattern":"From 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":81,"sourceCode":"    @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\",\n                                fromKeyValue.getKey().getType(),\n                                toKeyType,\n                                fromKeyValue.getValue().getType(),\n                                toValueType));\n            }\n            fromKeyList.addFirst(fromKeyValue.getKey());\n            fromValueList.addFirst(fromKeyValue.getValue());\n        }\n        SchemaCompatibilityCheck keyCheck = checkers.getOrDefault(toKeyType, SchemaCompatibilityCheck.DEFAULT);\n        SchemaCompatibilityCheck valueCheck = checkers.getOrDefault(toValueType, SchemaCompatibilityCheck.DEFAULT);\n        keyCheck.checkCompatible(fromKeyList, toKeyValue.getKey(), strategy);","sourceCodeStart":63,"sourceCodeEnd":99,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java#L63-L99","documentation":"KeyValueSchemaCompatibilityCheck requires the previous ('from') schema to also be of KEY_VALUE type; a non-KEY_VALUE previous schema cannot be compared against the new KEY_VALUE schema, so compatibility checking aborts with this error.","triggerScenarios":"checkCompatible(Iterable<SchemaData>, SchemaData, strategy) iterating the existing schema versions where one stored entry has type != KEY_VALUE while the new 'to' schema is KEY_VALUE.","commonSituations":"Schema history polluted by an earlier non-KV upload (e.g. someone temporarily uploaded a plain AVRO schema before KV enforcement); migration scripts that mixed schema types on one topic.","solutions":["Provide a KEY_VALUE schema on the producer side","Check that schema type negotiation did not degrade to a primitive type"],"exampleFix":"// before: mixed history\nadmin.schemas().createSchema(topic, avroSchemaInfo); // creates non-KV entry\nadmin.schemas().createSchema(topic, kvSchemaInfo);   // now fails\n// after: consistent KV history\ndeleteSchema(topic); createSchema(topic, kvSchemaInfo);","handlingStrategy":"validation","validationCode":"List<VersionedSchema> history = admin.schemas().getAllSchemas(topic);\nboolean allKv = history.stream().allMatch(v -> v.getSchema().getType() == SchemaType.KEY_VALUE);\nif (!allKv) throw new IllegalStateException(\"Schema history contains non-KV entries; clean history first\");","typeGuard":"boolean historyAllKeyValue(List<VersionedSchema> history) {\n    return history.stream().allMatch(v -> v.getSchema().getType() == SchemaType.KEY_VALUE);\n}","tryCatchPattern":"try {\n    admin.schemas().createSchema(topic, kvSchemaInfo);\n} catch (PulsarAdminException e) {\n    if (e.getMessage().contains(\"From schema is not a KEY_VALUE schema\")) {\n        // clean schema history: delete schema/topic and recreate with KV only\n    }\n}","preventionTips":["Never upload non-KV schemas to a KV-schema topic, even temporarily.","Keep one owner/team responsible for a topic's schema lifecycle.","Recreate topics when migrating schema families instead of overlaying.","Script schema setup so environments don't drift."],"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"}