PIP-356: Support Geo-Replication starts at earliest position

Background knowledge

Replication reads messages from the source cluster, and copies them to the remote cluster.

  • Registers a cursor named pulsar.repl.{remote-cluster} on the source cluster. Replicator reads messages relies on this cursor.
  • Registers a producer on the remote cluster. Replicator writes messages relies on this producer.

Motivation

If you have some older messages to migrate, the steps recommended are below, which was described at pulsar doc.

  1. Create the cursor that the replicator will use manually: pulsar-admin topics create-subscription -s pulsar.repl.{remote-cluster} -m earliest <topic>.
  2. Enable namespace-level/topic-level Geo-Replication.

The steps recommended are difficultly to use, for example:

  • Create cursor pulsar.repl.{remote-cluster} manually.
  • The namespace/topic was unloaded due to a re-balance.
    • The broker will remove the pulsar.repl.{remote-cluster} automatically because the Geo-Replication feature is disabled at this moment.
  • Enable namespace-level/topic-level Geo-Replication, but the cursor that was created manually has been deleted, the broker will create a new one with latest position, which is not expected.

Goals

Add an optional config(broker level, namespace level, and topic level) to support Geo-Replication starting at the earliest position.

Configuration

broker.conf

# The position that replication task start at, it can be set to "earliest" or "latest (default)".
replicationStartAt=latest

ServiceConfiguration

@FieldContext(
    category = CATEGORY_REPLICATION,
    dynamic = true,
    doc = "The position that replication task start at, it can be set to earliest or latest (default)."
)
String replicationStartAt = "latest";

Public API

V2/Namespaces.java

@POST
@Path("/{tenant}/{namespace}/replicationStartAt")
public void setNamespaceLevelReplicationStartAt(
    @PathParam("tenant") String tenant,
    @PathParam("namespace") String namespace,
    @QueryParam("replicationStartAt") String replicationStartAt) {
    ...
    ...
}

V2/PersistentTopics.java

@POST
@Path("/{tenant}/{namespace}/{topic}/replicationStartAt")
public void setNamespaceLevelReplicationStartAt(
    @PathParam("tenant") String tenant,
    @PathParam("namespace") String namespace,
    @PathParam("topic") @Encoded String encodedTopic,
    @QueryParam("isGlobal") @DefaultValue("false") boolean isGlobal,
    @QueryParam("replicationStartAt") String replicationStartAt) {
    ...
    ...
}

CLI

Namespaces command

pulsar-admin namespaces set-replication-start-at {earliest|latest} <namespace>

Topics command

pulsar-admin topics set-replication-start-at {earliest|latest} <topic>

Binary protocol

Nothing.

Metrics

Nothing.

Monitoring

Nothing.

Security Considerations

Nothing.

Backward & Forward Compatibility

You can do upgrading or reverting normally, no specified steps are needed to do.

Alternatives

Nothing.

General Notes

Links