apache/pulsar · error · RestException

RestException(conversionError)

Error message

RestException(conversionError)

What it means

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.

Source

Thrown at pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java:135

                .thenCompose(__ -> {
                    String schemaId = getSchemaId();
                    return pulsar().getSchemaRegistryService()
                            .deleteSchema(schemaId, defaultIfEmpty(clientAppId(), ""), force);
                });
    }

    public CompletableFuture<SchemaVersion> postSchemaAsync(PostSchemaPayload payload, boolean authoritative) {
        return validateOwnershipAndOperationAsync(authoritative, TopicOperation.PRODUCE)
                .thenCompose(__ -> getSchemaCompatibilityStrategyAsyncWithoutAuth())
                .thenCompose(schemaCompatibilityStrategy -> {
                    byte[] data;
                    if (SchemaType.KEY_VALUE.name().equals(payload.getType())) {
                        try {
                            data = DefaultImplementation.getDefaultImplementation()
                                    .convertKeyValueDataStringToSchemaInfoSchema(payload.getSchema()
                                            .getBytes(StandardCharsets.UTF_8));
                        } catch (IOException conversionError) {
                            throw new RestException(conversionError);
                        }
                    } else {
                        data = payload.getSchema().getBytes(StandardCharsets.UTF_8);
                    }
                    return pulsar().getSchemaRegistryService()
                            .putSchemaIfAbsent(getSchemaId(),
                                    SchemaData.builder().data(data).isDeleted(false).timestamp(clock.millis())
                                            .type(SchemaType.valueOf(payload.getType()))
                                            .user(defaultIfEmpty(clientAppId(), ""))
                                            .props(payload.getProperties())
                                            .build(),
                                    schemaCompatibilityStrategy);
                });
    }

    public CompletableFuture<Pair<Boolean, SchemaCompatibilityStrategy>> testCompatibilityAsync(
            PostSchemaPayload payload, boolean authoritative) {
        return validateDestinationAndAdminOperationAsync(authoritative)

View on GitHub (pinned to 820761864e)

Solutions

1. 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.
2. On the client, build the payload via KeyValueSchema encoding helpers (e.g. DefaultImplementation.convertKeyValueDataInfoToSchemaInfo inverse) rather than hand-writing the string.
3. 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.
4. Catch RestException on the putSchema call and inspect the cause to log the offending schema string before retrying.

Example fix

// before
admin.schemas().putSchema(topic, new SchemaDataImpl(
    "{\"key\":{...}}", SchemaType.KEY_VALUE, props)); // malformed hand-built string
// after
SchemaInfo kvInfo = DefaultImplementation.getDefaultImplementation()
    .convertKeyValueDataInfoToSchemaInfo(
        KeyValueSchemaInfo.encodeKeyValueSchemaInfo("kv", keySchema, valueSchema, KeyValueEncodingType.SEPARATED));
admin.schemas().putSchema(topic, kvInfo);

SchemaInfo info = admin.schemas().getSchemaInfo(topic); // client-side pre-check
if (payload.getType().equals("KEY_VALUE")) {
    try {
        DefaultImplementation.getDefaultImplementation()
            .convertKeyValueDataStringToSchemaInfo(payload.getSchema().getBytes(StandardCharsets.UTF_8));
    } catch (IOException e) {
        throw new IllegalArgumentException("Malformed KEY_VALUE schema data: " + e.getMessage());
    }
}

static boolean isWellFormedKeyValueSchemaString(String s) {
    try {
        DefaultImplementation.getDefaultImplementation()
            .convertKeyValueDataStringToSchemaInfo(s.getBytes(StandardCharsets.UTF_8));
        return true;
    } catch (IOException e) { return false; }
}

try {
    admin.schemas().putSchema(topic, schemaInfo);
} catch (PulsarAdminException e) {
    if (e.getCause() instanceof IOException) {
        log.error("KEY_VALUE schema data rejected: {}", e.getMessage());
    }
    throw new IllegalArgumentException("Invalid schema payload", e);
}

When it happens

Trigger: 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).

Common situations: 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.

Related errors

• Incompatible schema: exists schema type %s, new schema type
• cluster data is required
• Cannot delete non empty namespace
• Local cluster is not part of replicate cluster list
• Topic %s does not exist

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/c797522b462e1c0d. Report an issue: GitHub.