[merlimat]

Motivation

Problem. Two tests of CI - Unit - Brokers - Broker Group 2 fail their first attempt in nearly every CI run and
pass on the TestNG retry. TestNG reports a retried attempt as skipped, so the job stays green and Develocity counts
the method as passed: the failures show only in the <skipped message=...> testcases of the XML reports, which
green jobs don't upload, and in Develocity's per-class scan data. In the 59 scheduled master runs from 2026-10-05 to
2026-11-03:

Test	First attempt failed	Failure of the first attempt (XML reports of the 4 failed Broker Group 2 jobs of 10-30 .. 11-01)
BrokerInternalClientConfigurationOverrideTest.testPulsarServicePulsarClientConfiguration	59 of 59 runs, in 2-41 ms	AssertionError: expected [sensitive] but found []
BrokerTestUtilTest.testReceiveMessagesQuietTime	55 of 59 runs	TooFewActualInvocations: consumer.receiveAsync(); Wanted 3 times ... But was 1 time (3 jobs), TooManyActualInvocations ... Wanted 2 times ... But was 3 times for consumer2 (1 job)

They are the two most frequent hidden first-attempt failures of the unit test jobs, and because of them no
Broker Group 2 job passes without its retries.

1. testPulsarServicePulsarClientConfiguration sets the brokerClient_* properties and the broker client
authentication parameters on the running broker's configuration, then reads the configuration of the broker's
internal client:

Properties config = pulsar.getConfiguration().getProperties();
config.setProperty("brokerClient_operationTimeoutMs", "60000");
config.setProperty("brokerClient_statsIntervalSeconds", "10");
pulsar.getConfiguration().setBrokerClientAuthenticationParameters("sensitive");
ClientConfigurationData clientConf = ((PulsarClientImpl) pulsar.getClient()).getConfiguration();

But the broker creates that client only once, when it first needs it:

