apache/pulsar · error · org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException

Failed to add schema to an active topic with empty(BYTES) sc

Error message

Failed to add schema to an active topic with empty(BYTES) schema: new schema type ${schemaType}

What it means

A topic that was created without a schema has an empty (BYTES) schema. If data was already written/consumers attached, the broker cannot transparently switch it to a typed schema, so when adding a new schema and no compatible schema is found (NotExistSchemaException) it throws IncompatibleSchemaException wrapping 'Failed to add schema to an active topic with empty(BYTES) schema'. This prevents silently changing the interpretation of existing data on an active topic.

Source

Thrown at pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java:5011

        }
    }
    @Override
    public CompletableFuture<Void> addSchemaIfIdleOrCheckCompatible(SchemaData schema) {
        return hasSchema().thenCompose((hasSchema) -> {
            int numActiveConsumersWithoutAutoSchema = subscriptions.values().stream()
                    .mapToInt(subscription -> subscription.getConsumers().stream()
                            .filter(consumer -> consumer.getSchemaType() != SchemaType.AUTO_CONSUME)
                            .toList().size())
                    .sum();
            if (hasSchema
                    || (userCreatedProducerCount > 0)
                    || (numActiveConsumersWithoutAutoSchema != 0)
                    || (ledger.getTotalSize() != 0)) {
                return checkSchemaCompatibleForConsumer(schema)
                        .exceptionally(ex -> {
                            Throwable realCause = FutureUtil.unwrapCompletionException(ex);
                            if (realCause instanceof NotExistSchemaException) {
                                throw FutureUtil.wrapToCompletionException(
                                        new IncompatibleSchemaException("Failed to add schema to an active topic"
                                                + " with empty(BYTES) schema: new schema type " + schema.getType()));
                            }
                            throw FutureUtil.wrapToCompletionException(realCause);
                        });
            } else {
                return addSchema(schema).thenCompose(schemaVersion ->
                        CompletableFuture.completedFuture(null));
            }
        });
    }

    public synchronized void checkReplicatedSubscriptionControllerState() {
        AtomicBoolean shouldBeEnabled = new AtomicBoolean(false);
        subscriptions.forEach((name, subscription) -> {
            if (subscription.isReplicated()) {
                shouldBeEnabled.set(true);
            }

View on GitHub (pinned to 820761864e)

Solutions

Delete the topic (or its data) so it starts fresh, then connect with the typed schema before any BYTES writes/consumers
Configure producers/consumers with AutoConsume or auto-update schema policy (schemaAutoUpdateCompatibility / isAllowAutoUpdateSchema) so the schema can be added safely
Migrate to a new topic with the desired schema instead of retrofitting the active BYTES topic
If data is empty and no strict consumers, ensure conditions allow the fast path (no active non-auto consumers and empty ledger) then re-add the schema

Example fix

// before
Producer<byte[]> p = client.newProducer().topic("t").create(); // topic gets BYTES schema
// later: admin.schemas().upload("t", avroSchema) -> IncompatibleSchemaException
// after
Producer<MyAvro> p = client.newProducer(Schema.AVRO(MyAvro.class)).topic("t-new").create();
// or enable auto-update schema policy on the namespace before first write:
admin.namespaces().setSchemaAutoUpdateCompatibilityPolicy(namespace, AutoUpdateCompatibilityPolicy.BACKWARD);

Defensive patterns

Strategy: validation

Validate before calling

try {
    SchemaInfo existing = admin.schemas().getSchemaInfo(topic);
} catch (PulsarAdminException.NotFoundException e) {
    // topic has no schema yet: publish with the typed schema from the very first producer
    // never write to it with Schema.BYTES if a typed schema will be added later
}

Type guard

boolean topicHasTypedSchema(String topic) {
    try {
        SchemaInfo si = admin.schemas().getSchemaInfo(topic);
        return si != null && si.getSchemaType() != SchemaType.BYTES && si.getSchemaType() != SchemaType.NONE;
    } catch (PulsarAdminException.NotFoundException e) {
        return false;
    }
}

Try / catch

try {
    admin.schemas().uploadSchema(topic, schemaInfo);
} catch (PulsarAdminException e) {
    if (e.getCause() instanceof IncompatibleSchemaException
            && e.getMessage().contains("empty(BYTES) schema")) {
        // recreate topic or migrate to a new topic with the typed schema
    } else throw e;
}

Prevention

Never mix Schema.BYTES producers with future typed-schema plans on the same topic
Create the topic with the intended typed schema before the first write
Enable namespace schema auto-update with a compatible policy
Check existing schema with admin.schemas().getSchemaInfo before switching producer schema types

When it happens

Trigger: Calling admin schemas().upload/addSchema (or a producer with AutoConsume/typed schema) on a topic that currently has a null/BYTES schema while it is active — i.e. it has active consumers without AutoSchema, active non-auto-schema consumers, or ledger total size != 0 — and checkSchemaCompatibleForConsumer fails with NotExistSchemaException.

Common situations: Producer first connects with Schema.BYTES (or no schema) writing data, then the app is upgraded to use a typed schema (Avro/JSON) on the same topic; schema auto-update blocked because consumers are attached; reusing an existing non-schema topic for new typed producers.

Related errors

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