[fix][broker] Bound the local partition metadata retry when starting a geo-replicator

[merlimat]

Motivation

GeoPersistentReplicator.getLocalPartitionMetadata() asks the local admin API for the partitioned metadata of the topic being replicated. When that lookup returns 404, it takes the "legacy edge case" branch (a non-partitioned topic whose name ends in -partition-N, from PIP-433) and calls createRemoteTopicIfDoesNotExist(localTopicName), which performs the same local lookup again. That fallback is unconditional, so as long as the local lookup keeps returning 404 the replicator loops forever: no backoff, one new admin request per iteration, and the prepareCreateProducer() future never completes. AbstractReplicator.startProducer() therefore never reaches its exceptionally(...) retry/backoff path and the replicator is stuck in Starting.

For a regular (non-partitioned) topic, TopicName.getPartitionedTopicName() is the topic's own name, so the "fallback" repeats the identical request. The lookup returns 404 persistently when the local topic is deleted (or its namespace is deleted) while the replicator is starting, and nothing in the loop checks the replicator state.

Reproduced on two real brokers by creating a regular topic, deleting it, and calling the replicator's prepare step: on the unpatched code the future was still pending after 5 seconds, during which the local broker had answered 5,487 GET .../partitions requests with 404 (about 1,100 per second), while the replicator itself was already Terminated. The loop keeps going until the broker is restarted. The same loop was also triggered by segment:// topics, whose /partitions REST path does not exist; that trigger is being removed separately, but the recursion itself is unbounded by construction.

Modifications

• GeoPersistentReplicator.getLocalPartitionMetadata(): take the legacy fallback only when the failed lookup was not already for localTopicName. The fallback always passes localTopicName, so it now runs at most once and a second 404 fails the returned future with the NotFoundException, letting the existing backoff in startProducer() handle the retry. The legacy behaviour for a non-partitioned topic named -partition-N is unchanged.
• New GeoPersistentReplicatorPrepareCreateProducerTest (mocks only, broker-replication group). The mocked admin futures are completed from another thread, like the real admin client does, so that the unpatched code hangs instead of overflowing the stack.

With the fix, the real-broker reproduction above fails in 4 ms after a single local lookup.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• GeoPersistentReplicatorPrepareCreateProducerTest.testLocalTopicNotFoundFailsPreparation: with the local lookup always answering 404, the prepare future fails with NotFoundException and the local admin is queried exactly once for a regular topic and exactly twice (base name, then the -partition-0 name) for a legacy name; the remote cluster is never contacted. Both rows time out on the unpatched code because the future never completes.
• GeoPersistentReplicatorPrepareCreateProducerTest.testLegacyNonPartitionedTopicWithPartitionSuffix: the legacy fallback still creates the -partition-0 topic on the remote cluster as a non-partitioned topic (passes before and after the fix).
• Existing coverage of the prepare path still passes: OneWayReplicatorTest.testReplicatorCreateTopic, testReplicatorCreateTopicWhenTopicExistsWithDifferentTypeAcrossClusters, testReplicatorWhenPartitionCountsDiffer, AbstractReplicatorTest, PersistentTopicTest.testClosingReplicationProducerTwice / testAtomicReplicationRemoval, GeoShadowReplicatorUnsentEntryCleanupTest.

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