[improve][admin] Add counter for marker messages in PersistentTopics.analyzeSubscriptionBacklog() rest api

[oneby-wang]

Fixes #25082

Motivation

PersistentTopics.analyzeSubscriptionBacklog() rest api returns the counters of backlog, but the counters don't return the amount of marker messages.

Modifications

Add counter for marker messages in PersistentTopics.analyzeSubscriptionBacklog() rest api, add tests.

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#15

[github-actions]

@oneby-wang Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[lhotari]

LGTM

[oneby-wang]

/pulsarbot rerun-failure-checks

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 74.47%. Comparing base (e041fab) to head (bf0ed12).
⚠️ Report is 4 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #25091       +/-   ##
=============================================
+ Coverage     39.20%   74.47%   +35.27%     
- Complexity    13498    33675    +20177     
=============================================
  Files          1842     1899       +57     
  Lines        145491   149660     +4169     
  Branches      16907    17394      +487     
=============================================
+ Hits          57038   111465    +54427     
+ Misses        80844    29315    -51529     
- Partials       7609     8880     +1271     

Flag	Coverage Δ
inttests	26.35% <0.00%> (-0.32%)	⬇️
systests	23.05% <0.00%> (-0.08%)	⬇️
unittests	74.00% <100.00%> (+38.85%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...pulsar/broker/admin/impl/PersistentTopicsBase.java	69.44% <100.00%> (+56.25%)	⬆️
...ker/service/persistent/PersistentSubscription.java	76.65% <100.00%> (+27.10%)	⬆️

... and 1401 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.

[oneby-wang]

Since uncommitted or aborted messages will not be delivered to consumer, counting them into numMessages ormarkerMessages seems all improper.

The aborted messages is filtered by AbstractBaseDispatcher.

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

Lines 199 to 206 in 4495525

	} else if (((PersistentTopic) subscription.getTopic())
	.isTxnAborted(new TxnID(msgMetadata.getTxnidMostBits(), msgMetadata.getTxnidLeastBits()),
	entry.getPosition())) {
	individualAcknowledgeMessageIfNeeded(Collections.singletonList(entry.getPosition()),
	Collections.emptyMap());
	entries.set(i, null);
	entry.release();
	continue;

The uncommitted messages will not be read and delivered before committed.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

Lines 4789 to 4791 in 4495525

	public Position getMaxReadPosition() {
	return this.transactionBuffer.getMaxReadPosition();
	}

Maybe we should modify OpScan scan params and scan logic and filter aborted messages in analyzeBacklog method, so the backlog would be consistent with consumer backlog.

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java

Lines 129 to 132 in 4495525

	if (cursor.hasMoreEntries(searchPosition)) {
	OpReadEntry opReadEntry = OpReadEntry.create(cursor, searchPosition, batchSize,
	this, OpScan.this.ctx, null, null, false);
	ledger.asyncReadEntries(opReadEntry);

@lhotari WDYT?

[lhotari]

Since uncommitted or aborted messages will not be delivered to consumer, counting them into numMessages ormarkerMessages seems all improper.

I agree that it would be useful to have a separate counter for such transactional messages that are part of aborted transactions. For the transaction markers themselves, it's useful to have them part of marker messages.
I think that the aborted transactional message counter could be handled in a separate PR.

[oneby-wang]

/pulsarbot rerun-failure-checks