Pulsar geo-replication

Pulsar geo-replication

Geo-replication is the replication of persistently stored message data across multiple clusters of a Pulsar instance.

跨域复制是怎样工作的

下图说明了 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.

跨地域复制和 Pulsar 属性

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:

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

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上有所保障。

配置复制

正如在Geo-replication和脉冲星属性一节中所述，脉冲星中的Geo-replication是在租户级别进行管理的。

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

Connect replication clusters

To replicate data among clusters, you need to configure each cluster to connect to the other. You can use the pulsar-admin 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.

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

Tip
If you want to use a secure connection for a cluster, you can use the flags --broker-url-secure and --url-secure. For more information, see pulsar-admin clusters create.

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> \
  us-cent

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.

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

$ bin/pulsar-admin tenants create my-tenant \
  --admin-roles my-admin-role \
  --allowed-clusters us-west,us-east,us-cent

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

激活geo-replication名称空间

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

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

Initially, the namespace is not assigned to any cluster. You can assign the namespace to clusters using the set-clusters subcommand:

$ bin/pulsar-admin namespaces set-clusters my-tenant/my-namespace \
  --clusters us-west,us-east,us-cent

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

主题跨域复制

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

选择性复制

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

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

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

Topic 统计数据

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

$ bin/pulsar-admin persistent stats persistent://my-tenant/my-namespace/my-topic

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

删除主题跨域复制

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();

优点

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

缺点

当您启用复制订阅时，您正在创建一致的分布式快照，以建立来自不同集群的消息id之间的关联。快照是定期备份的，默认是1秒。这意味着因为故障而转移到其他集群的使用者可能会收到1秒的重复数据。你也可以在broker.conf文件中配置快照的频率。