[improve][cli] Add v4-client subcommands to pulsar-perf and pulsar-client - #26480

This change is necessary to be able to profile the V4 client changes and ordinary V4 topic behavior in master branch.
The profiling test https://github.com/apache/pulsar/blob/master/tests/integration/src/test/java/org/apache/pulsar/tests/integration/profiling/PulsarProfilingTest.java would be updated so that both V4 and V5 pulsar-perf could be used in profiling.

One follow-up from reading #26466, which this PR builds on. Not a defect in either change — the rationale comments for LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS describe the world as it was before #26466, and this PR carries the wording forward unchanged:

• PerfClientUtils.java:61 — "...and since these recorders are static fields the JVM pays for every perf subcommand, not just the one being run."
• PerfClientUtilsTest.java:237 — "The perf clients hold their latency recorders in static fields, so every subcommand allocates them on startup rather than only the one being run."

#26466 converted every one of those to an instance field in the same commit that added these comments — git grep -E "static.*Recorder [a-zA-Z]" -- pulsar-testclient/src/main returns nothing at bc1785dcd8, and it still returns nothing here. So the premise the paragraphs rest on is the exact thing that commit fixed.

The conclusion is untouched — 3 digits is still clearly right, and the ceiling the test pins is still worth pinning. It is only the "every subcommand pays" justification that no longer applies; the cost is now bounded to the one command being run. Measured on HdrHistogram 2.1.9, for the ranges the test uses:

