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

When a schema is uploaded to a non-persistent topic that already has active consumers/entries but currently has an empty (BYTES) schema, compatibility is checked; if the topic has no real schema (NotExistSchemaException), the add fails with IncompatibleSchemaException because attaching a typed schema to an active empty-schema topic could break existing consumers.

Source

Thrown at pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java:1316

                new UnsupportedOperationException("getLastMessageId is not supported on non-persistent topic"));
    }
    @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
                    || (!producers.isEmpty())
                    || (numActiveConsumersWithoutAutoSchema != 0)
                    || ENTRIES_ADDED_COUNTER_UPDATER.get(this) != 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));
            }
        });
    }

    @Override
    public void publishTxnMessage(TxnID txnID, ByteBuf headersAndPayload, PublishContext publishContext) {
        throw new UnsupportedOperationException("PublishTxnMessage is not supported by non-persistent topic");
    }

    @Override
    public CompletableFuture<Void> endTxn(TxnID txnID, int txnAction, long lowWaterMark) {

View on GitHub (pinned to 820761864e)

Solutions

Upload the schema before any consumer connects or entries are produced on the topic
Use auto-schema (AutoConsume/AutoProduce) consumers so schema updates are negotiated automatically
Set schemaValidationEnforced appropriately and align producers/consumers on the same schema from the start
Recreate the topic (delete and recreate) and register the schema before traffic

Example fix

// before
// schema uploaded after producers/consumers active on schema-less topic
admin.schemas().putSchema(topic, schemaData);
// after
admin.topics().unloadTopic(topic); // drain active state
admin.schemas().putSchema(topic, schemaData); // then set schema before reconnecting clients

Defensive patterns

Strategy: validation

Validate before calling

// register schema before topic has active consumers/entries
SchemaInfo existing = admin.schemas().getSchemaInfo(topic); // may throw Not Found = empty
if (existing == null || existing.getType() == SchemaType.BYTES) {
    admin.schemas().putSchema(topic, schemaInfo); // before traffic
}

Try / catch

try {
    admin.schemas().putSchema(topic, schemaInfo);
} catch (PulsarAdminException e) {
    if (e.getCause() instanceof IncompatibleSchemaException) {
        log.error("Topic already active with empty schema; unload and set schema first");
    }
}

Prevention

Register schemas at topic creation time, before producers/consumers connect
Use auto-produce/auto-consume schemas for evolving payloads
Avoid mixing schema-less and typed usage on the same topic

When it happens

Trigger: Calling admin schemas().putSchema (or producer with auto-produce schema) on a non-persistent topic that has active consumers without auto-schema, or existing entries, while the topic's stored schema is empty/BYTES and the new schema is incompatible/absent.

Common situations: Producers starting to send Avro/JSON payloads to a non-persistent topic that was previously used schema-less; CI/tests uploading schemas to already-in-use non-persistent topics; schema compatibility checks failing after topic traffic started.

Related errors

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