{"record":{"id":"77bc5b381e3f9db0","repo":"apache/pulsar","slug":"the-schema-is-not-a-keyvalueschema","errorCode":null,"errorMessage":"The schema is not a KeyValueSchema","messagePattern":"The schema is not a KeyValueSchema","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java","lineNumber":583,"sourceCode":"                    kvSchema.decode(getKeyBytes(), getData(), schemaVersion);\n            if (schema instanceof AutoConsumeSchema) {\n                return (T) AutoConsumeSchema.wrapPrimitiveObject(keyValue,\n                        ((AutoConsumeSchema) schema).getSchemaInfo(schemaVersion).getType(), schemaVersion);\n            } else {\n                return (T) keyValue;\n            }\n        } else {\n            return decode(schemaVersion);\n        }\n    }\n\n    @SuppressWarnings(\"unchecked\")\n    private T getKeyValueBySchemaId(byte[] schemaId) {\n        if (schema instanceof AutoConsumeSchema) {\n            throw new UnsupportedOperationException(\"AutoConsumeSchema is not supported with schemaId\");\n        }\n        if (!(schema instanceof KeyValueSchemaImpl<?, ?> kvSchema)) {\n            throw new IllegalStateException(\"The schema is not a KeyValueSchema\");\n        }\n        if (kvSchema.getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED) {\n            return (T) kvSchema.decode(topic, getKeyBytes(), getData(), schemaId);\n        } else {\n            return decodeBySchemaId(schemaId);\n        }\n    }\n\n    @SuppressWarnings(\"unchecked\")\n    private T getKeyValue() {\n        KeyValueSchemaImpl<?, ?> kvSchema = getKeyValueSchema();\n        if (kvSchema.getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED) {\n            org.apache.pulsar.common.schema.KeyValue<?, ?> keyValue =\n                    kvSchema.decode(getKeyBytes(), getData(), null);\n            if (schema instanceof AutoConsumeSchema) {\n                return (T) AutoConsumeSchema.wrapPrimitiveObject(keyValue,\n                        ((AutoConsumeSchema) schema).getSchemaInfo(getSchemaVersion()).getType(), null);\n            } else {","sourceCodeStart":565,"sourceCodeEnd":601,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L565-L601","documentation":"Thrown by getKeyValueBySchemaId when the message's schema is neither AutoConsumeSchema nor a KeyValueSchemaImpl, i.e. the code path that decodes key+value with an explicit schemaId only works on KeyValue schemas. It indicates the consumer is using a plain (non-KV) schema but invoked the KV-with-schemaId decode path. This is a programming/schema-mismatch error.","triggerScenarios":"Calling getKeyValue() on a message consumed with a non-KeyValue schema (e.g. Schema.AVRO, Schema.STRING) when the message carries a schemaId, routing into getKeyValueBySchemaId.","commonSituations":"Producer switched a topic to a KeyValue schema while the consumer still uses a plain value schema; mixing producers/consumers with different schema types on the same topic; misunderstanding that getKeyValue() requires Schema.KeyValue.","solutions":["Align the consumer's schema with the topic: use Schema.KeyValue(keySchema, valueSchema, encodingType) if the topic is KV-encoded","Check the topic's schema info (pulsar-admin schemas get) to confirm whether it is KEY_VALUE; adjust the consumer accordingly","If the topic should be value-only, fix the producer to stop writing KeyValue payloads","Catch IllegalStateException around getValue() and re-create the consumer with the correct schema"],"exampleFix":"// before\nConsumer<MyValue> c = client.newConsumer(Schema.AVRO(MyValue.class)).topic(\"t\")...;\nc.receive().getValue(); // IllegalStateException: The schema is not a KeyValueSchema\n// after\nConsumer<KeyValue<MyKey,MyValue>> c = client.newConsumer(\n    Schema.KeyValue(Schema.AVRO(MyKey.class), Schema.AVRO(MyValue.class), KeyValueEncodingType.INLINE))\n    .topic(\"t\")...;","handlingStrategy":"validation","validationCode":"SchemaInfo info = pulsarAdmin.schemas().getSchemaInfo(topic);\nboolean isKv = info.getSchemaDefinition().contains(\"key\") || info.getType() == SchemaType.KEY_VALUE;\n// choose Schema.KeyValue(...) when isKv, otherwise a plain value schema","typeGuard":"boolean isKeyValueSchema(Schema<?> s) {\n    return s instanceof KeyValueSchemaImpl<?, ?>;\n}","tryCatchPattern":"try {\n    T value = message.getValue();\n} catch (IllegalStateException e) {\n    // recreate consumer with Schema.KeyValue(keySchema, valueSchema, encodingType)\n}","preventionTips":["Confirm the topic's schema type before wiring the consumer schema","Coordinate schema changes between producer and consumer teams","Never mix plain and KeyValue schemas on one topic without a consumer migration plan"],"tags":["pulsar","schema","keyvalue","illegal-state"],"backgroundTag":"schema-type-mismatch","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"}