[fix][test] Fix flaky testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader

[lhotari]

Fixes #25081

Motivation

SystemTopicBasedTopicPoliciesServiceTest.testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader
is flaky. The test creates a real __change_events reader, closes it, and then expects a follow-up
prepareInitPoliciesCacheAsync(...) that reuses the closed reader to fail (so the cache-init cleanup path
runs). It fails intermittently in CI (e.g. an AssertionError at the Assert.fail() after
prepareFuture.get(), or a timeout on a follow-up assertion).

Two independent races, both rooted in the test relying on real asynchronous timing, cause the flakiness:

1. The closed reader does not reliably fail the read. initPolicesCache(...) calls
   reader.hasMoreEventsAsync(), which delegates to Reader.hasMessageAvailableAsync(). On a closed
   consumer this is non-deterministic: it can still answer from cached state instead of failing with
   AlreadyClosedException. When it does, initPolicesCache(...) reaches the end of the topic and completes
   successfully, so the expected exception never happens (and "Failed to check the move events for the system
   topic" is never logged).

2. A background re-initialization leaves a completed init future behind. Closing the reader drives
   readMorePoliciesAsync() into cleanPoliciesCacheInitMap(), which removes the namespace from
   policyCacheInitMap. But a background topic load can re-run prepareInitPoliciesCacheAsync(...) for the
   namespace right after, leaving a completed init future. The follow-up prepareInitPoliciesCacheAsync(...)
   then observes that future and short-circuits through the existing-future branch instead of
   re-initializing — so it never throws.

This is a test-only fragility; the production code is correct and self-heals in both cases.

Modifications

Test-only change in testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader, making the follow-up
initialization deterministically take the failure path the test asserts on:

• Reuse the closed reader through a Mockito spy whose hasMoreEventsAsync() deterministically fails with
  PulsarClientException.AlreadyClosedException, so initPolicesCache(...) always fails at the read.
• Drop any lingering init future (policyCacheInitMap.remove(namespace)) right before the call so it
  re-initializes instead of short-circuiting on a previously completed future.
• Stub createSystemTopicClient(namespace) to return that failing spy, so whichever initialization wins the
  race (the test's, or a background re-init) fails in the same asserted way.

The assertions the test makes (exceptional completion; policyCacheInitMap/readerCaches cleaned; the
"Failed to create reader" log absent; the "Failed to check the move events" log present;
cleanPoliciesCacheInitMap/cleanupFailedPolicyCacheInit each invoked once; no "Recursive update") are
unchanged — the fix only makes the conditions they assert on deterministic.

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by the existing test
SystemTopicBasedTopicPoliciesServiceTest.testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader;
the change makes it deterministic. Verified locally by running the test 70 times (40 in-process via
invocationCount and 30 in separate JVMs via --rerun-tasks), all green; before the fix the same loops
reproduced the failure.

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

This is a test-only change and does not affect any of the following: dependencies, public API, schema,
default configuration values, threading model, binary protocol, REST endpoints, admin CLI options, metrics,
or deployment.

• 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