[improve][cli] Add client side looping in "pulsar-admin topics analyze-backlog" cli to avoid potential HTTP call timeout by oneby-wang · Pull Request #25126 · apache/pulsar

oneby-wang · 2026-01-06T04:47:59Z

Motivation

Use client-side looping instead of increasing broker settings to avoid potential HTTP call timeout in "pulsar-admin topics analyze-backlog" cli.

Modifications

Add client-side looping, add test.

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

Documentation

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

Matching PR in forked repository

PR in forked repository: oneby-wang#21

…-backlog

lhotari · 2026-01-20T09:02:25Z

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java

+                    mergedResult.setLastMessageId(currentResult.getLastMessageId());
+                }
+
+                if (!mergedResult.isAborted() || mergedResult.getEntries() >= backlogScanMaxEntries) {


the !mergedResult.isAborted() part isn't correct.

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

Lines 119 to 128 in da0d116

if (remainingEntries.get() <= 0) {

log.warn("[{}] Scan abort after reading too many entries", OpScan.this.cursor);

callback.scanComplete(lastSeenPosition, ScanOutcome.ABORTED, OpScan.this.ctx);

return;

}

if (System.currentTimeMillis() - startTime > timeOutMs) {

log.warn("[{}] Scan abort after hitting the deadline", OpScan.this.cursor);

callback.scanComplete(lastSeenPosition, ScanOutcome.ABORTED, OpScan.this.ctx);

return;

}

The purpose of the CLI loop is to keep on iterating until the result is COMPLETED or the number of scanned entries exceeds backlogScanMaxEntries.

lhotari · 2026-01-20T09:04:46Z

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java

+                String[] messageIdSplits = mergedResult.getLastMessageId().split(":");
+                MessageIdImpl nextScanMessageId =
+                        new MessageIdImpl(Long.parseLong(messageIdSplits[0]), Long.parseLong(messageIdSplits[1]) + 1,
+                                partitionIndex);


Instead of parsing the message Id String, it would be better to use org.apache.pulsar.client.api.MessageIdAdv interface to get the ledgerId and entryId.

Could you provide me more information, I couldn't find one method that could parse ledgerId:entryId string to a MessageIdAdv instance.

lhotari · 2026-01-20T09:21:41Z

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java

+                    print("Analyze backlog progress, scanned entries: " + mergedResult.getEntries()
+                            + ", scan max entries: " + backlogScanMaxEntries);


It would be useful to print the current value of mergedResult in json format without linefeeds so that the CLI output can be parsed as NDJSON.

There are two ObjectMapper in CliCommand, so we should use the one without pretty printer to print json without linefeeds, and then pass the json string to print() method. Is my understanding correct?

pulsar/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CliCommand.java

Lines 140 to 141 in 85625e0

private static final ObjectMapper MAPPER = ObjectMapperFactory.create();

private static final ObjectWriter WRITER = MAPPER.writerWithDefaultPrettyPrinter();

pulsar/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CliCommand.java

Lines 120 to 130 in 85625e0

<T> void print(T item) {

try {

if (item instanceof String) {

commandSpec.commandLine().getOut().println(item);

} else {

prettyPrint(item);

}

} catch (Exception e) {

throw new RuntimeException(e);

}

}

lhotari · 2026-01-20T09:22:59Z

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java

+                                partitionIndex);
+                startPosition = Optional.of(nextScanMessageId);
+            }
+            print(mergedResult);


output would have to be json format without linefeeds to make the CLI output parseable as NDJSON. The last line in the output would be the final result. Perhaps there could be a command line option to configure whether ndjson should be used since it's not pretty printed.

Perhaps there could be a command line option to configure whether ndjson should be used since it's not pretty printed.

A little bit confused. Does this option only take effect on the final result, or does it also apply to the intermediate results?

lhotari · 2026-01-20T09:24:16Z

pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java

    }

+    @Test
+    public void topicsAnalyzeBacklog() throws Exception {


A test without mocks, similar to the tests in #25127 would be useful in addition since it would serve as an integration test.

I haven't found similar examples for reference. Could you provide me some examples.

I don't know how to invoke admin CLI in integration tests and how to receive the outputs(since they are printed to the console output).

Seems we need to handle the test in integration module like CLITest?

https://github.com/apache/pulsar/blob/85625e0f100479dd95fb1311aeb52411b6b0a25d/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/CLITest.java

Is there a better approach? Running the tests in integration module requires local image building.

Seems we need to handle the test in integration module like CLITest?

https://github.com/apache/pulsar/blob/85625e0f100479dd95fb1311aeb52411b6b0a25d/tests/integration/src/test/java/org/apache/pulsar/tests/integration/cli/CLITest.java

Is there a better approach? Running the tests in integration module requires local image building.

Yes it's a bit clumbersome.

To build an image quickly, use this script:

./build/build_java_test_image.sh

That requires setting export PULSAR_TEST_IMAGE_NAME=apachepulsar/java-test-image:latest environment variable so that the apachepulsar/java-test-image:latest docker image is used instead of the default apachepulsar/pulsar-test-latest-version:latest which takes much longer to build.
If you are running the test in IntelliJ, you can edit the default value at

pulsar/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java

Line 62 in 15daf18

"apachepulsar/pulsar-test-latest-version:latest");

