{"record":{"id":"5cff7069689265be","repo":"apache/pulsar","slug":"autoconsumeschema-is-not-supported-with-schemaid","errorCode":null,"errorMessage":"AutoConsumeSchema is not supported with schemaId","messagePattern":"AutoConsumeSchema is not supported with schemaId","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java","lineNumber":580,"sourceCode":"        byte[] schemaVersion = getSchemaVersion();\n        if (kvSchema.getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED) {\n            org.apache.pulsar.common.schema.KeyValue<?, ?> keyValue =\n                    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) {","sourceCodeStart":562,"sourceCodeEnd":598,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L562-L598","documentation":"MessageImpl.getKeyValueBySchemaId explicitly rejects AutoConsumeSchema because auto-consumption has no way to split/decode a KeyValue payload against a schemaId. If you consume with AutoConsumeSchema and the message carries a schemaId (KeyValue schema path), calling getKeyValue()/getValue() throws UnsupportedOperationException. It is a deliberate 'not implemented' guard, not a data problem.","triggerScenarios":"Consumer created with Schema.AUTO_CONSUME() on a topic whose messages carry a schemaId, then calling getKeyValueBySchemaId (via getKeyValue()) on such a message.","commonSituations":"Using AUTO_CONSUME on a KeyValue-structured topic (e.g. Pulsar IO source/sink state topics or KV topics written by newer clients supporting schemaIds); upgrading clients where the KV-with-schemaId code path is newer than the consumer's expectation.","solutions":["Use an explicit KeyValueSchema (Schema.KeyValue(KeySchema, ValueSchema, encoding)) instead of AUTO_CONSUME for this topic","Consume the value schema explicitly (e.g. Schema.AUTO_CONSUME on the value-only topic) rather than the KV topic","If you control the producer, stop attaching schemaIds or publish the key/value as separate fields so a plain schema can decode it","Catch UnsupportedOperationException and fall back to reading raw bytes via message.getData()"],"exampleFix":"// before\nConsumer<KeyValue<MyKey,MyValue>> c = client.newConsumer(Schema.AUTO_CONSUME())\n    .topic(\"kv-topic\")...;\nKeyValue<MyKey,MyValue> kv = c.receive().getValue(); // UnsupportedOperationException\n// after\nConsumer<KeyValue<MyKey,MyValue>> c = client.newConsumer(\n    Schema.KeyValue(Schema.AVRO(MyKey.class), Schema.AVRO(MyValue.class), KeyValueEncodingType.SEPARATED))\n    .topic(\"kv-topic\")...;","handlingStrategy":"validation","validationCode":"// before consuming a KV topic, don't use AUTO_CONSUME\nif (schema instanceof AutoConsumeSchema && topicSchemaType == SchemaType.KEY_VALUE) {\n    throw new IllegalArgumentException(\"Use Schema.KeyValue(...) for KV topics, not AUTO_CONSUME\");\n}","typeGuard":"boolean supportsKvDecode(Schema<?> schema) {\n    return schema instanceof KeyValueSchemaImpl<?, ?>;\n}","tryCatchPattern":"try {\n    T value = message.getValue();\n} catch (UnsupportedOperationException e) {\n    // switch to explicit KeyValueSchema consumer or read raw bytes: message.getData()\n}","preventionTips":["Match schema type to the topic: KeyValue schema for KV topics, AUTO_CONSUME only for plain topics","Check the topic's schema type with pulsar-admin schemas get before choosing the consumer schema","Read the producer's schema setup when consuming topics owned by other teams"],"tags":["pulsar","schema","keyvalue","unsupported-operation"],"backgroundTag":"schema-not-supported","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"}