[ISSUE #10654] Benchmark support grpc

[bxfjb]

Which Issue(s) This PR Fixes

• Fixes [Enhancement] Benchmark support grpc #10654

Brief Description

How Did You Test This Change?

[RockteMQ-AI]

Review by github-manager-bot

Summary

Adds gRPC-based benchmark producer and consumer tools (GrpcProducer, GrpcConsumer) using the rocketmq-client-java SDK, along with corresponding shell scripts for launching and shutting down the benchmarks. This extends the existing remoting-based benchmark suite with gRPC protocol support.

Findings

• [Warning] GrpcConsumer.java:103 — successList is a LinkedList shared across multiple scheduled tasks (lines 135–143) without synchronization. The scheduled consumer callback and the stats snapshot task both access it concurrently. Consider using ConcurrentLinkedQueue or Collections.synchronizedList() to avoid potential ConcurrentModificationException or lost counts.

• [Warning] GrpcProducer.java:259 — awaitTermination(1, TimeUnit.DAYS) is effectively a no-timeout wait. If any producer thread fails to exit (e.g. stuck in a blocking send), the process will hang indefinitely. Consider a more reasonable timeout (e.g. 60 seconds) followed by shutdownNow() and a warning log.

• [Info] grpc-consumer.sh:18 / grpc-producer.sh:18 — $@ is unquoted. If any argument contains spaces, it will be word-split. Consider "$@" for robustness. Note: the existing benchmark scripts (produce.sh, consume.sh) have the same pattern, so this is consistent but could be improved across the board.

• [Info] GrpcProducer.java:147-148 — accessKey/secretKey default to "ak"/"sk". This is fine for a benchmark tool, but worth a brief comment noting these are placeholder credentials for local testing only.

• [Info] example/pom.xml — The dependency uses rocketmq-client-java-noshade version 5.0.9-SNAPSHOT. If this is intentional for development, consider adding a comment. If a release version is available, prefer that for reproducibility.

• [Info] GrpcConsumer.java:152 — System.currentTimeMillis() is used for latency measurement. System.nanoTime() would be more accurate for measuring elapsed time (immune to wall-clock adjustments). Same applies to GrpcProducer.java timing.

Verdict

Good addition to the benchmark toolkit — gRPC protocol coverage is a natural complement to the existing remoting benchmarks. The main concerns are the thread-safety issue in GrpcConsumer (should be fixed) and the infinite termination timeout in GrpcProducer (should be bounded). The rest are minor suggestions.

---

Automated review by github-manager-bot

[bxfjb]

Review by github-manager-bot

Summary

Adds gRPC-based benchmark producer and consumer tools (GrpcProducer, GrpcConsumer) using the rocketmq-client-java SDK, along with corresponding shell scripts for launching and shutting down the benchmarks. This extends the existing remoting-based benchmark suite with gRPC protocol support.

Findings

• [Warning] GrpcConsumer.java:103 — successList is a LinkedList shared across multiple scheduled tasks (lines 135–143) without synchronization. The scheduled consumer callback and the stats snapshot task both access it concurrently. Consider using ConcurrentLinkedQueue or Collections.synchronizedList() to avoid potential ConcurrentModificationException or lost counts.
• [Warning] GrpcProducer.java:259 — awaitTermination(1, TimeUnit.DAYS) is effectively a no-timeout wait. If any producer thread fails to exit (e.g. stuck in a blocking send), the process will hang indefinitely. Consider a more reasonable timeout (e.g. 60 seconds) followed by shutdownNow() and a warning log.
• [Info] grpc-consumer.sh:18 / grpc-producer.sh:18 — $@ is unquoted. If any argument contains spaces, it will be word-split. Consider "$@" for robustness. Note: the existing benchmark scripts (produce.sh, consume.sh) have the same pattern, so this is consistent but could be improved across the board.
• [Info] GrpcProducer.java:147-148 — accessKey/secretKey default to "ak"/"sk". This is fine for a benchmark tool, but worth a brief comment noting these are placeholder credentials for local testing only.
• [Info] example/pom.xml — The dependency uses rocketmq-client-java-noshade version 5.0.9-SNAPSHOT. If this is intentional for development, consider adding a comment. If a release version is available, prefer that for reproducibility.
• [Info] GrpcConsumer.java:152 — System.currentTimeMillis() is used for latency measurement. System.nanoTime() would be more accurate for measuring elapsed time (immune to wall-clock adjustments). Same applies to GrpcProducer.java timing.

Verdict

Good addition to the benchmark toolkit — gRPC protocol coverage is a natural complement to the existing remoting benchmarks. The main concerns are the thread-safety issue in GrpcConsumer (should be fixed) and the infinite termination timeout in GrpcProducer (should be bounded). The rest are minor suggestions.

Automated review by github-manager-bot

The Bot's comment is confusing: looks like some comments point to blank line number or referred a non-exist variable.

[RockteMQ-AI]

@bxfjb Thanks for pointing that out — you're right, the previous review had several factual inaccuracies. I re-checked the actual diff and here are the corrected findings:

Corrected Findings

• [Warning] GrpcConsumer.java — snapshotList (a LinkedList<Long[]>) is shared across the scheduled consumer callback and the stats snapshot task without synchronization. This could lead to ConcurrentModificationException or lost counts. Consider using ConcurrentLinkedQueue or Collections.synchronizedList().

• [Warning] GrpcProducer.java — sendThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS) is effectively an infinite wait. If any producer thread fails to exit, the process will hang indefinitely. Consider a bounded timeout (e.g. 60 seconds) followed by shutdownNow() and a warning log.

