{"record":{"id":"139384b2bc01aeae","repo":"apache/pulsar","slug":"failed-to-add-schema-to-an-active-topic-with-empty","errorCode":null,"errorMessage":"Failed to add schema to an active topic with empty(BYTES) schema: new schema type ${schemaType}","messagePattern":"Failed to add schema to an active topic with empty\\(BYTES\\) schema: new schema type (.+?)","errorType":"exception","errorClass":"org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException","httpStatus":null,"severity":"error","filePath":"pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java","lineNumber":1316,"sourceCode":"                new UnsupportedOperationException(\"getLastMessageId is not supported on non-persistent topic\"));\n    }\n    @Override\n    public CompletableFuture<Void> addSchemaIfIdleOrCheckCompatible(SchemaData schema) {\n        return hasSchema().thenCompose((hasSchema) -> {\n            int numActiveConsumersWithoutAutoSchema = subscriptions.values().stream()\n                    .mapToInt(subscription -> subscription.getConsumers().stream()\n                            .filter(consumer -> consumer.getSchemaType() != SchemaType.AUTO_CONSUME)\n                            .toList().size())\n                    .sum();\n            if (hasSchema\n                    || (!producers.isEmpty())\n                    || (numActiveConsumersWithoutAutoSchema != 0)\n                    || ENTRIES_ADDED_COUNTER_UPDATER.get(this) != 0) {\n                return checkSchemaCompatibleForConsumer(schema)\n                        .exceptionally(ex -> {\n                            Throwable realCause = FutureUtil.unwrapCompletionException(ex);\n                            if (realCause instanceof NotExistSchemaException) {\n                                throw FutureUtil.wrapToCompletionException(\n                                        new IncompatibleSchemaException(\"Failed to add schema to an active topic\"\n                                                + \" with empty(BYTES) schema: new schema type \" + schema.getType()));\n                            }\n                            throw FutureUtil.wrapToCompletionException(realCause);\n                        });\n            } else {\n                return addSchema(schema).thenCompose(schemaVersion -> CompletableFuture.completedFuture(null));\n            }\n        });\n    }\n\n    @Override\n    public void publishTxnMessage(TxnID txnID, ByteBuf headersAndPayload, PublishContext publishContext) {\n        throw new UnsupportedOperationException(\"PublishTxnMessage is not supported by non-persistent topic\");\n    }\n\n    @Override\n    public CompletableFuture<Void> endTxn(TxnID txnID, int txnAction, long lowWaterMark) {","sourceCodeStart":1298,"sourceCodeEnd":1334,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java#L1298-L1334","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\n// schema uploaded after producers/consumers active on schema-less topic\nadmin.schemas().putSchema(topic, schemaData);\n// after\nadmin.topics().unloadTopic(topic); // drain active state\nadmin.schemas().putSchema(topic, schemaData); // then set schema before reconnecting clients","handlingStrategy":"validation","validationCode":"// register schema before topic has active consumers/entries\nSchemaInfo existing = admin.schemas().getSchemaInfo(topic); // may throw Not Found = empty\nif (existing == null || existing.getType() == SchemaType.BYTES) {\n    admin.schemas().putSchema(topic, schemaInfo); // before traffic\n}","typeGuard":null,"tryCatchPattern":"try {\n    admin.schemas().putSchema(topic, schemaInfo);\n} catch (PulsarAdminException e) {\n    if (e.getCause() instanceof IncompatibleSchemaException) {\n        log.error(\"Topic already active with empty schema; unload and set schema first\");\n    }\n}","preventionTips":["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"],"tags":["java","broker","schema","non-persistent","schema-compatibility"],"backgroundTag":"schema-incompatible","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"}