@VisibleForTesting
    boolean populateMessageSchema(MessageImpl msg, SendCallback callback) {
        MessageMetadata msgMetadataBuilder = msg.getMessageBuilder();
        if (msg.getSchemaInternal() == schema) {
            schemaVersion.ifPresent(v -> msgMetadataBuilder.setSchemaVersion(v));
            msg.setSchemaState(MessageImpl.SchemaState.Ready);
            return true;
        }
        // If the message is from the replicator and without replicated schema
        // Which means the message is written with BYTES schema
        // So we don't need to replicate schema to the remote cluster
        if (msg.hasReplicateFrom() && msg.getSchemaInfoForReplicator() == null) {
            msg.setSchemaState(MessageImpl.SchemaState.Ready);
            return true;
        }

        if (!isMultiSchemaEnabled(true)) {
            PulsarClientException.InvalidMessageException e = new PulsarClientException.InvalidMessageException(
                    format("The producer %s of the topic %s is disabled the `MultiSchema`", producerName, topic)
                    , msg.getSequenceId());
            completeCallbackAndReleaseSemaphore(msg.getUncompressedSize(), callback, e);
            return false;
        }

        byte[] schemaVersion = schemaCache.get(msg.getSchemaHash());
        if (schemaVersion != null) {
            if (schemaVersion != SchemaVersion.Empty.bytes()) {
                msgMetadataBuilder.setSchemaVersion(schemaVersion);
            }

            msg.setSchemaState(MessageImpl.SchemaState.Ready);
        }

        return true;
    }
@Test
    public void testPopulateMessageSchema() {
        MessageImpl<?> msg = mock(MessageImpl.class);
        when(msg.hasReplicateFrom()).thenReturn(true);
        when(msg.getSchemaInternal()).thenReturn(mock(Schema.class));
        when(msg.getSchemaInfoForReplicator()).thenReturn(null);
        ProducerImpl<?> producer = mock(ProducerImpl.class, withSettings()
                .defaultAnswer(Mockito.CALLS_REAL_METHODS));
        assertTrue(producer.populateMessageSchema(msg, null));
        verify(msg).setSchemaState(MessageImpl.SchemaState.Ready);
    }