github-advanced-security[bot] commented on code in PR #10597:
URL: https://github.com/apache/rocketmq/pull/10597#discussion_r3535417167
##########
proxy/src/main/java/org/apache/rocketmq/proxy/service/receipt/DefaultReceiptHandleManager.java:
##########
@@ -303,4 +309,214 @@
protected ProxyContext createContext(String actionName) {
return ProxyContext.createForInner(this.getClass().getSimpleName() +
actionName);
}
+
+ /**
+ * Query POP receipt handles for diagnostics.
+ * <p>
+ * Scans all receipt handle groups matching the given consumer group name,
+ * collects diagnostic information for each unacked message handle,
+ * and returns a summary with paginated handle details.
+ * <p>
+ * This method is thread-safe: it iterates over the ConcurrentHashMap
without
+ * locking, providing a weakly consistent snapshot of the current state.
+ */
+ @Override
+ public PopReceiptHandleDiagnosticResult describePopReceiptHandles(String
group, String topic, int pageNum, int pageSize) {
+ List<PopReceiptHandleInfo> allHandles = new ArrayList<>();
+ long totalRenewTimes = 0;
+ long totalRenewRetryTimes = 0;
+ int expiredHandles = 0;
+ int totalHandleCount = 0;
+ int totalMessageCount = 0;
+
+ long currentTime = System.currentTimeMillis();
+
+ // Scan all receipt handle groups matching the group name
+ for (Map.Entry<ReceiptHandleGroupKey, ReceiptHandleGroup> entry :
receiptHandleGroupMap.entrySet()) {
+ ReceiptHandleGroupKey key = entry.getKey();
+ if (!key.getGroup().equals(group)) {
+ continue;
+ }
+
+ ReceiptHandleGroup handleGroup = entry.getValue();
+ totalHandleCount += handleGroup.getHandleNum();
Review Comment:
## CodeQL / Implicit narrowing conversion in compound assignment
Implicit cast of source type long to narrower destination type [int](1).
[Show more
details](https://github.com/apache/rocketmq/security/code-scanning/26)
##########
proxy/src/main/java/org/apache/rocketmq/proxy/service/receipt/DefaultReceiptHandleManager.java:
##########
@@ -303,4 +309,214 @@
protected ProxyContext createContext(String actionName) {
return ProxyContext.createForInner(this.getClass().getSimpleName() +
actionName);
}
+
+ /**
+ * Query POP receipt handles for diagnostics.
+ * <p>
+ * Scans all receipt handle groups matching the given consumer group name,
+ * collects diagnostic information for each unacked message handle,
+ * and returns a summary with paginated handle details.
+ * <p>
+ * This method is thread-safe: it iterates over the ConcurrentHashMap
without
+ * locking, providing a weakly consistent snapshot of the current state.
+ */
+ @Override
+ public PopReceiptHandleDiagnosticResult describePopReceiptHandles(String
group, String topic, int pageNum, int pageSize) {
+ List<PopReceiptHandleInfo> allHandles = new ArrayList<>();
+ long totalRenewTimes = 0;
+ long totalRenewRetryTimes = 0;
+ int expiredHandles = 0;
+ int totalHandleCount = 0;
+ int totalMessageCount = 0;
+
+ long currentTime = System.currentTimeMillis();
+
+ // Scan all receipt handle groups matching the group name
+ for (Map.Entry<ReceiptHandleGroupKey, ReceiptHandleGroup> entry :
receiptHandleGroupMap.entrySet()) {
+ ReceiptHandleGroupKey key = entry.getKey();
+ if (!key.getGroup().equals(group)) {
+ continue;
+ }
+
+ ReceiptHandleGroup handleGroup = entry.getValue();
+ totalHandleCount += handleGroup.getHandleNum();
+ totalMessageCount += handleGroup.getMsgCount();
+
+ // Scan handles in this group
+ handleGroup.scan((msgID, handleStr, messageReceiptHandle) -> {
+ // Apply topic filter if specified
+ if (topic != null && !topic.isEmpty() &&
!topic.equals(messageReceiptHandle.getTopic())) {
+ return;
+ }
+
+ // Decode receipt handle to get invisible time info
+ ReceiptHandle receiptHandle =
ReceiptHandle.decode(messageReceiptHandle.getReceiptHandleStr());
+ boolean isExpired = receiptHandle.getNextVisibleTime() <=
currentTime;
+
+ PopReceiptHandleInfo info = new PopReceiptHandleInfo(
+ messageReceiptHandle.getGroup(),
+ messageReceiptHandle.getTopic(),
+ messageReceiptHandle.getQueueId(),
+ messageReceiptHandle.getMessageId(),
+ messageReceiptHandle.getQueueOffset(),
+ messageReceiptHandle.getReconsumeTimes(),
+ messageReceiptHandle.getRenewTimes(),
+ messageReceiptHandle.getRenewRetryTimes(),
+ messageReceiptHandle.getConsumeTimestamp(),
+ messageReceiptHandle.getReceiptHandleStr(),
+ receiptHandle.getNextVisibleTime(),
+ receiptHandle.getInvisibleTime(),
+ receiptHandle.getBrokerName(),
+ isExpired
+ );
+ allHandles.add(info);
+ });
+ }
+
+ // Aggregate summary statistics
+ for (PopReceiptHandleInfo info : allHandles) {
+ totalRenewTimes += info.getRenewTimes();
+ totalRenewRetryTimes += info.getRenewRetryTimes();
+ if (info.isExpired()) {
+ expiredHandles++;
+ }
+ }
+
+ PopReceiptHandleGroupSummary summary = new
PopReceiptHandleGroupSummary(
+ group, totalHandleCount, totalMessageCount,
+ totalRenewTimes, totalRenewRetryTimes, expiredHandles
+ );
+
+ // Apply pagination
+ long total = allHandles.size();
+ int fromIndex = (pageNum - 1) * pageSize;
+ int toIndex = Math.min(fromIndex + pageSize, allHandles.size());
+ List<PopReceiptHandleInfo> pagedHandles;
+ if (fromIndex >= allHandles.size()) {
+ pagedHandles = new ArrayList<>();
+ } else {
+ pagedHandles = allHandles.subList(fromIndex, toIndex);
+ }
+
+ return new PopReceiptHandleDiagnosticResult(summary, pagedHandles,
total, pageNum, pageSize);
+ }
+
+ /**
+ * Query batch consumption diagnostics, aggregated per client channel.
+ * <p>
+ * Scans all receipt handle groups matching the given consumer group,
+ * aggregates unacked message statistics per Channel, and returns
+ * a summary with paginated per-channel diagnostic data.
+ * <p>
+ * This method is thread-safe: it iterates over the ConcurrentHashMap
without
+ * locking, providing a weakly consistent snapshot of the current state.
+ */
+ @Override
+ public BatchConsumeDiagnosticResult describeBatchConsumeDiagnostics(String
group, String topic, int pageNum, int pageSize) {
+ long currentTime = System.currentTimeMillis();
+
+ // Aggregate per-channel data: Channel -> aggregated stats
+ Map<Channel, ChannelAggregator> channelAggregators = new HashMap<>();
+
+ // Scan all receipt handle groups matching the group name
+ for (Map.Entry<ReceiptHandleGroupKey, ReceiptHandleGroup> entry :
receiptHandleGroupMap.entrySet()) {
+ ReceiptHandleGroupKey key = entry.getKey();
+ if (!key.getGroup().equals(group)) {
+ continue;
+ }
+
+ Channel channel = key.getChannel();
+ ReceiptHandleGroup handleGroup = entry.getValue();
+ ChannelAggregator aggregator =
channelAggregators.computeIfAbsent(channel, ChannelAggregator::new);
+
+ // Scan handles in this group
+ handleGroup.scan((msgID, handleStr, messageReceiptHandle) -> {
+ // Apply topic filter if specified
+ if (topic != null && !topic.isEmpty() &&
!topic.equals(messageReceiptHandle.getTopic())) {
+ return;
+ }
+
+ // Decode receipt handle to check expiration
+ ReceiptHandle receiptHandle =
ReceiptHandle.decode(messageReceiptHandle.getReceiptHandleStr());
+ boolean isExpired = receiptHandle.getNextVisibleTime() <=
currentTime;
+
+ aggregator.unackedMessageCount++;
+ aggregator.totalRenewTimes +=
messageReceiptHandle.getRenewTimes();
+ aggregator.totalRenewRetryTimes +=
messageReceiptHandle.getRenewRetryTimes();
+ if (isExpired) {
+ aggregator.expiredHandleCount++;
+ }
+
aggregator.topicDistribution.merge(messageReceiptHandle.getTopic(), 1,
Integer::sum);
+ });
+
+ // Handle count is per-group (one ReceiptHandleGroup per
Channel+group key)
+ aggregator.unackedHandleCount += handleGroup.getHandleNum();
Review Comment:
## CodeQL / Implicit narrowing conversion in compound assignment
Implicit cast of source type long to narrower destination type [int](1).
[Show more
details](https://github.com/apache/rocketmq/security/code-scanning/27)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
