apache/pulsar · error · IllegalArgumentException

Only allowed to set schemaInfoForReplicator for a replicated

Error message

Only allowed to set schemaInfoForReplicator for a replicated message.

What it means

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.

Source

Thrown at pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java:474

            return null;
        }
        ensureSchemaIsLoaded();
        if (schema instanceof AutoConsumeSchema) {
            return ((AutoConsumeSchema) schema).getSchemaInfo(getSchemaVersion());
        }
        return schema.getSchemaInfo();
    }

    public SchemaHash getSchemaHash() {
        return schemaHash == null ? SchemaHash.empty() : schemaHash;
    }

    public void setSchemaInfoForReplicator(SchemaInfo schemaInfo) {
        if (msgMetadata.hasReplicatedFrom()) {
            this.schemaInfoForReplicator = schemaInfo;
            this.schemaHash = SchemaHash.of(schemaInfo);
        } else {
            throw new IllegalArgumentException("Only allowed to set schemaInfoForReplicator for a replicated message.");
        }
    }

    public SchemaInfo getSchemaInfoForReplicator() {
        return msgMetadata.hasReplicatedFrom() ? this.schemaInfoForReplicator : null;
    }

    @Override
    public T getValue() {
        SchemaInfo schemaInfo = getSchemaInfo();
        var schemaIdOp = getSchemaId();
        var schemaId = schemaIdOp.map(SchemaIdUtil::removeMagicHeader).orElse(null);
        if (schemaInfo != null && SchemaType.KEY_VALUE == schemaInfo.getType()) {
            if (schemaIdOp.isPresent()) {
                return getKeyValueBySchemaId(schemaId);
            }
            if (schema.supportSchemaVersioning()) {
                return getKeyValueBySchemaVersion();

View on GitHub (pinned to 820761864e)

Solutions

1. Only call setSchemaInfoForReplicator on messages obtained from replicated entries (messages with replicated-from metadata).
2. For producer-side schema, use the Producer/Schema configuration instead of this method.
3. In tests, build the message metadata with hasReplicatedFrom set before calling the setter.

Example fix

// before
MessageImpl<byte[]> msg = MessageImpl.create(...); // local message
msg.setSchemaInfoForReplicator(schemaInfo); // throws
// after
if (msg.getMessageMetadata().hasReplicatedFrom()) {
  msg.setSchemaInfoForReplicator(schemaInfo);
}

// guard before calling
if (!msg.getMessageMetadata().hasReplicatedFrom()) return; // skip setter

boolean isReplicatedMessage(Message<?> msg) {
  return msg instanceof MessageImpl<?> m && m.getMessageMetadata().hasReplicatedFrom();
}

try {
  msg.setSchemaInfoForReplicator(schemaInfo);
} catch (IllegalArgumentException e) {
  // not a replicated message: skip or route to local-producer schema path
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Can not enable for all producers but denies for replicators,
• Failed to add schema to an active topic with empty(BYTES) sc
• External schema is not compatible with the other schema type
• payload (${payloadSize} bytes) cannot be decoded with schema
• Failed to decode message from topic ${topic} with schemaId $

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