{"record":{"id":"944c7ddbac7dde78","repo":"apache/pulsar","slug":"key-schemas-or-value-schemas-are-different-schema","errorCode":null,"errorMessage":"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","messagePattern":"Key schemas or Value schemas are different schema type, from key schema type is (.+?) and to key schema is (.+?), from value schema is (.+?) and to value schema is (.+?)","errorType":"exception","errorClass":"IncompatibleSchemaException","httpStatus":null,"severity":"error","filePath":"pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java","lineNumber":85,"sourceCode":"            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);\n        valueCheck.checkCompatible(fromValueList, toKeyValue.getValue(), strategy);\n    }\n}\n","sourceCodeStart":67,"sourceCodeEnd":103,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/KeyValueSchemaCompatibilityCheck.java#L67-L103","documentation":"Once both sides are KEY_VALUE schemas, their inner key and value schema types must match exactly (e.g. key AVRO + value AVRO on both sides). If either the key schema type or the value schema type differs between the stored schema and the new schema, the update is rejected with a formatted message listing all four types.","triggerScenarios":"checkCompatible where fromKeyValue.getKey().getType() != toKeyType or fromKeyValue.getValue().getType() != toValueType after decoding both KeyValue schema payloads.","commonSituations":"Producers switching from Schema.KeyValue(AVRO, AVRO) to Schema.KeyValue(STRING, AVRO) or KeyValue(JSON, JSON) on the same topic; mixed client configurations across services.","solutions":["Align the key and value schema types with the existing topic schema (inspect via admin.schemas() to see current types).","If the type change is intentional, delete the topic schema and re-create it with the new KeyValue configuration.","Standardize the KeyValue schema definition in a shared library used by all producers/consumers."],"exampleFix":"// before\nSchema.KeyValue(Schema.STRING, Schema.AVRO(Value.class)) // topic was KV(AVRO, AVRO)\n// after\nSchema.KeyValue(Schema.AVRO(Key.class), Schema.AVRO(Value.class))","handlingStrategy":"validation","validationCode":"SchemaInfo existing = admin.schemas().getSchemaInfo(topic);\n// decode existing KV schema types and compare to the planned one before upload\nKeyValue<SchemaType, SchemaType> planned = KeyValue.of(SchemaType.AVRO, SchemaType.AVRO);\n// existing types come from decoding existing.getSchemaData(); abort if they differ from planned","typeGuard":"boolean kvTypesMatch(SchemaType fromKey, SchemaType toKey, SchemaType fromVal, SchemaType toVal) {\n    return fromKey == toKey && fromVal == toVal;\n}","tryCatchPattern":"try {\n    admin.schemas().createSchema(topic, kvSchemaInfo);\n} catch (PulsarAdminException e) {\n    if (e.getMessage().startsWith(\"Key schemas or Value schemas are different schema type\")) {\n        // align key/value schema types with the existing topic schema\n    }\n}","preventionTips":["Record the topic's KV (key type, value type) pair in your service config and validate against it.","Use a shared schema registry module for KeyValue schema definitions.","Avoid ad-hoc Schema.KeyValue combinations in producer code.","Check the existing schema with the admin API before any change."],"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-14T11:17:12.474Z"}