(The "11-22 MB" figure holds if you include the pre-#26466 33 h range, which came to 21.00 MB.)

Since PerfClientUtils.java is already touched here, it may be easiest to reword both in this PR — something like "a single Recorder over the ranges used here costs 14-16 MB at 5 digits, per running command" — rather than leaving a comment that contradicts the field declarations a few lines away. Equally fine as a separate cleanup if you would rather not widen this one.

Thank you for this — rebuilding the pre-split files and diffing them against base + V5 subclass for
all seven splits is exactly the check this PR needed, and it found a real regression I had missed.
I have addressed everything below. Three findings I could not reproduce and am pushing back on with
mechanism; a few others were right on the fact but wrong on the impact, and I have noted where,
because I think the corrected version is worth having on the record.

Blocking

1. ProducerSocket.onMessage lost its null guard — fixed.
You are right, and it was my mistake. Restored both the if (this.result != null) guard and the
base's log.info().attr("ack", msg).log("Received ack").

One correction on provenance: the class was not copied from 820300f93b^. That revision has both
the guard and "Received ack" — in fact every historical revision has the guard.
git log -S'ack= ' -- .../CmdProduce.java returns exactly two commits: #3835 (2019), which
introduced the slf4j LOG.info("ack= {}", msg), and #25511, which converted it to slog. So what I
wrote was a hand-reconstruction of the pre-#25511 format string with the guard dropped, not a
copy of any single revision. The onConnect(Session) throws InterruptedException in the same class
came from ConsumerSocket, which confirms it was reassembled by hand. I have dropped that dead
throws too.

Also a small correction on reachability, which is why I have not treated it as blocking for the
release: against Pulsar's own WebSocket endpoint the window is unreachable — every
sendAckResponse(...) call site in ProducerHandler is inside onWebSocketText or the sendAsync
continuation it starts, and error paths go through session.close(...), i.e. @OnWebSocketClose.
Only a non-Pulsar endpoint behind --url ws://… can send an unsolicited text frame. And "or after
close()" does not hold: by then send() has assigned result at least once, so a late frame
completes an already-completed future, which is a no-op rather than an NPE. It is still a real
fidelity regression in a commit presented as a move, so it is fixed regardless.

V5 behaviour changes not disclosed in the description (2, 3, 4, 6, 7)

All correct, and the description was wrong to claim "unchanged in behaviour" without qualification.
I have added a Behaviour changes on the V5 commands subsection to Modifications (text in Part 3
below) listing every one of them. Kept as they are — they are fixes — with these notes:

• 2 (send() reordering) — kept. Worth spelling out that it and finding 1 are the same bug seen
  from two sides: the result != null guard was the band-aid over exactly this window, so the fix is
  "reorder and keep the guard", not one or the other. result is volatile, the caller is
  strictly serialised, and ProducerHandler sends exactly one ack per inbound message and nothing
  unsolicited, so the new ordering introduces no window of its own.
• 3 (close() null guard) — kept, disclosed.
• 4 (broadened catch) — kept, disclosed, with two corrections. The fall-through was not
  reachable in practice at base: both existing exit procedures
  (PerformanceTransactionTest:74-79, Oauth2PerformanceTransactionTest:112-117) throw on a
  non-zero code and production uses System::exit; and under a returning procedure the
  message.id() NPE landed inside the pre-existing catch (Exception ackEx), so it was miscounted
  as an ack failure rather than escaping. Also, widening to Exception is forced, not
  discretionary — the V4 and V5 overrides throw two different PulsarClientException classes
  (o.a.p.client.api vs o.a.p.client.api.v5), so no narrower shared catch exists, and the try body
  is only receive(consumer).
• 6 (consumerType attribute) — restored. scalableConsumerType is V5-only so it cannot live in
  the shared base, but the base can ask for it: consumerTypeForLog() returns the shared
  subscriptionType by default and PerformanceConsumer overrides it with scalableConsumerType.
  The V5 line is byte-identical to base again, and consume-v4 gets the equivalent.
• 7 (new user-visible output) — disclosed, and the list is longer than the four you found. The
  same sweep also picks up CmdProduce.java:80-81, PerformanceConsumer.java:94,
  PerformanceReader.java:63 and PerformanceTransaction.java:104, plus a brand-new
  unconditional INFO line at PerformanceConsumer.java:88 ("Using V5 scalable-topic consumer API")
  on every pulsar-perf consume run. Nine sites, all listed in the description now.

Findings I could not reproduce (5, 10, 18)

I checked each of these three carefully and I do not think the mechanism exists. Happy to be shown
wrong.

5. PerformanceReaderBase dropping FutureUtil.waitForAll — no behavioural difference.
FutureUtil.waitForAll is CompletableFuture.allOf (FutureUtil.java:56-61), and allOf is not
fail-fast: BiRelay.tryFire returns without firing unless both operand results are already present,
so the composed future settles only after every input has settled. The drain loop therefore reports a
failure no later than the removed waitForAll(...).get(), not later. The throwable is identical
too — allOf's BiRelay tests the left operand first and so reports the leftmost failure, which is
also where the in-order drain throws, and reportGet unwraps the CompletionException, so both
surface ExecutionException(cause). Neither version closes already-created readers on failure, so no
leak changed. And PerformanceConsumer.java:328-331 already used the bare drain loop before this PR
— PerformanceReader.java:139 was the module's only waitForAll caller — so this is a
normalisation onto the existing pattern rather than a dropped guard. I tried restoring it and then
reverted; happy to add it back if you still want the symmetry, but it would make failure reporting
strictly later.

10. Positional getDeliverAtTime() assertions — the ordering hazard is not reachable.
The consumer at :154 uses the default Exclusive subscription type
(ConsumerConfigurationData:90), which is served by PersistentDispatcherSingleActiveConsumer — and
trackDelayedDelivery is overridden only by PersistentDispatcherMultipleConsumers and its Classic
variant, so the single-active dispatcher inherits the no-op default at Dispatcher.java:118 and the
delayed-delivery tracker is never consulted. Even on a Shared subscription it would not defer:
InMemoryDelayedDeliveryTracker.addMessage rejects deliverAt <= getCutoffTime()
(clock.millis() + tickTimeMillis), and deliverAfter(0, SECONDS) is already due at send time —
"one tick of deferral" is precisely the window that cutoff excludes. With one non-batching producer
on a non-partitioned topic and each send(...).get() awaited, publish order is fixed and arrival
order follows it.

That said, your instinct that the test reads fragilely is fair, so I took the half of the suggestion
that costs nothing: the assertions now also check the payload identity of each message
(assertThat(new String(first.getData(), UTF_8)).isEqualTo("delayed")), with a comment recording why
the order is deterministic. A future reordering now fails as "expected delayed, got plain" instead of
as a confusing deliverAtTime mismatch, and the incidental ordering check is still pinned.

18. isNotBlank guards in buildV4ClientBuilder — the divergence cannot occur.
rootParams.tlsTrustCertsFilePath is not independent of client.conf; it is filled from it. The
option declares descriptionKey = "tlsTrustCertsFilePath" (PulsarClientTool.java:92-95) and the
commander's default-value provider is PulsarClientPropertiesProvider extends PropertiesDefaultProvider over the very same Properties object that becomes the loadConf map
(:138-139, :176, :228). So line 257 rewrites the value loadConf just applied, or applies the
genuine CLI override; it can only be null when the conf key is absent too, i.e. null over null.
Adding a guard would also remove the ability to clear a conf-supplied path with
--tlsTrustCertsFilePath "".

serviceUrl is the opposite of the suggested change: v4 ClientBuilderImpl.serviceUrl rejects a
blank value with checkArgument rather than nulling the field, and the call is load-bearing because
loadConf only reads the serviceUrl key while the shipped conf/client.conf supplies
brokerServiceUrl / webServiceUrl (which PulsarClientPropertiesProvider.create synthesises).
Guarding it would turn a clear "Param serviceUrl must not be blank" into a vaguer failure from
build(). I have left both unconditional and added a comment saying why, since it clearly reads as a
bug otherwise.

Test coverage (8, 9)

8. transaction-v4 had no functional test — added, and mutation-verified.
Correct, and this was the right thing to catch. New PerformanceTransactionV4Test (reusing
PerformanceTransactionTest's coordinator setup but on plain persistent:// topics, verified with
the v4 SDK client the base test provides):

• testTransactionV4CommitsWhatItProduced — a committed run makes what it produced visible and
  drops the consume backlog by one per transaction, so the acks really were committed.
• testTransactionV4AbortUndoesBothTheSendAndTheAck — -abort leaves nothing visible on the produce
  topic, leaves the consume backlog at 10, and leaves all ten messages immediately redeliverable on
  the tool's own subscription. That last part is what separates a real abort from a transaction that
  was merely never ended (whose acks would stay pending until the timeout).

Both pass. Mutation-tested to confirm they are load-bearing:

I want to be straight about that last row rather than claim more than I verified — and chasing it
turned up a stale comment of my own. .sendTimeout(0, TimeUnit.SECONDS) was annotated in this PR
with "a send timeout and a transaction are mutually exclusive on the v4 client". That was true:
ProducerBase.newMessage(Transaction) threw "Only producers disabled sendTimeout are allowed to
produce transactional messages" — until bbf2a47867f "[fix][txn] Allow producer enable send timeout
in transaction (#16519)" deleted the check in 2022. Today the method only does
checkArgument(txn instanceof TransactionImpl).

So the line is still right, for a weaker reason: a send timeout is now permitted, but it still fails
the send on its own schedule and takes the transaction with it (that is exactly what #16519's own
testSendTxnMessageTimeout asserts), so disabling it leaves --txn-timeout as the only deadline.
Both stale comments (PerformanceTransactionV4 and PerformanceProducerV4) now say that instead.
The line stays unpinned by the test because the failure only appears once a send actually times out,
which five quick transactions never do — pinning it would need a deliberately stalled transaction,
which I did not think was worth the CI minutes.

One correction to the finding: the V5 counterpart has one real test for the transaction command
(PerformanceTransactionTest:139), not three — the other two drive PerformanceProducer and
PerformanceConsumer with -txn. Lower baseline, same conclusion.

9. consume-v4's timestamp flags only tested negatively — added, and mutation-verified.
Correct, and your mutation claim was right: both lines are dead under the defaults
(startTimestamp = 0L fails the > 0L guard, endTimestamp = Long.MAX_VALUE can never be
exceeded). New PulsarClientToolTest.testConsumeV4StartAndEndTimestamp publishes two messages, reads
their real publish time back as the boundary, waits for the millisecond clock to cross it, publishes
two more, then asserts -stp boundary+1 from -p Earliest prints only the newer pair and
-etp boundary with -n 4 prints only the older pair.

Worth noting -etp has been equally untested since #24521 — this PR restores the -stp seek that
#25917 dropped, it did not introduce the gap.

Test quality (10, 11, 12, 13)

• 10 — see above.
• 11 (backlog assert without Awaitility) — wrapped in Awaitility.untilAsserted. Your stated
  mechanism is not quite the one that bites, though: the assert is not reached straight off the
  latch, because stop(thread) joins the perf thread and that thread cannot die until
  PerfClientUtils.closeClient has run a full client.close(), which flushes the grouped acks and
  blocks on the CloseConsumer round trip the broker can only answer after applying them. On the
  happy path it is ordered. What is genuinely unguarded is the unchecked 30 s join and the
  interrupt reportDone delivers after the latch, which can abort that close and delete the only
  happens-before edge. Awaitility closes both, and costs nothing on the ordered path — so the fix is
  right, for a different reason.
• 12 (timeOut and the discarded cause) — both fixed: timeOut = 60000 on each, and future.get()
  before the receive, so a non-zero produce-v4 exit surfaces as its own exception instead of as
  assertNotNull(message) ten seconds later. One correction: "every other @Test in the file"
  carries a timeOut is not so — :74, :492 and :512 are bare @Test, and :512 blocks on a
  10 s receive just like these. And the PR did not introduce the swallowing; the bodies are
  unchanged from base apart from produce → produce-v4, and the same pattern is in the untouched
  testProducePartitioningKey. Fixed here anyway, because this PR is what makes these two execute
  in CI again.
• 13 (reflection in TestCmdConsume) — done: the three lines, the java.lang.reflect.Field
  import and the now-unnecessary throws Exception on setUp are all gone, replaced by
  cmdConsume.subscriptionName = "my-sub";. Note the same pattern in TestCmdRead:24 is not
  fixable this way — startMessageId is still private in CmdRead.

v4-only correctness (14, 15)

• 14 (interrupt on the wrong thread) — fixed by capturing the worker thread and interrupting
  that, so V5 is unchanged (there the captured thread is the current thread) and v4 no longer
  interrupts a client-owned executor thread. Your mechanism is exactly right on all three sub-claims.
  One correction on impact: I could not find any path that completes a v4 ack future with an
  InterruptedException in its cause chain — every such catch in ConsumerBase/ConsumerImpl is in
  the synchronous .get() wrappers — so it is a latent wrong-thread hazard rather than an observable
  miscount today. Still worth fixing since the ack-side callback is new here.

  On reflection I applied it to all three callbacks, not just the ack: sendAndRecord and
  endTransaction in the same file interrupt Thread.currentThread() from futures that complete on
  client-internal threads too, and all three are invoked from the same runWorker loop, so the
  capture is equally valid. You noted the send side is pre-existing and it is — but leaving two of
  three wrong next to a comment explaining why the third is right just invites the question. It is
  four lines and no behaviour change on V5.

• 15 (nativeAvroSchemaOrNull silently shipping raw JSON) — changed to throw. One correction:
  it is not reachable from any CLI invocation today. The only AVRO-typed schema CmdProduceV4 can
  build is Schema.AUTO_PRODUCE_BYTES(Schema.generic(…AVRO)), and AvroBaseStructSchema.getNativeSchema()
  is a hard Optional.of(schema); a malformed avro: definition already throws in parseAvroSchema
  during construction. And in the counterfactual, AUTO_PRODUCE_BYTES validates on encode, so raw
  JSON would throw SchemaSerializationException at send() — a different loud failure, not corrupt
  data on the topic. So it is an invariant worth keeping explicit rather than a live defect; I used
  IllegalStateException to match that (the argument-level errors in this class use
  IllegalArgumentException).

Nits (16, 17, 19, 20)

• 16 (dead startMessageId()) — deleted, all three sites. Confirmed no callers repo-wide; not
  picocli-bound and not used reflectively (TestCmdRead reflects on the field).
• 17 (unreachable proxy-protocol throw) — deleted. Confirmed unreachable: updateConfig() is the
  preRun() hook, the execution strategy runs it before RunLast and returns early on a non-zero
  code, and the lazy supplier is only created after that guard. It is redundant a second time over
  — ClientBuilderImpl.proxyServiceUrl already does checkArgument(proxyProtocol != null, …). I
  kept the outer isNotBlank guard: unlike the V5 path this builder uses loadConf(), so passing
  nulls through would clear a proxyServiceUrl / proxyProtocol coming from client.conf.
• 19 (--jsse-provider gated on V5, unconditional on v4/admin) — real observation, but I do not
  think either direction of alignment is an improvement, so I have left it. Removing the wantsTls
  gate is not possible: PulsarClientBuilderV5#tlsPolicy sets useTls=true for CLIENT_DEFAULT, so
  a provider flag on a pulsar:// URL would force a TLS handshake against a plaintext broker — the
  exact bug that gate and its javadoc exist to prevent. Gating the v4/admin writes instead is also
  wrong: wantsTls inspects arguments.serviceURL, not the separate adminUrl the admin builder is
  handed, and the v4 write is not always inert —
  ClientTlsFactorySupport.foldOAuth2IdpPolicy inherits conf.getJsseProvider() /
  getJcaProvider() into the CLIENT_OAUTH2 IdP policy, and needsClientTlsFactory() builds the
  factory on a plaintext broker when the OAuth2 plugin carries IdP TLS material. If anything it is
  the V5 side that drops a meaningful pin in that narrow case, and closing it would need
  tlsPolicy(CLIENT_OAUTH2, …), which would clobber the plugin's own IdP trust via putIfAbsent.
  That is a PIP-478 question, not this PR's.
• 20 (fully-qualified java.util types) — fixed; imported Arrays, Set, TreeSet and dropped
  the qualifiers. A scan of all 34 changed files found no other occurrence.

Follow-up comment: stale LATENCY_HISTOGRAM_SIGNIFICANT_DIGITS rationale

Right on both counts, and thank you for catching that the premise was the very thing #26466 fixed.
Every Recorder under pulsar-testclient/src/main is an instance field at HEAD (15 declarations),
and the static→instance conversion happened in the same commit that added those comments.

I re-measured against HdrHistogram 2.2.2 and your figures reproduce exactly — 16.00 MB / 0.18 MB for
1 h in µs, 14.00 MB / 0.16 MB for 10 d in ms; the old "11-22 MB" is what the pre-#26466 ranges
(SECONDS.toMicros(120000) / toMillis(120000)) produce, which is the other half of the same
staleness. Both comments are reworded to "14-16 MB", with the per-command cost expressed as "a
command holds several of them (a live and a cumulative recorder per measured latency)" rather than
the old "every subcommand pays". The 3-digit conclusion and the 512 KiB test bound are unchanged.

Not addressed

Nothing. Every numbered finding plus the follow-up comment is either fixed, disclosed in the
description, or answered above with the mechanism.