[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