[fix][broker] Resolve replicator remote cluster by prefix so cluster names containing a dot work

[SEPURI-SAI-KRISHNA]

Fixes #26450

Motivation

Cluster names are allowed to contain . — NamedEntity.NAMED_ENTITY_PATTERN is ^[-=:.\w]*$,
and the comment above it states this applies to "property, namespace, cluster and topic names".

AbstractReplicator builds replicator cursor/subscription names as
<replicatorPrefix>.<remoteCluster> in getReplicatorName, but recovered the cluster by splitting
on . and taking the last segment:

public static String getRemoteCluster(String remoteCursor) {
    String[] split = remoteCursor.split("\\.");
    return split[split.length - 1];
}

The two are therefore not inverses. For a cluster named remote.east, the cursor
pulsar.repl.remote.east resolves to east. Two consequences:

Every admin operation addressed at a replicator subscription fails. Six call sites parse the
name, then look the replicator up under the wrong cluster and get null — clear-backlog, skip
messages, expire messages by position, expire messages by timestamp, getReplicatorReference
(peek and replicator stats), and namespace-wide clear-backlog. The first five return
404 Replicator not found for a subscription that topics stats plainly lists; the sixth targets
the wrong subscription. Each already guards with subName.startsWith(replicatorPrefix) on the
line immediately above, so the subscription is recognised as a replicator's — only the cluster
it belongs to is derived wrongly.

The orphan-cursor sweep misidentifies live replicators.
PersistentTopic#removeOrphanReplicationCursors() runs inside initialize(), i.e. on every topic
load. getRemoteCluster("pulsar.repl.remote.east") returned east, which is not among the topic's
replication clusters, so a live, correctly-configured replicator was declared orphaned and
removeReplicator("east") was called. That rebuilds the name as pulsar.repl.east and calls
asyncDeleteCursor on it; no such cursor exists, so the callback fails with
CursorNotFoundException and the initialize() chain fails. The cursor survived only because the
reconstructed name was wrong too — the sweep fully intended to delete a cursor that was not
orphaned. #22890 documents that wrongly removing a replicator cursor loses the entire replication
backlog, so this sits on a code path already known to be destructive when it misfires.

Modifications

• AbstractReplicator#getRemoteCluster now takes the replicator prefix and strips it, making it
  the exact inverse of getReplicatorName(replicatorPrefix, cluster). A name that does not carry
  the prefix is returned unchanged, so callers fail their replicator lookup exactly as before.
• Updated the seven call sites in PersistentTopic, PersistentTopicsBase and NamespacesBase.
  All of them already had the prefix in scope from the startsWith guard, so no plumbing was
  needed beyond passing it in.

Note on the signature: this replaces the single-argument public static getRemoteCluster(String)
rather than adding an overload. Keeping the old one as a deprecated delegate is not possible
without the prefix — the missing prefix is the bug — and leaving a knowingly-wrong method in
place seemed worse than removing it. Happy to reconsider if a deprecated overload is preferred for
broker-plugin compatibility.

Fixing the parsing was chosen over rejecting dotted cluster names: deployments may already use
them, and turning those into a validation error would break working clusters on upgrade.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• AbstractReplicatorTest#testGetRemoteClusterRoundTripsClusterNamesContainingDots — asserts
  getRemoteCluster(getReplicatorName(prefix, cluster)) == cluster for us-west,
  us-east.prod, a.b.c, cluster:1 and r3.
• AbstractReplicatorTest#testGetRemoteClusterLeavesNonReplicatorNamesUnchanged — a name without
  the prefix is returned untouched.
• PersistentTopicTest#testReplicatorCursorOfClusterWithDotInNameIsNotTreatedAsOrphan — creates a
  topic with a live replicator cursor for cluster remote.east, loads it, and asserts the orphan
  sweep logs no removal warning and the cursor survives. This test fails on unpatched code with
  the live replicator of cluster remote.east was treated as an orphan: [Remove the orphan replicator because the cluster does not exist].

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

