{"record":{"id":"d6b6d737ebe7a9c1","repo":"apache/pulsar","slug":"only-allowed-to-set-schemainfoforreplicator-for-a","errorCode":null,"errorMessage":"Only allowed to set schemaInfoForReplicator for a replicated message.","messagePattern":"Only allowed to set schemaInfoForReplicator for a replicated message\\.","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java","lineNumber":474,"sourceCode":"            return null;\n        }\n        ensureSchemaIsLoaded();\n        if (schema instanceof AutoConsumeSchema) {\n            return ((AutoConsumeSchema) schema).getSchemaInfo(getSchemaVersion());\n        }\n        return schema.getSchemaInfo();\n    }\n\n    public SchemaHash getSchemaHash() {\n        return schemaHash == null ? SchemaHash.empty() : schemaHash;\n    }\n\n    public void setSchemaInfoForReplicator(SchemaInfo schemaInfo) {\n        if (msgMetadata.hasReplicatedFrom()) {\n            this.schemaInfoForReplicator = schemaInfo;\n            this.schemaHash = SchemaHash.of(schemaInfo);\n        } else {\n            throw new IllegalArgumentException(\"Only allowed to set schemaInfoForReplicator for a replicated message.\");\n        }\n    }\n\n    public SchemaInfo getSchemaInfoForReplicator() {\n        return msgMetadata.hasReplicatedFrom() ? this.schemaInfoForReplicator : null;\n    }\n\n    @Override\n    public T getValue() {\n        SchemaInfo schemaInfo = getSchemaInfo();\n        var schemaIdOp = getSchemaId();\n        var schemaId = schemaIdOp.map(SchemaIdUtil::removeMagicHeader).orElse(null);\n        if (schemaInfo != null && SchemaType.KEY_VALUE == schemaInfo.getType()) {\n            if (schemaIdOp.isPresent()) {\n                return getKeyValueBySchemaId(schemaId);\n            }\n            if (schema.supportSchemaVersioning()) {\n                return getKeyValueBySchemaVersion();","sourceCodeStart":456,"sourceCodeEnd":492,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L456-L492","documentation":"schemaInfoForReplicator may only be set on messages that were replicated from another cluster (msgMetadata.hasReplicatedFrom()). Setting it on a locally produced message is invalid because the schema cache on the remote cluster only needs schema info for replicated copies, so the method throws IllegalArgumentException.","triggerScenarios":"Calling message.setSchemaInfoForReplicator(schemaInfo) on a MessageImpl that does not carry the hasReplicatedFrom metadata — e.g. a locally published message or a raw message read from the origin topic rather than a replicated copy processed by replicateEntries.","commonSituations":"Custom replication logic invoking the setter outside the broker/repl-client path; unit tests constructing plain MessageImpl and calling the setter; confusion between setSchema (producer side) and this replicator-only API.","solutions":["Only call setSchemaInfoForReplicator on messages obtained from replicated entries (messages with replicated-from metadata).","For producer-side schema, use the Producer/Schema configuration instead of this method.","In tests, build the message metadata with hasReplicatedFrom set before calling the setter."],"exampleFix":"// before\nMessageImpl<byte[]> msg = MessageImpl.create(...); // local message\nmsg.setSchemaInfoForReplicator(schemaInfo); // throws\n// after\nif (msg.getMessageMetadata().hasReplicatedFrom()) {\n  msg.setSchemaInfoForReplicator(schemaInfo);\n}","handlingStrategy":"type-guard","validationCode":"// guard before calling\nif (!msg.getMessageMetadata().hasReplicatedFrom()) return; // skip setter","typeGuard":"boolean isReplicatedMessage(Message<?> msg) {\n  return msg instanceof MessageImpl<?> m && m.getMessageMetadata().hasReplicatedFrom();\n}","tryCatchPattern":"try {\n  msg.setSchemaInfoForReplicator(schemaInfo);\n} catch (IllegalArgumentException e) {\n  // not a replicated message: skip or route to local-producer schema path\n}","preventionTips":["Use setSchemaInfoForReplicator only inside replication pipelines (replicateEntries).","For producer-side schema use Producer builder Schema, not this API.","Check hasReplicatedFrom before invoking replicator-only methods."],"tags":["pulsar","replication","schema","message"],"backgroundTag":"schema-replication-misuse","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"}