[Bug] Timed-out topic load future remains cached and prevents retry

# Apache Pulsar issue body

## Search before reporting

- [x] I searched the Apache Pulsar issues and found nothing describing a timed-out `BrokerService.getTopic` future remaining in the `topics` cache and preventing subsequent load attempts.

## Read release policy

- [x] I understand that unsupported versions do not receive bug fixes. The production incident occurred on 4.0.10. Source inspection confirms that the relevant behavior remains in 4.0.11, 4.1.3, 4.2.2, and local `master` at commit `6d8df1058a` (2026-06-16). A deterministic regression test against `master` is planned as part of the proposed PR.

## User environment

- Broker image: `apachepulsar/pulsar-all:4.0.10`
- Deployment: Kubernetes on Linux, four brokers
- Topic type: persistent, partitioned topic
- Relevant broker configuration: defaults were used
- `topicLoadTimeoutSeconds=60`
- `managedLedgerMetadataOperationsTimeoutSeconds=60`
- `metadataStoreOperationTimeoutSeconds=30`
- Clients that observed the failure: producers, consumers, and Admin REST requests
- Client-specific details are not believed to be relevant because the failure is in the broker-side topic future cache.

## Issue description

A persistent topic partition failed to complete a cold load before `topicLoadTimeoutSeconds`. The future returned by `BrokerService.getTopic` completed exceptionally with `TimeoutException`, but the same failed future remained in `BrokerService.topics`.

Every later producer, consumer, and admin request reused that failed future and failed immediately. No request initiated a new topic-load attempt. The partition therefore remained unavailable indefinitely rather than recovering through normal client retries.

Expected behavior:

- After a topic-load attempt reaches its deadline, a later request should be able to start a fresh load attempt.
- Cleanup from the older attempt must not remove or replace a newer successful cache entry.

Actual behavior:

- The exceptionally completed future remains cached.
- `getTopic` returns it directly on every subsequent request.
- Normal retries do not execute the topic-loading path again.
- In the production incident, an admin unload failed immediately as well. Recovery required externally clearing the affected topic state; the available recovery action was destructive and caused retained history to be lost.

The incident was isolated to one partition on one broker. Other partitions in the same namespace and bundle remained healthy, and there was no cluster-wide topic-load congestion.

The exact lower-level asynchronous operation that originally stalled was not identified. This report does not claim that BookKeeper, managed-ledger metadata, topic policies, or another subsystem was definitively responsible. The cache-retention problem applies regardless of which asynchronous stage fails to finish.

## Error messages

```text
FutureUtil$LowOverheadTimeoutException:
Failed to load topic within timeout
at BrokerService.futureWithDeadline(...)(Unknown Source)
```

The broker recorded a topic-load failure after approximately the configured 60-second deadline. Subsequent calls failed immediately by reusing the cached future.

No corresponding `ManagedLedgerException`, `NotEnoughBookies`, `BKException`, fencing error, or topic-policy initialization error was found in the available broker logs.

## Relevant source behavior

In `pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java`, `getTopic` returns any cached future without checking whether it completed exceptionally:

```java
// If topic future exists in the cache returned directly regardless of whether it fails or timeout.
CompletableFuture> tp = topics.get(topicName.toString());
if (tp != null) {
return tp;
}
```

The topic future has a load deadline:

```java
final CompletableFuture> topicFuture = FutureUtil.createFutureWithTimeout(
Duration.ofSeconds(pulsar.getConfiguration().getTopicLoadTimeoutSeconds()), executor(),
() -> FAILED_TO_LOAD_TOPIC_TIMEOUT_EXCEPTION);
```

Its exception callback logs and records the failure, but it does not evict the failed future from `topics`:

```java
topicFuture.exceptionally(t -> {
// Logging omitted
pulsarStats.recordTopicLoadFailed();
return Optional.empty();
});
```

