This is an automated email from the ASF dual-hosted git repository.
RongtongJin 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 bf5d874770 [ISSUE #11041] Fix off-by-one in paginated metadata sync
dropping the last entry (#11042)
bf5d874770 is described below
commit bf5d874770f709e2f42c008756cbcd1cb2ed0912
Author: 烤化の初雪 <[email protected]>
AuthorDate: Tue Oct 27 11:08:42 2026 +0800
[ISSUE #11041] Fix off-by-one in paginated metadata sync dropping the last
entry (#11042)
The split-metadata pagination loops in MQClientAPIImpl and
BrokerOuterAPI terminated on seq >= totalNum - 1 while the broker pages
[seq, seq + maxNum) without overlap. For any entry count equal to one
over a page boundary (N % pageSize == 1) the last topic/subscription
group was silently never requested. SlaveSynchronize#syncTopicConfig
uses the same loop, so slaves permanently missed the newest topic.
Terminate on seq >= totalNum instead.
Co-authored-by: root <[email protected]>
---
.../apache/rocketmq/broker/out/BrokerOuterAPI.java | 4 +-
.../rocketmq/client/impl/MQClientAPIImpl.java | 4 +-
.../rocketmq/client/impl/MQClientAPIImplTest.java | 96 ++++++++++++++++++++++
3 files changed, 100 insertions(+), 4 deletions(-)
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
index ba4ba2ccf9..fb44e8e223 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
@@ -831,7 +831,7 @@ public class BrokerOuterAPI {
continue;
}
- if (topicSeq >= totalTopicNum - 1) {
+ if (topicSeq >= totalTopicNum) {
LOGGER.info("get all topic config, totalTopicNum: {}",
totalTopicNum);
break;
}
@@ -981,7 +981,7 @@ public class BrokerOuterAPI {
continue;
}
- if (groupSeq >= totalGroupNum - 1) {
+ if (groupSeq >= totalGroupNum) {
LOGGER.info("get all subscription group config,
totalGroupNum: {}", totalGroupNum);
break;
}
diff --git
a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
index 6999859491..d921bc554f 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
@@ -3030,7 +3030,7 @@ public class MQClientAPIImpl implements
NameServerUpdateCallback, StartAndShutdo
continue;
}
- if (groupSeq >= totalGroupNum - 1) {
+ if (groupSeq >= totalGroupNum) {
log.info("get all subscription group config,
totalGroupNum: {}", totalGroupNum);
break;
}
@@ -3122,7 +3122,7 @@ public class MQClientAPIImpl implements
NameServerUpdateCallback, StartAndShutdo
continue;
}
- if (topicSeq >= totalTopicNum - 1) {
+ if (topicSeq >= totalTopicNum) {
log.info("get all topic config, totalTopicNum: {}",
totalTopicNum);
break;
}
diff --git
a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
index 096853c6e6..1439ec4e96 100644
---
a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
+++
b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
@@ -65,6 +65,7 @@ import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
import org.apache.rocketmq.remoting.netty.ResponseFuture;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
@@ -110,6 +111,8 @@ import
org.apache.rocketmq.remoting.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
import
org.apache.rocketmq.remoting.protocol.header.GetConsumerListByGroupResponseBody;
import
org.apache.rocketmq.remoting.protocol.header.GetConsumerListByGroupResponseHeader;
+import
org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupResponseHeader;
+import
org.apache.rocketmq.remoting.protocol.header.GetAllTopicConfigResponseHeader;
import
org.apache.rocketmq.remoting.protocol.header.GetEarliestMsgStoretimeResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.GetMaxOffsetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.GetMinOffsetResponseHeader;
@@ -156,6 +159,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
import java.io.UnsupportedEncodingException;
+import java.nio.ByteBuffer;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.util.ArrayList;
@@ -1740,6 +1744,88 @@ public class MQClientAPIImplTest {
assertEquals(0, actual.getDataVersion().getStateVersion());
}
+ @Test
+ public void
testGetAllTopicConfigFetchesLastEntryWhenTotalIsOneOverPageSize()
+ throws Exception {
+ final int pageSize = 100;
+ final int total = pageSize + 1;
+ ClientConfig pagedClientConfig = new ClientConfig();
+ pagedClientConfig.setMaxPageSizeInGetMetadata(pageSize);
+ MQClientAPIImpl pagedApi = new MQClientAPIImpl(new
NettyClientConfig(), null, null, pagedClientConfig);
+ Field field = MQClientAPIImpl.class.getDeclaredField("remotingClient");
+ field.setAccessible(true);
+ field.set(pagedApi, remotingClient);
+
+ DataVersion dataVersion = new DataVersion();
+ TopicConfigSerializeWrapper page1 = new TopicConfigSerializeWrapper();
+ for (int i = 0; i < pageSize; i++) {
+ page1.getTopicConfigTable().put("topic-" + i, new
TopicConfig("topic-" + i));
+ }
+ page1.setDataVersion(dataVersion);
+ TopicConfigSerializeWrapper page2 = new TopicConfigSerializeWrapper();
+ page2.getTopicConfigTable().put("topic-" + pageSize, new
TopicConfig("topic-" + pageSize));
+ page2.setDataVersion(dataVersion);
+
+ GetAllTopicConfigResponseHeader header1 = new
GetAllTopicConfigResponseHeader();
+ header1.setTotalTopicNum(total);
+ RemotingCommand resp1 =
wireResponse(RemotingCommand.createResponseCommandWithHeader(ResponseCode.SUCCESS,
header1));
+ resp1.setBody(RemotingSerializable.encode(page1));
+ GetAllTopicConfigResponseHeader header2 = new
GetAllTopicConfigResponseHeader();
+ header2.setTotalTopicNum(total);
+ RemotingCommand resp2 =
wireResponse(RemotingCommand.createResponseCommandWithHeader(ResponseCode.SUCCESS,
header2));
+ resp2.setBody(RemotingSerializable.encode(page2));
+ when(remotingClient.invokeSync(anyString(),
any(RemotingCommand.class), anyLong()))
+ .thenReturn(resp1, resp2);
+
+ TopicConfigSerializeWrapper actual =
pagedApi.getAllTopicConfig(defaultBrokerAddr, defaultTimeout);
+
+ assertEquals(total, actual.getTopicConfigTable().size());
+ assertEquals("topic-" + pageSize,
actual.getTopicConfigTable().get("topic-" + pageSize).getTopicName());
+ }
+
+ @Test
+ public void
testGetAllSubscriptionGroupFetchesLastEntryWhenTotalIsOneOverPageSize()
+ throws Exception {
+ final int pageSize = 100;
+ final int total = pageSize + 1;
+ ClientConfig pagedClientConfig = new ClientConfig();
+ pagedClientConfig.setMaxPageSizeInGetMetadata(pageSize);
+ MQClientAPIImpl pagedApi = new MQClientAPIImpl(new
NettyClientConfig(), null, null, pagedClientConfig);
+ Field field = MQClientAPIImpl.class.getDeclaredField("remotingClient");
+ field.setAccessible(true);
+ field.set(pagedApi, remotingClient);
+
+ DataVersion dataVersion = new DataVersion();
+ SubscriptionGroupWrapper page1 = new SubscriptionGroupWrapper();
+ for (int i = 0; i < pageSize; i++) {
+ SubscriptionGroupConfig config = new SubscriptionGroupConfig();
+ config.setGroupName("group-" + i);
+ page1.getSubscriptionGroupTable().put("group-" + i, config);
+ }
+ page1.setDataVersion(dataVersion);
+ SubscriptionGroupWrapper page2 = new SubscriptionGroupWrapper();
+ SubscriptionGroupConfig lastConfig = new SubscriptionGroupConfig();
+ lastConfig.setGroupName("group-" + pageSize);
+ page2.getSubscriptionGroupTable().put("group-" + pageSize, lastConfig);
+ page2.setDataVersion(dataVersion);
+
+ GetAllSubscriptionGroupResponseHeader header1 = new
GetAllSubscriptionGroupResponseHeader();
+ header1.setTotalGroupNum(total);
+ RemotingCommand resp1 =
wireResponse(RemotingCommand.createResponseCommandWithHeader(ResponseCode.SUCCESS,
header1));
+ resp1.setBody(RemotingSerializable.encode(page1));
+ GetAllSubscriptionGroupResponseHeader header2 = new
GetAllSubscriptionGroupResponseHeader();
+ header2.setTotalGroupNum(total);
+ RemotingCommand resp2 =
wireResponse(RemotingCommand.createResponseCommandWithHeader(ResponseCode.SUCCESS,
header2));
+ resp2.setBody(RemotingSerializable.encode(page2));
+ when(remotingClient.invokeSync(anyString(),
any(RemotingCommand.class), anyLong()))
+ .thenReturn(resp1, resp2);
+
+ SubscriptionGroupWrapper actual =
pagedApi.getAllSubscriptionGroup(defaultBrokerAddr, defaultTimeout);
+
+ assertEquals(total, actual.getSubscriptionGroupTable().size());
+ assertNotNull(actual.getSubscriptionGroupTable().get("group-" +
pageSize));
+ }
+
@Test
public void testUpdateNameServerConfig() throws RemotingException,
InterruptedException, MQClientException, UnsupportedEncodingException {
mockInvokeSync();
@@ -2138,6 +2224,16 @@ public class MQClientAPIImplTest {
when(response.getBody()).thenReturn(RemotingSerializable.encode(responseBody));
}
+ /**
+ * Round-trips the response through encode/decode so response headers are
carried as extFields,
+ * mirroring the wire format the client actually parses.
+ */
+ private static RemotingCommand wireResponse(RemotingCommand response)
throws RemotingCommandException {
+ ByteBuffer frame = response.encode();
+ frame.position(4); // strip the total length the transport decoder
consumes
+ return RemotingCommand.decode(frame.slice());
+ }
+
private void mockInvokeSync() throws RemotingConnectException,
RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
when(response.getCode()).thenReturn(ResponseCode.SUCCESS);
when(response.getVersion()).thenReturn(1);
