Pulsar geo-replication

下图说明了 Pulsar 在不同集群之间跨地域复制的过程：

In this diagram, whenever P1, P2, and P3 producers publish messages to the T1 topic on Cluster-A, Cluster-B, and Cluster-C clusters respectively, those messages are instantly replicated across clusters. Once the messages are replicated, C1 and C2 consumers can consume those messages from their respective clusters.

Without geo-replication, C1 and C2 consumers are not able to consume messages that P3 producer publishes.

You must enable geo-replication on a per-tenant basis in Pulsar. You can enable geo-replication between clusters only when a tenant is created that allows access to both clusters.

尽管必须在两个集群间启用地理复制，但跨域复制是在命名空间级别管理的。 You must complete the following tasks to enable geo-replication for a namespace:

配置该名称空间使其可以跨两个或多个集群复制

Any message published on any topic in that namespace is replicated to all clusters in the specified set.

当消息发送到 Pulsar 的主题中，消息首先在本地集群中持久化，然后再被异步转发到远程集群。

在正常情况下，当连接没有问题时，消息会被立即复制，并同时分发给本地的消费者。通常情况下，网络round-trip time(RTT) 定义了两个远程地域之间端到端的传输延时。

应用可以在任何集群中创建生产者和消费者，即使无法访问远程集群(比如在网络分区内)。

生产者和消费者能够在 Pulsar 实例中的任意集群生产消息和消费消息。然而，订阅不仅是创建订阅的集群的本地订阅，而且还可以在启用复制订阅后在集群之间传输。一旦启用订阅复制，你能够保持同步的订阅状态。因此，主题能够异步的跨多个地域进行复制。如果发生故障，消费者能够在其他的集群从这个故障点重新消费消息。

In the aforementioned example, the T1 topic is replicated among three clusters, Cluster-A, Cluster-B, and Cluster-C.

这三个集群中的任何一个集群生成的所有消息都交付给其他集群中的所有订阅。 In this case, C1 and C2 consumers receive all messages that P1, P2, and P3 producers publish. 指令仍然在每个生产者的basis上有所保障。

The following example connects three clusters: us-east, us-west, and us-cent.

To replicate data among clusters, you need to configure each cluster to connect to the other. You can use the tool to create a connection.

示例

Suppose that you have 3 replication clusters: us-west, us-cent, and us-east.

Configure the connection from us-west to us-east.

Run the following command on us-west.

Configure the connection from us-west to us-cent.

Run the following command on us-west.

$ bin/pulsar-admin clusters create \
  --broker-url pulsar://<DNS-OF-US-CENT>:<PORT> \
  --url http://<DNS-OF-US-CENT>:<PORT> \

Run similar commands on us-east and us-cent to create connections among clusters.

授予属性权限

To replicate to a cluster, the tenant needs permission to use that cluster. You can grant permission to the tenant when you create the tenant or grant later.

创建租户的时候，指定所有预期的集群。

要更新现有租户的权限，请使用update而不是create。

你可以使用如下示例命令创建名称空间。

$ bin/pulsar-admin namespaces create my-tenant/my-namespace

你能够随时改变命名空间的复制集群，而不会中断正在进行的通信。一旦配置发生更改，复制通道将立即在所有集群中设置或停止。

主题跨域复制

一旦你在命名空间级别创建了跨域复制，任何生产者或者消费者在这个命名空间创建的主题都会被复制到所有的集群中。通常情况下，每个应用应该使用自己本地集群的serviceUrl。

选择性复制

By default, messages are replicated to all clusters configured for the namespace. 您可以通过为消息指定需要复制的目标集群列表，来选择性地复制数据到目标集群。然后该消息将只复制到指定的目标集群。

如下是示例。当你构建Message 对象时，可以使用setReplicationClusters 方法来指定目标集群:

List<String> restrictReplicationTo = Arrays.asList(
        "us-west",
        "us-east"
);
Producer producer = client.newProducer()
producer.newMessage()
        .value("my-payload".getBytes())
        .setReplicationClusters(restrictReplicationTo)
        .send();

Topic 统计数据

特定主题的地跨域复制统计信息可以通过 pulse -admin 工具和 API 查看

每个集群会生成自己的本地统计信息报告，包括传入和传出的复制率和队列容量。

删除主题跨域复制

Given that geo-replication topics exist in multiple regions, directly deleting a geo-replication topic is not possible. Instead, you should rely on automatic topic garbage collection.

In Pulsar, a topic is automatically deleted when the topic meets the following three conditions:

no producers or consumers are connected to it;
no subscriptions to it;
no more messages are kept for retention. For geo-replication topics, each region uses a fault-tolerant mechanism to decide when deleting the topic locally is safe.

可以通过在中设置brokerDeleteInactiveTopicsEnabled为false来显式禁用主题垃圾收集。

要删除跨域复制主题，请关闭该主题上的所有生产者和消费者，并删除每个复制集群中的所有本地订阅。 When Pulsar determines that no valid subscription for the topic remains across the system, it will garbage collect the topic.

Pulsar 支持复制订阅关系，所以能够在不到1秒的时间内，在不同集群间保持订阅状态的同步。主题的上下文信息也能在跨多个物理地域间进行异步复制。

如果发生故障，消费者重启后能够在其他的集群从这个消费失败的点开始消费。

Replicated subscription is disabled by default. You can enable replicated subscription when creating a consumer.

Consumer<String> consumer = client.newConsumer(Schema.STRING)
            .topic("my-topic")
            .subscriptionName("my-subscription")
            .replicateSubscriptionState(true)
            .subscribe();

优点

逻辑很容易实现。
可以选择启用或禁用复制订阅。
当启用的时候，它的间接成本很低，也很容易配置。

Geo-replication