{"record":{"id":"c797522b462e1c0d","repo":"apache/pulsar","slug":"restexception-conversionerror","errorCode":null,"errorMessage":"RestException(conversionError)","messagePattern":"RestException\\(conversionError\\)","errorType":"exception","errorClass":"RestException","httpStatus":500,"severity":"error","filePath":"pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java","lineNumber":135,"sourceCode":"                .thenCompose(__ -> {\n                    String schemaId = getSchemaId();\n                    return pulsar().getSchemaRegistryService()\n                            .deleteSchema(schemaId, defaultIfEmpty(clientAppId(), \"\"), force);\n                });\n    }\n\n    public CompletableFuture<SchemaVersion> postSchemaAsync(PostSchemaPayload payload, boolean authoritative) {\n        return validateOwnershipAndOperationAsync(authoritative, TopicOperation.PRODUCE)\n                .thenCompose(__ -> getSchemaCompatibilityStrategyAsyncWithoutAuth())\n                .thenCompose(schemaCompatibilityStrategy -> {\n                    byte[] data;\n                    if (SchemaType.KEY_VALUE.name().equals(payload.getType())) {\n                        try {\n                            data = DefaultImplementation.getDefaultImplementation()\n                                    .convertKeyValueDataStringToSchemaInfoSchema(payload.getSchema()\n                                            .getBytes(StandardCharsets.UTF_8));\n                        } catch (IOException conversionError) {\n                            throw new RestException(conversionError);\n                        }\n                    } else {\n                        data = payload.getSchema().getBytes(StandardCharsets.UTF_8);\n                    }\n                    return pulsar().getSchemaRegistryService()\n                            .putSchemaIfAbsent(getSchemaId(),\n                                    SchemaData.builder().data(data).isDeleted(false).timestamp(clock.millis())\n                                            .type(SchemaType.valueOf(payload.getType()))\n                                            .user(defaultIfEmpty(clientAppId(), \"\"))\n                                            .props(payload.getProperties())\n                                            .build(),\n                                    schemaCompatibilityStrategy);\n                });\n    }\n\n    public CompletableFuture<Pair<Boolean, SchemaCompatibilityStrategy>> testCompatibilityAsync(\n            PostSchemaPayload payload, boolean authoritative) {\n        return validateDestinationAndAdminOperationAsync(authoritative)","sourceCodeStart":117,"sourceCodeEnd":153,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java#L117-L153","documentation":"In postSchemaAsync, when a schema upload declares type KEY_VALUE, the broker converts the uploaded schema data string into a KeyValues-backed SchemaInfo via Pulsar's DefaultImplementation. If that string is malformed JSON (not the expected {\"schema\":..., \"type\":\"KEY_VALUE\", \"schemaDataFormat\":...} structure) an IOException is thrown and wrapped as RestException(conversionError), producing a 4xx/5xx HTTP response to the admin PUT /schemas call.","triggerScenarios":"PUT to /admin/v2/schemas/{tenant}/{namespace}/{topic} (or the Java admin client schemas().putSchema(...)) with payload.type == \"KEY_VALUE\" and a payload.schema string that is not valid key-value schema JSON, or whose embedded info is missing required fields (type/schema props).","commonSituations":"Clients hand-crafting the JSON schema data instead of using SchemaInfo encoder; uploading a JSON/Avro schema string while declaring KEY_VALUE; cross-version clients emitting an older key-value encoding the current broker's DefaultImplementation can't parse; whitespace/truncated schema strings from config files or templates.","solutions":["Validate the KEY_VALUE schema string is the exact JSON structure Pulsar expects: keys 'schema' (the inner schema JSON string), 'type' ('JSON','AVRO', or 'PROTOBUF'), and 'schemaDataFormat' where applicable.","On the client, build the payload via KeyValueSchema encoding helpers (e.g. DefaultImplementation.convertKeyValueDataInfoToSchemaInfo inverse) rather than hand-writing the string.","Check broker and client Pulsar versions match — the key-value wire encoding changed across versions; upgrade the client or broker so both use the same encoding.","Catch RestException on the putSchema call and inspect the cause to log the offending schema string before retrying."],"exampleFix":"// before\nadmin.schemas().putSchema(topic, new SchemaDataImpl(\n    \"{\\\"key\\\":{...}}\", SchemaType.KEY_VALUE, props)); // malformed hand-built string\n// after\nSchemaInfo kvInfo = DefaultImplementation.getDefaultImplementation()\n    .convertKeyValueDataInfoToSchemaInfo(\n        KeyValueSchemaInfo.encodeKeyValueSchemaInfo(\"kv\", keySchema, valueSchema, KeyValueEncodingType.SEPARATED));\nadmin.schemas().putSchema(topic, kvInfo);","handlingStrategy":"validation","validationCode":"SchemaInfo info = admin.schemas().getSchemaInfo(topic); // client-side pre-check\nif (payload.getType().equals(\"KEY_VALUE\")) {\n    try {\n        DefaultImplementation.getDefaultImplementation()\n            .convertKeyValueDataStringToSchemaInfo(payload.getSchema().getBytes(StandardCharsets.UTF_8));\n    } catch (IOException e) {\n        throw new IllegalArgumentException(\"Malformed KEY_VALUE schema data: \" + e.getMessage());\n    }\n}","typeGuard":"static boolean isWellFormedKeyValueSchemaString(String s) {\n    try {\n        DefaultImplementation.getDefaultImplementation()\n            .convertKeyValueDataStringToSchemaInfo(s.getBytes(StandardCharsets.UTF_8));\n        return true;\n    } catch (IOException e) { return false; }\n}","tryCatchPattern":"try {\n    admin.schemas().putSchema(topic, schemaInfo);\n} catch (PulsarAdminException e) {\n    if (e.getCause() instanceof IOException) {\n        log.error(\"KEY_VALUE schema data rejected: {}\", e.getMessage());\n    }\n    throw new IllegalArgumentException(\"Invalid schema payload\", e);\n}","preventionTips":["Always build KEY_VALUE payloads with DefaultImplementation.convertKeyValueDataInfoToSchemaInfo, never hand-written JSON.","Round-trip test: convert your schema string back to SchemaInfo before uploading.","Pin broker and client to the same Pulsar version so key-value encodings match.","Log the schema string on failure for reproducible bug reports."],"tags":["pulsar","schema-registry","rest-api","json","key-value-schema"],"backgroundTag":"schema-conversion-failed","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}