{"record":{"id":"50f575d7d70e511a","repo":"apache/pulsar","slug":"failed-to-add-schema-to-an-active-topic-with-empty-50f575","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/persistent/PersistentTopic.java","lineNumber":5011,"sourceCode":"        }\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                    || (userCreatedProducerCount > 0)\n                    || (numActiveConsumersWithoutAutoSchema != 0)\n                    || (ledger.getTotalSize() != 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 ->\n                        CompletableFuture.completedFuture(null));\n            }\n        });\n    }\n\n    public synchronized void checkReplicatedSubscriptionControllerState() {\n        AtomicBoolean shouldBeEnabled = new AtomicBoolean(false);\n        subscriptions.forEach((name, subscription) -> {\n            if (subscription.isReplicated()) {\n                shouldBeEnabled.set(true);\n            }","sourceCodeStart":4993,"sourceCodeEnd":5029,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L4993-L5029","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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"],"exampleFix":"// before\nProducer<byte[]> p = client.newProducer().topic(\"t\").create(); // topic gets BYTES schema\n// later: admin.schemas().upload(\"t\", avroSchema) -> IncompatibleSchemaException\n// after\nProducer<MyAvro> p = client.newProducer(Schema.AVRO(MyAvro.class)).topic(\"t-new\").create();\n// or enable auto-update schema policy on the namespace before first write:\nadmin.namespaces().setSchemaAutoUpdateCompatibilityPolicy(namespace, AutoUpdateCompatibilityPolicy.BACKWARD);","handlingStrategy":"validation","validationCode":"try {\n    SchemaInfo existing = admin.schemas().getSchemaInfo(topic);\n} catch (PulsarAdminException.NotFoundException e) {\n    // topic has no schema yet: publish with the typed schema from the very first producer\n    // never write to it with Schema.BYTES if a typed schema will be added later\n}","typeGuard":"boolean topicHasTypedSchema(String topic) {\n    try {\n        SchemaInfo si = admin.schemas().getSchemaInfo(topic);\n        return si != null && si.getSchemaType() != SchemaType.BYTES && si.getSchemaType() != SchemaType.NONE;\n    } catch (PulsarAdminException.NotFoundException e) {\n        return false;\n    }\n}","tryCatchPattern":"try {\n    admin.schemas().uploadSchema(topic, schemaInfo);\n} catch (PulsarAdminException e) {\n    if (e.getCause() instanceof IncompatibleSchemaException\n            && e.getMessage().contains(\"empty(BYTES) schema\")) {\n        // recreate topic or migrate to a new topic with the typed schema\n    } else throw e;\n}","preventionTips":["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"],"tags":["pulsar","broker","schema","incompatible-schema","topic"],"backgroundTag":"schema-incompatible-active-topic","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"}