• [Info] grpc-consumer.sh:18 / grpc-producer.sh:18 — $@ is unquoted. If any argument contains spaces, it will be word-split. Consider "$@" for robustness. (The existing benchmark scripts have the same pattern, so this is consistent but could be improved.)

• [Info] GrpcProducer.java / GrpcConsumer.java — accessKey/secretKey default to "12345678"/"rocketmq2". Fine for a benchmark tool, but worth a brief comment noting these are placeholder credentials for local testing only.

• [Info] GrpcConsumer.java / GrpcProducer.java — System.currentTimeMillis() is used for latency measurement. System.nanoTime() would be more accurate for elapsed time (immune to wall-clock adjustments).

Apologies for the earlier inaccuracies. The two Warning items (thread-safety of snapshotList and unbounded awaitTermination) are the most important ones to address.

---

Automated review by github-manager-bot (corrected)

[RockteMQ-AI]

Corrected Findings

• [Warning] GrpcConsumer.java:109 — snapshotList (a LinkedList<Long[]>) is shared across multiple scheduled tasks (lines ~115–140) without synchronization. The scheduled consumer callback adds to it and the stats snapshot task reads from it concurrently. Consider using ConcurrentLinkedQueue or Collections.synchronizedList() to avoid potential ConcurrentModificationException.

• [Warning] GrpcProducer.java:186 — sendThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS) is effectively an infinite wait. If any producer thread fails to exit, the process will hang indefinitely. Consider a more reasonable timeout (e.g. 60 seconds) followed by shutdownNow() and a warning log.

• [Info] grpc-consumer.sh:18 / grpc-producer.sh:18 — $@ is unquoted. If any argument contains spaces, it will be word-split. Consider "$@" for robustness. (Note: the existing benchmark scripts have the same pattern, so this is consistent.)

• [Info] GrpcProducer.java:74-75 — accessKey/secretKey default to "12345678"/"rocketmq2". Fine for a benchmark tool, but worth a brief comment noting these are placeholder credentials for local testing only.

• [Info] GrpcConsumer.java:157 / GrpcProducer.java:157 — System.currentTimeMillis() is used for latency measurement. System.nanoTime() would be more accurate for measuring elapsed time (immune to wall-clock adjustments).

Apologies for the confusion in the previous review. The core concerns (thread safety in GrpcConsumer and unbounded termination timeout in GrpcProducer) remain valid.

---

Automated review by github-manager-bot