(rocketmq) branch develop updated: [ISSUE #10284] Fix consumerOffset deserialization error and add test (#10287)

[yuzhou]

This is an automated email from the ASF dual-hosted git repository.

yuz10 pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new bd903d56e1 [ISSUE #10284] Fix consumerOffset deserialization error and 
add test (#10287)
bd903d56e1 is described below

commit bd903d56e1da5ef3761d8f45f1169662751b0efb
Author: ChineseTony <taowan...@163.com>
AuthorDate: Tue May 12 09:18:48 2026 +0800

    [ISSUE #10284] Fix consumerOffset deserialization error and add test 
(#10287)
    
    Co-authored-by: wangtao_ <wangtao...@huawei.com>
---
 .../org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java  | 2 +-
 .../rocketmq/broker/processor/AdminBrokerProcessorTest.java       | 6 ++++++
 .../remoting/protocol/body/ConsumerOffsetSerializeWrapper.java    | 8 ++++++++
 3 files changed, 15 insertions(+), 1 deletion(-)

diff --git 
a/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
 
b/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
index e062ceca96..f9debf3857 100644
--- 
a/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
+++ 
b/broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOffsetManager.java
@@ -51,7 +51,7 @@ public class ConsumerOffsetManager extends ConfigManager {
     protected final ConcurrentMap<String, ConcurrentMap<Integer, Long>> 
resetOffsetTable =
         new ConcurrentHashMap<>(512);
 
-    private final ConcurrentMap<String/* topic@group */, 
ConcurrentMap<Integer, Long>> pullOffsetTable =
+    private final transient ConcurrentMap<String/* topic@group */, 
ConcurrentMap<Integer, Long>> pullOffsetTable =
         new ConcurrentHashMap<>(512);
 
     protected transient BrokerController brokerController;
diff --git 
a/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
 
b/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
index 656c783e1f..c342400d14 100644
--- 
a/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
+++ 
b/broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java
@@ -74,6 +74,7 @@ import 
org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 import org.apache.rocketmq.remoting.protocol.RequestCode;
 import org.apache.rocketmq.remoting.protocol.ResponseCode;
 import org.apache.rocketmq.remoting.protocol.body.AclInfo;
+import 
org.apache.rocketmq.remoting.protocol.body.ConsumerOffsetSerializeWrapper;
 import org.apache.rocketmq.remoting.protocol.body.CreateTopicListRequestBody;
 import org.apache.rocketmq.remoting.protocol.body.GroupList;
 import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
@@ -170,7 +171,9 @@ import java.util.concurrent.atomic.LongAdder;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
@@ -946,7 +949,10 @@ public class AdminBrokerProcessorTest {
         
when(consumerOffsetManager.encode()).thenReturn(JSON.toJSONString(consumerOffset));
         RemotingCommand request = 
RemotingCommand.createRequestCommand(RequestCode.GET_ALL_CONSUMER_OFFSET, null);
         RemotingCommand response = 
adminBrokerProcessor.processRequest(handlerContext, request);
+        ConsumerOffsetSerializeWrapper consumerOffsetSerializeWrapper = 
ConsumerOffsetSerializeWrapper.decode(response.getBody(), 
ConsumerOffsetSerializeWrapper.class);
         assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
+        assertFalse(new 
String(response.getBody()).contains("pullOffsetTable"));
+        
assertTrue(consumerOffsetSerializeWrapper.getPullOffsetTable().isEmpty());
     }
 
     @Test
diff --git 
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/ConsumerOffsetSerializeWrapper.java
 
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/ConsumerOffsetSerializeWrapper.java
index 407be4670e..1d5897e443 100644
--- 
a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/ConsumerOffsetSerializeWrapper.java
+++ 
b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/body/ConsumerOffsetSerializeWrapper.java
@@ -25,6 +25,10 @@ import 
org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 public class ConsumerOffsetSerializeWrapper extends RemotingSerializable {
     private ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, 
Long>> offsetTable =
         new ConcurrentHashMap<>(512);
+
+    private final ConcurrentMap<String/* topic@group */, 
ConcurrentMap<Integer, Long>> pullOffsetTable =
+            new ConcurrentHashMap<>(512);
+
     private DataVersion dataVersion;
 
     public ConcurrentMap<String, ConcurrentMap<Integer, Long>> 
getOffsetTable() {
@@ -35,6 +39,10 @@ public class ConsumerOffsetSerializeWrapper extends 
RemotingSerializable {
         this.offsetTable = offsetTable;
     }
 
+    public ConcurrentMap<String/* topic@group */, ConcurrentMap<Integer, 
Long>> getPullOffsetTable() {
+        return pullOffsetTable;
+    }
+
     public DataVersion getDataVersion() {
         return dataVersion;
     }