{"record":{"id":"554e8e16046238b4","repo":"apache/pulsar","slug":"can-t-get-accurate-schema-information-for-topicn","errorCode":null,"errorMessage":"Can't get accurate schema information for ${topicName} using KeyValueSchemaImpl because SchemaInfoProvider is not set yet","messagePattern":"Can't get accurate schema information for (.+?) using KeyValueSchemaImpl because SchemaInfoProvider is not set yet","errorType":"exception","errorClass":"SchemaSerializationException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaImpl.java","lineNumber":446,"sourceCode":"     * We cannot call this method in getSchemaInfo.\n     * @see AutoConsumeSchema#fetchSchemaIfNeeded(SchemaVersion)\n     */\n    public void fetchSchemaIfNeeded(String topicName, SchemaVersion schemaVersion) throws SchemaSerializationException {\n        if (schemaInfo != null) {\n            if (keySchema instanceof AutoConsumeSchema) {\n                ((AutoConsumeSchema) keySchema).fetchSchemaIfNeeded(schemaVersion);\n            }\n            if (valueSchema instanceof AutoConsumeSchema) {\n                ((AutoConsumeSchema) valueSchema).fetchSchemaIfNeeded(schemaVersion);\n            }\n            return;\n        }\n        setSchemaInfoProviderOnSubschemas();\n        if (schemaVersion == null) {\n            schemaVersion = BytesSchemaVersion.of(new byte[0]);\n        }\n        if (schemaInfoProvider == null) {\n            throw new SchemaSerializationException(\"Can't get accurate schema information for \" + topicName + \" \"\n                    + \"using KeyValueSchemaImpl because SchemaInfoProvider is not set yet\");\n        } else {\n            SchemaInfo schemaInfo;\n            try {\n                schemaInfo = schemaInfoProvider.getSchemaByVersion(schemaVersion.bytes()).get();\n                if (schemaInfo == null) {\n                    // schemaless topic\n                    schemaInfo = BytesSchema.of().getSchemaInfo();\n                }\n                configureSchemaInfo(topicName, \"topic\", schemaInfo);\n            } catch (InterruptedException | ExecutionException e) {\n                if (e instanceof InterruptedException) {\n                    Thread.currentThread().interrupt();\n                }\n                log.error().attr(\"topic\", topicName).log(\"Can't get last schema for topic using KeyValueSchemaImpl\");\n                throw new SchemaSerializationException(e.getCause());\n            }\n            log.info().attr(\"schema\", schemaVersion)","sourceCodeStart":428,"sourceCodeEnd":464,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaImpl.java#L428-L464","documentation":"KeyValueSchemaImpl.fetchSchemaIfNeeded needs to look up the schema for the message's schema version, but no SchemaInfoProvider has been attached to this schema (usually set by the consumer/reader when connecting). Without a provider the client cannot resolve the versioned schema, so it throws SchemaSerializationException.","triggerScenarios":"Calling decode/fetchSchemaIfNeeded on a standalone KeyValueSchemaImpl (not obtained through a consumer that installed a schemaInfoProvider), or decoding before the consumer finished attaching the provider.","commonSituations":"Using Schema.KeyValue(...) directly in unit tests or tooling without a ConsumerBuilder schema context; decoding messages offline; auto-schema (Latest/Versioned) KeyValue topics read via a manually constructed schema instance.","solutions":["Obtain/decode the schema through a ConsumerBuilder/ReaderBuilder with schema configured, so the client sets the SchemaInfoProvider automatically","Set the schema info provider explicitly (schema.setSchemaInfoProvider(...)) if you construct the schema manually","If you only have inline schema info, configure it before decoding so no provider lookup is needed"],"exampleFix":"// before\nKeyValueSchemaImpl kv = (KeyValueSchemaImpl) Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.INLINE);\nkv.decode(topic, data, schemaVersion); // no provider -> throws\n// after\nPulsarClient.newClient().newConsumer(Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.INLINE)).topic(topic)...subscribe(); // consumer installs the provider","handlingStrategy":"try-catch","validationCode":"if (schema instanceof KeyValueSchemaImpl && ((KeyValueSchemaImpl<?,?>) schema).getSchemaInfoProvider() == null) { throw new IllegalStateException(\"KeyValue schema has no SchemaInfoProvider; obtain it from a consumer/reader\"); }","typeGuard":null,"tryCatchPattern":"try { kv = kvSchema.decode(topic, data, schemaVersion); } catch (SchemaSerializationException e) { if (e.getMessage().contains(\"SchemaInfoProvider is not set\")) { kv = decodeViaConsumer(message); } else throw e; }","preventionTips":["Always decode KeyValue schemas through a Pulsar consumer/reader","In tests, install a mock SchemaInfoProvider before decoding","Avoid caching schema instances across clients; the provider is client-scoped"],"tags":["schema","keyvalue","schema-provider"],"backgroundTag":"schema-info-provider-not-set","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"}