The REST endpoints and admin CLI options are unchanged in shape; the affected operations simply
stop returning 404 for replicator subscriptions of dotted cluster names.

Documentation

• doc-required
• doc-not-needed
• doc
• doc-complete

Bug fix; no documented behaviour changes.

Matching PR in forked repository

PR in forked repository: N/A (branch built and tested locally; broker tests listed above pass)

Assisted-by: Claude Code

[lhotari]

Thanks for contributing!

The prefix-based extraction fixes dotted cluster names, and all three new regression tests pass with retries disabled. Please make the helper return Optional<String> so callers can use one operation for both detecting a replicator name and extracting its remote cluster. Also adapt the existing code to use this new API.

[lhotari]

[QUALITY] Return Optional<String> to unify detection and extraction

Could this helper return Optional<String>? It would make the exact-prefix check, including the separator, authoritative here and let callers avoid repeating a separate startsWith check:

public static Optional<String> getRemoteCluster(String replicatorPrefix, String replicatorCursorName) {
    String prefix = replicatorPrefix + ".";
    return replicatorCursorName.startsWith(prefix)
            ? Optional.of(replicatorCursorName.substring(prefix.length()))
            : Optional.empty();
}

For example, the caller in PersistentTopicsBase.java:4212-4223 could use the same result for selection and for its not-found message:

final MessageExpirer messageExpirer;
var remoteCluster = AbstractReplicator.getRemoteCluster(topic.getReplicatorPrefix(), subName);
if (remoteCluster.isPresent()) {
    messageExpirer = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster.get());
} else {
    messageExpirer = topic.getSubscription(subName);
}
if (messageExpirer == null) {
    final String message = remoteCluster.isPresent()
            ? "Replicator not found" : getSubNotFoundErrorMessage(topicName.toString(), subName);
    resultFuture.completeExceptionally(new RestException(Status.NOT_FOUND, message));
    return;
}

Add the AbstractReplicator import in PersistentTopicsBase for this example. Please cover a dotted/custom prefix, an ordinary subscription, and a similar-looking name that does not match the full prefix plus separator.

[SEPURI-SAI-KRISHNA]

Done in 14774ae, the helper now returns Optional<String>, and the callers use that single result instead of their own startsWith check, including the two expire-messages sites that reuse it for the not-found message.

Making the separator part of the check here also fixed a case that was wrong before it: a subscription like pulsar.replication-state matched the bare prefix pulsar.repl at the call sites and was handled as a replicator. Details and the added test cases are in the PR comment.

[SEPURI-SAI-KRISHNA]

Thanks for the review, done in the latest commit.

AbstractReplicator#getRemoteCluster now returns Optional<String>, and all seven call sites use it for both detection and extraction, so their separate startsWith guards are gone. The two expire-messages sites reuse the same Optional for the not-found message, as in your example, and PersistentTopicsBase now imports AbstractReplicator.

One thing worth flagging: making the check authoritative inside the helper also fixed a case the previous version got wrong. Because the callers guarded with a bare startsWith(replicatorPrefix), a subscription such as pulsar.replication-state matched the prefix pulsar.repl without the separator and was handled as a replicator, on master it resolved to the cluster replication-state, and in my first version it fell through unchanged. It is now correctly treated as an ordinary subscription. That affects removeOrphanReplicationCursors, where such a cursor would previously have been swept as an orphan.

Test coverage as requested:

• testGetRemoteClusterRoundTripsClusterNamesContainingDots now runs the round trip across the prefixes pulsar.repl, repl and my.custom.repl, each against the clusters us-west, us-east.prod, a.b.c, cluster:1 and r3.
• testGetRemoteClusterIsEmptyForNamesThatAreNotReplicators covers an ordinary subscription, a different prefix entirely, pulsar.replication-state (prefix characters without the separator), the bare prefix with no separator or cluster, a name shorter than the prefix, and the empty string.

AbstractReplicatorTest, PersistentTopicTest, admin.PersistentTopicsTest and quickCheck all pass locally with retries disabled.