[fix][cli][branch-4.2] Don't pass unset pulsar-perf pending-message options to the producer

[lhotari]

Fixes #26340

This targets branch-4.2 only. On master, #25887 moved pulsar-perf to the V5 client, where these options no longer reach a producer at all, so there is nothing to fix there. The client-side half of this — making the no-memory-limit defaults behave as defaults rather than being overwritten by a pass-through — is #26342 on master.

Motivation

pulsar-perf produce can exhaust the client's direct memory against a slow broker (#26340), because on branch-4.2 nothing bounds its producer at all:

• PerformanceBaseArguments.memoryLimit is a long with no initializer, and PerfClientUtils passes it on unconditionally, so the client memory limit is disabled unless --memory-limit is given.
• --max-outstanding defaults to ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES, which is 0, and that value was passed to maxPendingMessages(...) unconditionally. 0 is the client's "no message-count limit", so the pending-message queue is unbounded too.

With both bounds gone, a producer that outruns its broker buffers without limit. ProducerImpl only creates its semaphore if (conf.getMaxPendingMessages() > 0).

#15283 already fixed exactly this shape for --max-outstanding-across-partitions, by only applying the option when it was set. It could not fix --max-outstanding the same way, because at the time there was nothing for an unset value to fall back to on the producer path this tool uses. There is: PulsarClient.newProducer(Schema) gives a producer the pre-PIP-120 defaults (1000 / 50000) when the client memory limit is disabled — #15723 added that precisely so a producer without byte-based backpressure still has some. pulsar-perf misses it only because it calls the no-argument newProducer(), which never got that treatment.

Note that #15283's fix is for partitioned topics: the across-partitions budget is what PartitionedProducerImpl divides between partitions, and it is ignored on a non-partitioned topic. --max-outstanding is the one that matters there, and it is the one still being passed through.

Modifications

pulsar-testclient:

• --max-outstanding and --max-outstanding-across-partitions become Integer with no default, so "the user did not pass this flag" is distinguishable from "the user asked for 0". Each is applied to the builder only when it was given, which is [fix][tools] Only apply maxPendingMessagesAcrossPartitions if it presents #15283's fix generalised to both options.
• createProducerBuilder uses newProducer(Schema.BYTES) instead of the no-argument newProducer(). They are the same builder for a byte[] producer, except that this one carries the client's defaults for a client with the memory limit disabled — which is what an unset option now falls back to.

pulsar-client:

• ProducerConfigurationData.setMaxPendingMessagesAcrossPartitions no longer rejects a value below maxPendingMessages. That check makes the two setters order-dependent, and it is what forced [fix][tools] Only apply maxPendingMessagesAcrossPartitions if it presents #15283 to guard on > 0 rather than on "was it set": with the per-producer default now in place at 1000, pulsar-perf -p 500 would throw IllegalArgumentException before this change. The relationship is enforced where it is used — PartitionedProducerImpl lowers the per-partition limit to its share of the budget when a budget is set, and the budget means nothing on a non-partitioned topic.

Effective behaviour for pulsar-perf produce with no flags, on a client whose memory limit is disabled: maxPendingMessages = 1000, maxPendingMessagesAcrossPartitions = 50000, blockIfQueueFull = true — so it throttles instead of buffering without limit, on partitioned and non-partitioned topics alike. Passing -o or -p still overrides, including -o 0 for the old unbounded behaviour.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• PerformanceProducerTest#testPendingMessageLimitsAreLeftToTheClientWhenUnset — with no flags, the builder carries positive limits. Fails before this change, where they are 0.
• PerformanceProducerTest#testGivenPendingMessageLimitsAreApplied — -p on its own is applied and does not fail producer creation. Fails without the pulsar-client change.
• ProducerBuilderImplTest#testAcrossPartitionsLimitBelowMaxPendingMessagesIsAccepted — pins the relaxed validation.
• ProducerBuilderImplTest#testProducerBuilderImplWhenMaxPendingMessagesAcrossPartitionsPropertyIsInvalidErrorMessages — updated for the new message; a negative value is still rejected.
• The existing PerformanceProducerTest#testMaxOutstanding (added by [fix][tools] Only apply maxPendingMessagesAcrossPartitions if it presents #15283) and #testBatchingDisabled still pass.

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

• pulsar-perf produce without -o / -p now has a bounded pending-message queue where it previously had none. It blocks rather than failing, since it sets blockIfQueueFull(true), so a run against a slow broker is throttled instead of ending in an OutOfMemoryError. A run that relied on unbounded buffering can restore it with -o 0, or raise the bound with -o <n>.

• -o and -p no longer report a default in --help, because they no longer have one.

• maxPendingMessagesAcrossPartitions no longer throws IllegalArgumentException when set below maxPendingMessages; it is accepted, and the per-partition limit is lowered to its share as before. Only code that relied on the exception is affected, and such a call could not previously succeed.

Documentation

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

The option help text states that an unset value is left to the client.

[void-ptr974]

Thanks for the fix. One small edge case worth handling: maxPendingMessagesAcrossPartitions / numPartitions can become 0 for high-partition topics. Since 0 means unlimited, this may accidentally disable the pending-message limit instead of enforcing the across-partitions budget.