[Studio] feat: Feature/rip 2 pr1 proto serialization

WORKSPACE

@@ -158,6 +158,22 @@ load("@rules_java//java:repositories.bzl", "rules_java_dependencies", "rules_jav
 rules_java_dependencies()
 rules_java_toolchains()
 
+# rules_proto 5.3.0-21.7 — last 5.x release, WORKSPACE-compatible (avoids
+# bazel_features Bzlmod requirement introduced in 6.0.0).
+# Uses git archive URL because the GitHub Release asset was removed (returns 404).
+http_archive(
+    name = "rules_proto",
+    sha256 = "dc3fb206a2cb3441b485eb1e423165b231235a1ea9b031b4433cf7bc1fa460dd",
+    strip_prefix = "rules_proto-5.3.0-21.7",
+    urls = [
+        "https://github.com/bazelbuild/rules_proto/archive/refs/tags/5.3.0-21.7.tar.gz",
+    ],
+)
+
+load("@rules_proto//proto:repositories.bzl", "rules_proto_dependencies", "rules_proto_toolchains")
+rules_proto_dependencies()
+rules_proto_toolchains()
+
 load("@rules_java//toolchains:local_java_repository.bzl", "local_java_repository")
 local_java_repository(
   name = "jdk8",

common/src/test/java/org/apache/rocketmq/common/ServiceThreadTest.java

@@ -161,8 +161,11 @@ public void testNoWakeupLostUnderStress() throws Exception {
     public void serviceThreadShouldNotLoseWakeupUnderStress() throws Exception {
         final int stressIterations = 10000;
         final int wakerThreads = 4;
-        final long waitTimeoutMs = 20;
-        final long lostWakeupThresholdMs = 18;
+        // A delivered wakeup returns in microseconds while a lost one blocks for the whole
+        // interval. Keep the interval and the threshold far above CI scheduler/GC jitter
+        // (which can easily exceed tens of milliseconds) so the two cases stay distinguishable.
+        final long waitTimeoutMs = 2000;
+        final long lostWakeupThresholdMs = 1000;
 
         StressServiceThread service = new StressServiceThread();
         AtomicInteger activeIteration = new AtomicInteger(-1);

proxy/BUILD.bazel

@@ -15,12 +15,26 @@
 # limitations under the License.
 #
 load("//bazel:GenTestRules.bzl", "GenTestRules")
+load("@rules_proto//proto:defs.bzl", "proto_library")
+
+proto_library(
+    name = "proxy_admin_proto",
+    srcs = ["src/main/proto/proxy_admin.proto"],
+)
+
+# java_proto_library: native rule in Bazel 6.x; rules_java 7.x does not export it.
+java_proto_library(
+    name = "proxy_admin_java_proto",
+    deps = [":proxy_admin_proto"],
+    visibility = ["//visibility:public"],
+)
 
 java_library(
     name = "proxy",
     srcs = glob(["src/main/java/**/*.java"]),
     visibility = ["//visibility:public"],
     deps = [
+        ":proxy_admin_java_proto",
         "//auth",
         "//broker",
         "//client",
@@ -80,6 +94,7 @@ java_library(
     deps = [
 	    "//auth",
         ":proxy",
+        ":proxy_admin_java_proto",
         "//:test_deps",
         "//broker",
         "//client",

proxy/pom.xml

@@ -32,8 +32,47 @@
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
         <project.root>${basedir}/..</project.root>
+        <protobuf.version>3.20.1</protobuf.version>
+        <grpc.version>1.53.0</grpc.version>
     </properties>
 
+    <build>
+        <extensions>
+            <extension>
+                <groupId>kr.motd.maven</groupId>
+                <artifactId>os-maven-plugin</artifactId>
+                <version>1.7.1</version>
+            </extension>
+        </extensions>
+        <plugins>
+            <plugin>
+                <groupId>org.xolstice.maven.plugins</groupId>
+                <artifactId>protobuf-maven-plugin</artifactId>
+                <version>0.6.1</version>
+                <configuration>
+                    <protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
+                    <protoSourceRoot>${basedir}/src/main/proto</protoSourceRoot>
+                    <outputDirectory>${basedir}/target/generated-sources/protobuf/java</outputDirectory>
+                    <clearOutputDirectory>false</clearOutputDirectory>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-checkstyle-plugin</artifactId>
+                <configuration>
+                    <excludes>**/generated-sources/**/*.java</excludes>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
     <dependencies>
         <dependency>
             <groupId>org.apache.rocketmq</groupId>

proxy/src/main/java/org/apache/rocketmq/proxy/common/BatchConsumeClientDiagnostics.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.common;
+
+import java.util.Map;
+
+/**
+ * Diagnostic information for batch consumption, aggregated per client channel.
+ * <p>
+ * Combines data from ReceiptHandleManager (unacked message statistics)
+ * with enrichment from ConsumerManager and GrpcClientSettingsManager.
+ * <p>
+ * Key diagnostic use cases:
+ * - Identify clients with excessive unacked messages (batch too large)
+ * - Detect clients with high expired handle counts (timeout issues)
+ * - Monitor renewal patterns per client (ChangeInvisibleTime frequency)
+ * - Correlate unacked count with configured receiveBatchSize
+ */
+public class BatchConsumeClientDiagnostics {
+    private final String clientId;
+    private final String channelId;
+    private final int unackedMessageCount;
+    private final int unackedHandleCount;
+    private final long totalRenewTimes;
+    private final long totalRenewRetryTimes;
+    private final int expiredHandleCount;
+    private final Map<String, Integer> topicDistribution;
+    private final String consumeType;
+    private final String messageModel;
+    private final int receiveBatchSize;
+    private final long longPollingTimeoutMs;
+    private final long lastRttMs;
+    private final long connectTime;
+
+    public BatchConsumeClientDiagnostics(String clientId, String channelId,
+        int unackedMessageCount, int unackedHandleCount,
+        long totalRenewTimes, long totalRenewRetryTimes, int expiredHandleCount,
+        Map<String, Integer> topicDistribution,
+        String consumeType, String messageModel,
+        int receiveBatchSize, long longPollingTimeoutMs,
+        long lastRttMs, long connectTime) {
+        this.clientId = clientId;
+        this.channelId = channelId;
+        this.unackedMessageCount = unackedMessageCount;
+        this.unackedHandleCount = unackedHandleCount;
+        this.totalRenewTimes = totalRenewTimes;
+        this.totalRenewRetryTimes = totalRenewRetryTimes;
+        this.expiredHandleCount = expiredHandleCount;
+        this.topicDistribution = topicDistribution;
+        this.consumeType = consumeType;
+        this.messageModel = messageModel;
+        this.receiveBatchSize = receiveBatchSize;
+        this.longPollingTimeoutMs = longPollingTimeoutMs;
+        this.lastRttMs = lastRttMs;
+        this.connectTime = connectTime;
+    }
+
+    public String getClientId() { return clientId; }
+    public String getChannelId() { return channelId; }
+    public int getUnackedMessageCount() { return unackedMessageCount; }
+    public int getUnackedHandleCount() { return unackedHandleCount; }
+    public long getTotalRenewTimes() { return totalRenewTimes; }
+    public long getTotalRenewRetryTimes() { return totalRenewRetryTimes; }
+    public int getExpiredHandleCount() { return expiredHandleCount; }
+    public Map<String, Integer> getTopicDistribution() { return topicDistribution; }
+    public String getConsumeType() { return consumeType; }
+    public String getMessageModel() { return messageModel; }
+    public int getReceiveBatchSize() { return receiveBatchSize; }
+    public long getLongPollingTimeoutMs() { return longPollingTimeoutMs; }
+    public long getLastRttMs() { return lastRttMs; }
+    public long getConnectTime() { return connectTime; }
+}

proxy/src/main/java/org/apache/rocketmq/proxy/common/BatchConsumeGroupSummary.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.common;
+
+/**
+ * Aggregated summary statistics for batch consumption diagnostics across a consumer group.
+ * <p>
+ * Provides a high-level view of the batch consumption state for a group,
+ * useful for quickly identifying groups with excessive unacked messages,
+ * high renewal rates, or expired handles across all clients.
+ */
+public class BatchConsumeGroupSummary {
+    private final String group;
+    private final int totalClients;
+    private final int totalUnackedMessages;
+    private final int totalUnackedHandles;
+    private final int totalExpiredHandles;
+    private final long totalRenewTimes;
+    private final long totalRenewRetryTimes;
+
+    public BatchConsumeGroupSummary(String group, int totalClients,
+        int totalUnackedMessages, int totalUnackedHandles, int totalExpiredHandles,
+        long totalRenewTimes, long totalRenewRetryTimes) {
+        this.group = group;
+        this.totalClients = totalClients;
+        this.totalUnackedMessages = totalUnackedMessages;
+        this.totalUnackedHandles = totalUnackedHandles;
+        this.totalExpiredHandles = totalExpiredHandles;
+        this.totalRenewTimes = totalRenewTimes;
+        this.totalRenewRetryTimes = totalRenewRetryTimes;
+    }
+
+    public String getGroup() { return group; }
+    public int getTotalClients() { return totalClients; }
+    public int getTotalUnackedMessages() { return totalUnackedMessages; }
+    public int getTotalUnackedHandles() { return totalUnackedHandles; }
+    public int getTotalExpiredHandles() { return totalExpiredHandles; }
+    public long getTotalRenewTimes() { return totalRenewTimes; }
+    public long getTotalRenewRetryTimes() { return totalRenewRetryTimes; }
+}

proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java

@@ -29,4 +29,6 @@ public class ContextVariable {
     public static final String PROTOCOL_TYPE = "protocol-type";
     public static final String NAMESPACE = "namespace";
     public static final String CLIENT_TYPE = "client-type";
+    public static final String SSL_ENABLED = "ssl-enabled";
+    public static final String AUTH_USERNAME = "auth-username";
 }

proxy/src/main/java/org/apache/rocketmq/proxy/common/PopReceiptHandleGroupSummary.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.common;
+
+/**
+ * Aggregated summary statistics for POP receipt handles in a consumer group.
+ * <p>
+ * Provides a high-level view of the POP consumption state for a group,
+ * useful for quickly identifying groups with excessive unacked messages,
+ * high renewal rates, or expired handles.
+ */
+public class PopReceiptHandleGroupSummary {
+    private final String group;
+    private final int totalHandles;
+    private final int totalMessages;
+    private final long totalRenewTimes;
+    private final long totalRenewRetryTimes;
+    private final int expiredHandles;
+
+    public PopReceiptHandleGroupSummary(String group, int totalHandles, int totalMessages,
+        long totalRenewTimes, long totalRenewRetryTimes, int expiredHandles) {
+        this.group = group;
+        this.totalHandles = totalHandles;
+        this.totalMessages = totalMessages;
+        this.totalRenewTimes = totalRenewTimes;
+        this.totalRenewRetryTimes = totalRenewRetryTimes;
+        this.expiredHandles = expiredHandles;
+    }
+
+    public String getGroup() { return group; }
+    public int getTotalHandles() { return totalHandles; }
+    public int getTotalMessages() { return totalMessages; }
+    public long getTotalRenewTimes() { return totalRenewTimes; }
+    public long getTotalRenewRetryTimes() { return totalRenewRetryTimes; }
+    public int getExpiredHandles() { return expiredHandles; }
+}