For running integration tests on command line, this is the sequence I have used:

# compile integration test dependencies mvn -am -pl tests/integration -Dcheckstyle.skip=true -Dlicense.skip=true -Dspotbugs.skip=true -DskipTests install ./build/build_java_test_image.sh export PULSAR_TEST_IMAGE_NAME=apachepulsar/java-test-image:latest # run the test mvn -DintegrationTests -pl tests/integration -DtestRetryCount=0 -DredirectTestOutputToFile=false test \ -Dtest=TestClassNameHere

If you are working on the test alone, you obviously don't have to rebuild the image after each change.

oneby-wang · 2026-02-05T07:43:07Z

@lhotari Could help me solve the above questions when you have a moment? Especially about how to write integration tests easily in admin CLI module.

I'll refactor this PR using the API that PR #25127 provided once I'm back.

oneby-wang added 8 commits January 4, 2026 14:34

[improve][cli] Add client side looping to pulsar-admin topics analyze…

db9b65b

…-backlog

[improve][cli] Add test

f96b397

[improve][cli] Modify code and test

4cfda14

[improve][cli] Fix checkstyle

78efb48

[improve][cli] Add verifyNoMoreInteractions

0b6084a

[improve][cli] Optimize code layout

25b93c6

[improve][cli] Optimize code

0588618

[improve][cli] Add guard safe

7f0172b

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Jan 6, 2026

[improve][cli] Modify option name

b48e0eb

Technoboy- assigned oneby-wang Jan 7, 2026

lhotari reviewed Jan 20, 2026

View reviewed changes

oneby-wang mentioned this pull request Jan 21, 2026

[improve][admin] Add client side looping to analyze-backlog in Topics to avoid potential HTTP call timeout #25127

Open

15 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[improve][cli] Add client side looping in "pulsar-admin topics analyze-backlog" cli to avoid potential HTTP call timeout#25126

[improve][cli] Add client side looping in "pulsar-admin topics analyze-backlog" cli to avoid potential HTTP call timeout#25126
oneby-wang wants to merge 9 commits intoapache:masterfrom
oneby-wang:pulsar_cli_client_side_analyze_backlog

oneby-wang commented Jan 6, 2026

Uh oh!

lhotari Jan 20, 2026

Uh oh!

lhotari Jan 20, 2026

Uh oh!

oneby-wang Jan 21, 2026 •

edited

Loading

Uh oh!

lhotari Jan 20, 2026

Uh oh!

oneby-wang Jan 21, 2026

Uh oh!

lhotari Jan 20, 2026

Uh oh!

oneby-wang Jan 21, 2026

Uh oh!

lhotari Jan 20, 2026

Uh oh!

oneby-wang Jan 21, 2026 •

edited

Loading

Uh oh!

oneby-wang Jan 21, 2026 •

edited

Loading

Uh oh!

lhotari Feb 5, 2026

Uh oh!

oneby-wang commented Feb 5, 2026 •

edited

Loading

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

	if (remainingEntries.get() <= 0) {
	log.warn("[{}] Scan abort after reading too many entries", OpScan.this.cursor);
	callback.scanComplete(lastSeenPosition, ScanOutcome.ABORTED, OpScan.this.ctx);
	return;
	}
	if (System.currentTimeMillis() - startTime > timeOutMs) {
	log.warn("[{}] Scan abort after hitting the deadline", OpScan.this.cursor);
	callback.scanComplete(lastSeenPosition, ScanOutcome.ABORTED, OpScan.this.ctx);
	return;
	}

		print("Analyze backlog progress, scanned entries: " + mergedResult.getEntries()
		+ ", scan max entries: " + backlogScanMaxEntries);

	private static final ObjectMapper MAPPER = ObjectMapperFactory.create();
	private static final ObjectWriter WRITER = MAPPER.writerWithDefaultPrettyPrinter();

	<T> void print(T item) {
	try {
	if (item instanceof String) {
	commandSpec.commandLine().getOut().println(item);
	} else {
	prettyPrint(item);
	}
	} catch (Exception e) {
	throw new RuntimeException(e);
	}
	}

Conversation

oneby-wang commented Jan 6, 2026

Motivation

Modifications

Verifying this change

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

Documentation

Matching PR in forked repository

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

oneby-wang Jan 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

oneby-wang Jan 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

oneby-wang Jan 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

oneby-wang commented Feb 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

oneby-wang Jan 21, 2026 •

edited

Loading

oneby-wang Jan 21, 2026 •

edited

Loading

oneby-wang Jan 21, 2026 •

edited

Loading

oneby-wang commented Feb 5, 2026 •

edited

Loading