public synchronized PulsarClient getClient() throws PulsarServerException {
    if (this.client == null) {
        try {
            this.client = createClientImpl(null, null);

TestNG runs the class's methods in name order, and testOldNamespacePolicy (added by #22535) runs before this test.
It creates a producer, so the broker loads the topic, and the topic policies service creates its reader on
__change_events with pulsar.getClient(). The two methods before it had set the same brokerClient_* properties
for the clients that they create, so the client gets those, but the authentication parameters are still the
default empty ones when this test reads them. The retry passes because TestRetrySupport restarts the broker
before it, and the retry is then the first to ask for the client. So the failure doesn't depend on timing: it
happened in every run, and also when the class ran a minute into a warm test JVM.

Example. One run of the class with -PtestRetryCount=0: 5 tests completed, 1 failed, in this order:

testBrokerServicePulsarClientConfiguration     0.006 s  ok
testNamespaceServicePulsarClientConfiguration  0.002 s  ok
testOldNamespacePolicy                         0.349 s  ok
  [pulsar-ordered-OrderedExecutor-0-0:NamespaceEventsSystemTopicFactory] - Create topic policies system topic
  client {client=persistent://prop/oldNsWithDefaultNonNullValues/__change_events ...
testPulsarServiceAdminClientConfiguration      0.003 s  ok
testPulsarServicePulsarClientConfiguration     0.001 s  FAILED: expected [sensitive] but found []

testPulsarServiceAdminClientConfiguration has the same dependency on pulsar.getAdminClient(), which the broker
also creates once; it passes only because nothing creates the admin client earlier.

2. testReceiveMessagesQuietTime checks that BrokerTestUtil.receiveMessages measures its quiet time across
all its consumers (#23876): consumer1 gets a message at 300 ms and consumer2 one at 400 ms, so with a 500 ms quiet
time consumer1 must receive once more after its own quiet time has passed (times(3)), and the call must end at
900 ms. The test ran on real time, with the messages delivered by delayed executors:

CompletableFuture<Message> messageCompletableFuture =
        CompletableFuture.supplyAsync(() -> mock(Message.class),
                CompletableFuture.delayedExecutor(consumer1DelayMs, TimeUnit.MILLISECONDS));

That left it 100-250 ms of slack: each message had to arrive before its receive timed out at 500 ms (100 ms early
for consumer2's), consumer1's second receive timeout had to fire less than 100 ms late, so that it saw consumer2's
message as recent, and the whole call had to end within 1150 ms.

In CI the class runs first in its test JVM, or nearly first: in each of the 58 swept runs before #27299 it was the
first class of its fork, and since #27299 it is the third class of testShard2, 1.5 s after the JVM starts. The
other three test JVMs start brokers at the same time, and the 4 vCPU runner is at 99-100 % CPU during the first
attempt (Develocity's CPU samples of scan 22v6ot3ibswp2). The first attempt then misses its slack: in the
dominant failure (But was 1 time), neither consumer1's message nor consumer2's has arrived when consumer1's first
receive times out at 500 ms, so consumer1 stops after one receive. The retry, in a JVM that has run everything once,
passes.

Example. To see where the time goes I ran the same scenario with a timeline, in a Linux container on two CPUs of
an Apple M5 Pro, four JVMs at a time (two copies of the scenario and two JVMs running a broker test class: four test
JVMs on about the CPU of CI's runner), with -XX:ActiveProcessorCount=4 for CI's pool sizes. The original failed 6
of 8 runs, all as in CI with consumer1 receiving only once, and in 5 of them neither message arrived before its
500 ms timeout. Inside the timed part:

• the delayed supplier's mock(Message.class), the JVM's first mock of that type, took 59-272 ms in the runs that
  lasted long enough to measure it;
• the delayed tasks and the receive timeouts fired up to 0.6 s late.

Creating the message mocks before the timed part is not enough. In the same setup it still failed 2 of 8 runs: once
consumer1's second receive timed out 194 ms late and stopped (Wanted 3 times ... But was 2 times), and once the
first receives were set up 245 ms after the start, so the call ended at 1174 ms. A real-time version of this test
keeps those margins, so on a busy runner it stays flaky.

Modifications

Change.

1. BrokerInternalClientConfigurationOverrideTest.setup() sets the overrides before the broker starts, and the
   tests of the two clients that the broker creates once, the internal client and the admin client, only read their
   configuration:

   @BeforeClass
   @Override
   protected void setup() throws Exception {
       // The broker creates its internal client and admin client once, when it first needs them (in this class, at
       // the latest when testOldNamespacePolicy loads a topic), so their configuration must be set before it starts.
       Properties config = conf.getProperties();
       config.setProperty("brokerClient_operationTimeoutMs", "60000");
       config.setProperty("brokerClient_statsIntervalSeconds", "10");
       conf.setBrokerClientAuthenticationParameters("sensitive");
       super.baseSetup();
   }

   The tests of the replication and namespace clients, which create new clients, still set their properties
   themselves.

2. BrokerTestUtil.receiveMessages takes its clock, its receive timeouts and the executor of each next receive from
   a package-private ReceiveClock. Every caller gets ReceiveClock.SYSTEM, which behaves as before: System.nanoTime(),
   CompletableFuture.orTimeout(), and ForkJoinPool.commonPool(), which is the executor that thenComposeAsync
   uses without one. On JDK 25 and later the default async pool is the common pool itself; on JDK 21 it is too,
   except when the common pool can't run tasks in parallel, and then CompletableFuture also swaps an explicit
   common pool for its default pool (checked on 21.0.10, 25.0.5 and 26.0.1, with one CPU and more).

   return clock.orTimeout(consumer.receiveAsync(), receiveTimeoutNanos)
           .handle((msg, t) -> {
               long currentNanos = clock.nanoTime();
               ...
           }).thenComposeAsync(receiveMoreAndNextTimeout -> {
               ...
           }, clock.executor());

   testReceiveMessagesQuietTime runs the same scenario on a manual clock: time moves from one event to the next
   (a message, a receive timeout), and the receive loop runs in the test's thread, so the outcome depends only on the
   times of the scenario. Its assertions are the same, and the end time is now exact: the call ends at 900 ms,
   consumer2's message plus the quiet time. The test no longer waits ~1 s of real time.

   CompletableFuture<Long> endNanos = BrokerTestUtil.receiveMessagesAsync(messageHandler,
                   Duration.ofMillis(quietTimeMs), clock, Stream.of(consumer1, consumer2))
           .thenApply(__ -> clock.nanoTime());
   clock.run();
   assertThat(endNanos).isCompleted();

   The manual clock fires a timeout 1 ns after its deadline, as a timer fires some time after it: at exactly the
   deadline the quiet time would have passed by 0 ns, and the receive loop would wait 0 ns more, again and again.

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests: it changes the two tests themselves.

Every run below had the retries off (-PtestRetryCount=0, or -DtestRetryCount=0 with TestNG run directly) and a
fresh JVM. "testShard2's classes" means its first three, PulsarClusterMetadataTeardownTest,
PulsarInitialNamespaceSetupTest and BrokerTestUtilTest, in one JVM in that order, as in CI.

Where	Before	After
Apple M5 Pro, ./gradlew :pulsar-broker:test --rerun	each class alone: quiet time passed 6 of 6 (an idle Mac meets its margins), config override failed 1 of 1	testShard2's classes, then the config-override class, in one JVM (-PtestMaxParallelForks=1): both passed 20 of 20
Linux container on 2 CPUs of that Mac, JDK 25, -XX:ActiveProcessorCount=4: four JVMs at a time, two with testShard2's classes and two with the config-override class, 10 rounds	quiet time failed 7 of 20 (Wanted 3 times ... But was 2 times), config override failed 20 of 20	both passed 20 of 20
The same on 1 CPU, 6 rounds		both passed 12 of 12

• The quiet-time test still catches the bug it was added for: with BrokerTestUtil changed back to a quiet time per
  consumer (each consumer with its own last-message time, as before [fix][test] Fix quiet time implementation in BrokerTestUtil.receiveMessages #23876), it fails with
  TooFewActualInvocations ... Wanted 3 times.
• It no longer waits for real time: it takes 0.55 s on the Mac, almost all of it Mockito's start, instead of 1.4 s.
• The other users of receiveMessages pass with the retries off: the four KeySharedSubscriptionTest methods that
  call it, UnloadSubscriptionTest (32 tests) and TransactionEndToEndTest.testAfterDeleteTopicOtherTopicCanRecover.
• ./gradlew quickCheck passes.

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

[codelipenghui]

Code Review approval for commit 0bdadb2.

Summary:

No actionable findings identified within the reviewed scope. The test-only approach correctly fixes fixture initialization order and makes shared quiet-time assertions deterministic. The configuration baseline reproduced the reported failure; the new timing test rejected a deliberate regression. All seven final selected tests passed with retries disabled.