public class Policies {

    // 需要復制的集群信息
    public Set<String> replication_clusters = new HashSet<>();

  // ...

}

public void readEntriesComplete(List<Entry> entries, Object ctx) {
    if (log.isDebugEnabled()) {
        log.debug("[{}][{} -> {}] Read entries complete of {} messages", topicName, localCluster, remoteCluster,
                entries.size());
    }

    int maxReadBatchSize = topic.getBrokerService().pulsar().getConfiguration().getDispatcherMaxReadBatchSize();
    if (readBatchSize < maxReadBatchSize) {
        int newReadBatchSize = Math.min(readBatchSize * 2, maxReadBatchSize);
        if (log.isDebugEnabled()) {
            log.debug("[{}][{} -> {}] Increasing read batch size from {} to {}", topicName, localCluster,
                    remoteCluster, readBatchSize, newReadBatchSize);
        }

        readBatchSize = newReadBatchSize;
    }

    readFailureBackoff.reduceToHalf();

    boolean atLeastOneMessageSentForReplication = false;
    boolean isEnableReplicatedSubscriptions =
            brokerService.pulsar().getConfiguration().isEnableReplicatedSubscriptions();

    try {
        // This flag is set to true when we skip atleast one local message,
        // in order to skip remaining local messages.
        boolean isLocalMessageSkippedOnce = false;
        for (int i = 0; i < entries.size(); i++) {
            Entry entry = entries.get(i);
            int length = entry.getLength();
            ByteBuf headersAndPayload = entry.getDataBuffer();
            MessageImpl msg;
            try {
                msg = MessageImpl.deserializeSkipBrokerEntryMetaData(headersAndPayload);
            } catch (Throwable t) {
                log.error("[{}][{} -> {}] Failed to deserialize message at {} (buffer size: {}): {}", topicName,
                        localCluster, remoteCluster, entry.getPosition(), length, t.getMessage(), t);
                cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
                entry.release();
                continue;
            }

            if (isEnableReplicatedSubscriptions) {
                checkReplicatedSubscriptionMarker(entry.getPosition(), msg, headersAndPayload);
            }

            if (msg.isReplicated()) {
                // Discard messages that were already replicated into this region
                cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
                entry.release();
                msg.recycle();
                continue;
            }

            if (msg.hasReplicateTo() && !msg.getReplicateTo().contains(remoteCluster)) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}][{} -> {}] Skipping message at position {}, replicateTo {}", topicName,
                            localCluster, remoteCluster, entry.getPosition(), msg.getReplicateTo());
                }
                cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
                entry.release();
                msg.recycle();
                continue;
            }

            if (msg.isExpired(messageTTLInSeconds)) {
                msgExpired.recordEvent(0/* no value stat */);
                if (log.isDebugEnabled()) {
                    log.debug("[{}][{} -> {}] Discarding expired message at position {}, replicateTo {}", topicName,
                            localCluster, remoteCluster, entry.getPosition(), msg.getReplicateTo());
                }
                cursor.asyncDelete(entry.getPosition(), this, entry.getPosition());
                entry.release();
                msg.recycle();
                continue;
            }

            if (STATE_UPDATER.get(this) != State.Started || isLocalMessageSkippedOnce) {
                // The producer is not ready yet after having stopped/restarted. Drop the message because it will
                // recovered when the producer is ready
                if (log.isDebugEnabled()) {
                    log.debug("[{}][{} -> {}] Dropping read message at {} because producer is not ready", topicName,
                            localCluster, remoteCluster, entry.getPosition());
                }
                isLocalMessageSkippedOnce = true;
                entry.release();
                msg.recycle();
                continue;
            }

            dispatchRateLimiter.ifPresent(rateLimiter -> rateLimiter.tryDispatchPermit(1, entry.getLength()));

            // Increment pending messages for messages produced locally
            PENDING_MESSAGES_UPDATER.incrementAndGet(this);

            msgOut.recordEvent(headersAndPayload.readableBytes());

            msg.setReplicatedFrom(localCluster);

            headersAndPayload.retain();

            getSchemaInfo(msg).thenAccept(schemaInfo -> {
                msg.setSchemaInfoForReplicator(schemaInfo);
                producer.sendAsync(msg, ProducerSendCallback.create(this, entry, msg));
            }).exceptionally(ex -> {
                log.error("[{}][{} -> {}] Failed to get schema from local cluster", topicName,
                        localCluster, remoteCluster, ex);
                returnnull;
            });

            atLeastOneMessageSentForReplication = true;
        }
    } catch (Exception e) {
        log.error("[{}][{} -> {}] Unexpected exception: {}", topicName, localCluster, remoteCluster, e.getMessage(),
                e);
    }

    HAVE_PENDING_READ_UPDATER.set(this, FALSE);

    if (atLeastOneMessageSentForReplication && !isWritable()) {
        // Don't read any more entries until the current pending entries are persisted
        if (log.isDebugEnabled()) {
            log.debug("[{}][{} -> {}] Pausing replication traffic. at-least-one: {} is-writable: {}", topicName,
                    localCluster, remoteCluster, atLeastOneMessageSentForReplication, isWritable());
        }
    } else {
        readMoreEntries();
    }
}

} else if (availablePermits == -1) {
     // no permits from rate limit
     topic.getBrokerService().executor().schedule(
         () -> readMoreEntries(), MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);
 }