There is a `topics.remove(topicName.toString(), topicFuture)` later in `getTopic`, but it belongs to the topic-policy initialization exception branch. It is not executed when the outer topic-load deadline expires after the future has been inserted and loading has begun. This distinction was verified in 4.0.10, 4.0.11, 4.1.3, 4.2.2, and `master`.

The method Javadoc also states both that a cached future is returned “regardless of whether it fails or timeout” and that exceptions from `computeIfAbsent` remove the future. The timeout path is the uncovered case.

Cleanup exists if the underlying load eventually returns: a late-created `PersistentTopic` is closed and the matching future is removed. That does not recover when the underlying operation never completes, which was the observed failure mode.

The relevant early cache return and absence of timeout eviction are unchanged in the inspected versions:

| Version | Returns failed cached future | Removes matching future on outer load timeout |
|---|---:|---:|
| 4.0.10 | Yes | No |
| 4.0.11 | Yes | No |
| 4.1.3 | Yes | No |
| 4.2.2 | Yes | No |
| `master` (`6d8df1058a`) | Yes | No |

Known adjacent topic-load fixes are already present in 4.0.10 and therefore did not resolve this incident mechanism:

- #23004 — message-deduplication replay timeout causing topic loading to become stuck
- #23772 — managed-ledger data-ledger future never finishing
- #24785 — topic-loading latency metric and timeout handling
- #24829 — intermittent topic-policies-service error while loading topics

## Reproducing the issue

A deterministic broker unit test can reproduce the cache behavior without reproducing the unknown production stall:

1. Configure a short `topicLoadTimeoutSeconds`.
2. Make the first `loadOrCreatePersistentTopic` invocation return its `TopicLoadingContext.topicFuture` without completing the underlying load.
3. Call `BrokerService.getTopic` and wait for the expected topic-load timeout.
4. Confirm that `BrokerService.topics` still contains the exceptionally completed future.
5. Call `getTopic` for the same topic again.
6. Observe that the exact same failed future is returned and `loadOrCreatePersistentTopic` is not invoked a second time.

The desired regression test should then verify that:

1. The matching failed future is removed after timeout.
2. A second request creates a distinct load future and can succeed.
3. Late completion or cleanup from the first attempt cannot remove the replacement entry.
4. If the timeout occurs while prerequisite metadata or topic-policy work is pending, delayed completion of that prerequisite cannot insert the already-expired future into `topics`.

## Proposed direction

On topic-load timeout, conditionally remove only the matching attempt:

```java
topics.remove(topicName.toString(), topicFuture);
```

The identity-based removal is important so cleanup from an older attempt cannot evict a newer load. The path that reaches `computeIfAbsent` after prerequisite work should also avoid inserting a future that has already timed out.

This requires concurrency-focused review. Eviction permits a new attempt while the original asynchronous operation may still be running. Existing late-completion logic closes a topic whose future has already failed, but a permanently hanging lower-level operation could otherwise leave multiple stale attempts over time. The PR should keep retry behavior bounded and include tests for replacement-entry safety.

## Additional information

Production evidence supporting the impact:

- Only one partition on one broker was affected.
- The topic-load pending gauge peaked at one on that broker and zero on the other brokers.
- Topic-load failure metrics increased only on the affected broker.
- Other partitions in the same bundle remained available.
- Producer, consumer, and admin retries all reused the failed cache state.
- Clearing the topic/cache state allowed a new load attempt and the partition eventually became healthy.

Aggregate partitioned-topic stats omitted the failed partition rather than surfacing its failure, which made detection harder, but that monitoring behavior is outside the scope of this issue.

## Are you willing to submit a PR?

- [x] I’m willing to submit a PR with a focused broker fix and deterministic regression tests.

Read pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java, starting at getTopic and its timeout and cache paths. Reproduce the described deterministic scenario with a short topic-load timeout, then add focused regression coverage showing that a timed-out attempt can be retried without late cleanup removing a replacement entry.

Written by the indexing model from the issue text.