Fixes #24908

Motivation

NonPersistentTopicTest.testMsgDropStat fails intermittently on CI:

java.lang.AssertionError: expected [true] but found [false]
    at org.apache.pulsar.client.api.NonPersistentTopicTest.testMsgDropStat(NonPersistentTopicTest.java:905)

Failure point: assertTrue(latch.await(5, TimeUnit.SECONDS)) times out because no publisher drop is observed within 5 seconds (no MessageIdImpl with entryId == -1).

Root cause (in the original test loop)

Why it flakes:

• The test sets maxConcurrentNonPersistentMessagePerConnection = 1. In ServerCnx#handleSend, the broker drops only when nonPersistentPendingMessages > maxNonPersistentMessagePerConnection. With maxConcurrentNonPersistentMessagePerConnection = 1, the first concurrent send increments the counter from 0 to 1 and is accepted. The second concurrent send checks 1 > 1 (false), then increments the counter to 2 and is accepted. Only a third overlapping send sees 2 > 1 and is dropped. Therefore, at least three overlapping sends are required to trigger any drop. The original loop does not guarantee that level of overlap.
  

  pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

  Lines 1999 to 2013 in fbc50b0

  	if (producer.isNonPersistentTopic()) {
  	// avoid processing non-persist message if reached max concurrent-message limit
  	if (nonPersistentPendingMessages > maxNonPersistentPendingMessages) {
  	final long producerId = send.getProducerId();
  	final long sequenceId = send.getSequenceId();
  	final long highestSequenceId = send.getHighestSequenceId();
  	service.getTopicOrderedExecutor().executeOrdered(producer.getTopic().getName(), () -> {
  	commandSender.sendSendReceiptResponse(producerId, sequenceId, highestSequenceId, -1, -1);
  	});
  	producer.recordMessageDrop(send.getNumMessages());
  	return;
  	} else {
  	nonPersistentPendingMessages++;
  	}
  	}

• The executor threads that call producer.send() are not coordinated to start together, so overlap of send() calls on the same connection is left to scheduler timing and is therefore probabilistic.
• The loop uses Thread.sleep(10), introducing timing assumptions that vary under CI load.
• Latch gating creates an additional race: The latch is decremented only when a drop is observed after 20% of messages have been sent. If drops occur before the 20% threshold and later sends do not drop, the latch never opens and the test times out at the await. (See NonPersistentTopicTest.java lines 0887-0894 for the latch condition and line 0905 for the timeout.)

This PR makes the publisher-drop trigger reliable without changing the test scenario or API usage.

Modifications

Test-only changes in NonPersistentTopicTest.testMsgDropStat. No production code or docs changed.

• Replace the unsynchronized, sleep-based loop with a small synchronized batch of send() calls.
  Use a CyclicBarrier to start a fixed number of producer.send() calls at the same time, explicitly creating the required overlap without sleeps or ad-hoc timing. Execute the batch inside a bounded Awaitility wait until at least one send() returns a MessageIdImpl with entryId == -1, then assert that condition.

• Changed maxConcurrentNonPersistentMessagePerConnection from 1 to 0

  conf.setMaxConcurrentNonPersistentMessagePerConnection(0);

  Rational: In ServerCnx#handleSend for non-persistent topics, a publish is dropped only when nonPersistentPendingMessages > maxNonPersistentMessagePerConnection. Lowering the limit from 1 to 0 reduces the overlap requirement from three in-flight sends to two, which, combined with the synchronized start, makes the publisher drop reliably detectable in this test.

Verifying this change

• Make sure that the change passes the CI checks.

Personal CI Results

Tested in Personal CI fork: vinkal-chudgar#3

Status: All checks have passed (48 successful checks, 3 skipped)

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

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: vinkal-chudgar#3