[fix][test] Wait for txn.abort() to complete to avoid AdminApiTransactionTest.testAnalyzeSubscriptionBacklogWithTransactionMarker() flaky test

[oneby-wang]

Motivation

Fix AdminApiTransactionTest.testAnalyzeSubscriptionBacklogWithTransactionMarker() flaky test.

Example failure:

  Error:  org.apache.pulsar.broker.admin.v3.AdminApiTransactionTest.testAnalyzeSubscriptionBacklogWithTransactionMarker
  [INFO]   Run 1: PASS
  Error:    Run 2: AdminApiTransactionTest.testAnalyzeSubscriptionBacklogWithTransactionMarker:1098 expected [20] but found [19]

https://github.com/oneby-wang/pulsar/actions/runs/20709883704/job/59448327174?pr=22

Modifications

Wait for txn.abort() to complete to avoid flaky test.

I checked two other test methods for topics.peekMessages() that do not use txn.abort().get(); I ran these tests a lot of times locally and found them are not flaky.

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java

Lines 970 to 979 in bf98773

	for (int i = 0; i < 2 * n; i++) {
	Transaction txn = pulsarClient.newTransaction().build().get();
	if (i % 2 == 0) {
	producer.newMessage(txn).value("msg").send();
	txn.commit().get();
	} else {
	producer.newMessage(txn).value("msg-aborted").send();
	txn.abort();
	}
	}

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java

Lines 1019 to 1027 in bf98773

	for (int i = 0; i < 2 * n; i++) {
	Transaction txn = pulsarClient.newTransaction().build().get();
	if (i % 2 == 0) {
	producer.newMessage(txn).value("msg").send();
	txn.commit().get();
	} else {
	producer.newMessage(txn).value("msg-aborted").send();
	txn.abort();
	}

I read the source code and found that topics.peekMessages() returns only when the specified numMessages have been retrieved.

pulsar/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java

Lines 921 to 927 in bf98773

	private void peekMessagesAsync(String topic, String subName, int numMessages,
	List<Message<byte[]>> messages, CompletableFuture<List<Message<byte[]>>> future, int nthMessage,
	boolean showServerMarker, TransactionIsolationLevel transactionIsolationLevel) {
	if (numMessages <= 0) {
	future.complete(messages);
	return;
	}

Verifying this change

• Make sure that the change passes the CI checks.

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: oneby-wang#23

[oneby-wang]

/pulsarbot rerun-failure-checks

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.48%. Comparing base (f101811) to head (fae9a7f).
⚠️ Report is 3 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #25125       +/-   ##
=============================================
+ Coverage     38.75%   74.48%   +35.73%     
- Complexity    13215    33694    +20479     
=============================================
  Files          1842     1899       +57     
  Lines        145545   149716     +4171     
  Branches      16916    17405      +489     
=============================================
+ Hits          56400   111516    +55116     
+ Misses        81460    29325    -52135     
- Partials       7685     8875     +1190     

Flag	Coverage Δ
inttests	26.35% <ø> (-0.24%)	⬇️
systests	23.02% <ø> (-0.05%)	⬇️
unittests	74.00% <ø> (+39.45%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.
see 1407 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.