[ISSUE #9638] Add PushConsumer mqadmin diagnostic command

[btlqql]

Which Issue(s) This PR Fixes

Fixes #9638

Problem

mqadmin consumeMessage uses DefaultMQPullConsumer, so operators who need to reproduce PushConsumer delivery behavior must compile and run a separate demo. That also makes it difficult to inspect live delivery metadata, per-queue progress, orderly consumption, and message tracing from the standard administration tool.

Root Cause

The tools module did not expose a PushConsumer-backed command or a bounded execution model suitable for both interactive diagnostics and automation.

Changes

• add mqadmin consumeMessageByPush backed by DefaultMQPushConsumer
• support concurrent or orderly listeners, tag expressions, configurable batch size, first/last/timestamp start positions, and optional message tracing
• support message-count and time limits while retaining an unlimited interactive mode
• print message metadata, optional body/properties, and a deterministic per-queue progress summary
• validate invalid booleans, ranges, charsets, trace settings, timestamp format, and wait-time overflow before starting the consumer
• register the command in MQAdminStartup
• add focused tests for command wiring, configuration validation, listener accounting, output formatting, shutdown, interruption, and failure paths

Impact

The change is isolated to the tools module. Existing commands and client APIs are unchanged. Operators can now diagnose PushConsumer behavior with the same mqadmin distribution used for other RocketMQ administration tasks.

Validation

• mvn -pl tools -Dtest=PushConsumeMessageConfigTest,PushConsumeMessageObserverTest,PushConsumeMessagePrinterTest,PushConsumeMessageCommandTest -Dspotbugs.skip=true -Djacoco.skip=true test
• 47 tests passed
• Checkstyle passed with 0 violations

[RockteMQ-AI]

Review by github-manager-bot

Summary

This PR adds a mqadmin consumeMessageByPush command that uses DefaultMQPushConsumer to reproduce PushConsumer delivery behavior from the standard administration tool. It fills a real gap: operators currently need to compile and run a separate demo to inspect live delivery metadata, per-queue progress, orderly consumption, and message tracing.

Findings

• [Info] PushConsumeMessageCommand — Clean separation of concerns: Config (parsing/validation), Observer (message handling), Printer (output), Command (orchestration). The factory pattern for consumerFactory enables thorough unit testing without a real broker.

• [Info] PushConsumeMessageConfig — Comprehensive validation:

  • maxMessages > 0 || maxMessages == -1 (unlimited)
  • durationSeconds > 0 || durationSeconds == -1
  • batchSize range check (1–1024)
  • Timestamp format validation for TIMESTAMP consume mode
  • The fromCommandLine() static factory is clean and testable.

• [Info] PushConsumeMessageObserver — Implements both MessageListenerConcurrently and MessageListenerOrderly. The CountDownLatch completion mechanism with awaitCompletion() is a good pattern for bounded execution. The snapshot() method provides atomic summary data.

• [Info] PushConsumeMessagePrinter — The output format includes per-queue progress, delivery latency, and message details. The printStarted() banner confirms subscription before messages arrive, which is helpful for interactive use.

• [Warning] PushConsumeMessageObserver.consumeMessage() — When maxMessages is reached, the latch counts down but the consumer continues running until shutdown() in the finally block. There is a small window where additional messages could be delivered between the latch countdown and the consumer shutdown. This is benign (messages are still processed normally), but consider setting a flag to stop accepting new messages once the limit is reached, to get an accurate count.

• [Info] --orderly flag — Correctly switches between MessageListenerOrderly and MessageListenerConcurrently. The cast is safe because the observer implements both interfaces.

Positive Observations

• Well-structured: Config/Observer/Printer/Command separation makes each piece independently testable.
• Good test coverage: Tests cover config validation, observer behavior (both concurrent and orderly), printer output, and command lifecycle.
• Feature parity: Supports trace, batch, consume-from (latest/earliest/timestamp), and sub-expression filtering — matching the existing consumeMessage command options.
• Clean shutdown: Consumer is always shut down in the finally block, even on error.

Suggestions

• Consider adding a --timeout option as an alternative to --maxMessages and --duration for time-bounded consumption (currently --duration serves this role, but a separate --timeout for overall wall-clock time could be clearer).
• The printStarted() output could include the resolved broker addresses for the subscribed topic, helping operators verify routing before messages arrive.

Overall this is a practical addition to the operations toolkit with clean architecture. 👍

---

Automated review by github-manager-bot