opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-21 13:49:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix test for consumer offset

Browse Source
This commit is contained in:
dongeforever
2022-02-08 10:38:41 +08:00
committed by RongtongJin
parent b3f9fbd3a7
commit a767cc1424
2 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+6 -4
View File
@@ -1176,6 +1176,8 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
continue;
}
TopicQueueMappingDetail mappingDetail = this.brokerController.getTopicQueueMappingManager().getTopicQueueMapping(topic);
{
SubscriptionData findSubscriptionData =
this.brokerController.getConsumerManager().findSubscriptionData(requestHeader.getConsumerGroup(), topic);
@@ -1206,14 +1208,14 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
// the consumerOffset cannot be zero for static topic because of the "double read check" strategy
// just remain the logic for dynamic topic
// maybe we should remove it in the future
if (consumerOffset < 0)
consumerOffset = 0;
if (mappingDetail == null) {
if (consumerOffset < 0)
consumerOffset = 0;
}
offsetWrapper.setBrokerOffset(brokerOffset);
offsetWrapper.setConsumerOffset(consumerOffset);
// the consumeOffset is not in this broker for static topic
// and may get the wrong result
long timeOffset = consumerOffset - 1;
if (timeOffset >= 0) {
long lastTimestamp = this.brokerController.getMessageStore().getMessageStoreTimeStamp(topic, i, timeOffset);
test/src/test/java/org/apache/rocketmq/test/statictopic/StaticTopicIT.java
+3
View File
@@ -292,6 +292,7 @@ public class StaticTopicIT extends BaseConf {
String group = initConsumerGroup();
RMQNormalProducer producer = getProducer(nsAddr, topic);
RMQNormalConsumer consumer = getConsumer(nsAddr, group, topic, "*", new RMQNormalListener());
long start = System.currentTimeMillis();
int queueNum = 10;
int msgEachQueue = 100;
@@ -314,6 +315,7 @@ public class StaticTopicIT extends BaseConf {
Assert.assertNotNull(wrapper);
Assert.assertEquals(msgEachQueue, wrapper.getBrokerOffset());
Assert.assertEquals(msgEachQueue, wrapper.getConsumerOffset());
Assert.assertTrue(wrapper.getLastTimestamp() > start);
}
List<String> brokers = ImmutableList.of(broker2Name, broker3Name, broker1Name);
@@ -332,6 +334,7 @@ public class StaticTopicIT extends BaseConf {
Assert.assertNotNull(wrapper);
Assert.assertEquals(msgEachQueue + brokers.size() * TopicQueueMappingUtils.DEFAULT_BLOCK_SEQ_SIZE, wrapper.getBrokerOffset());
Assert.assertEquals(msgEachQueue, wrapper.getConsumerOffset());
Assert.assertTrue(wrapper.getLastTimestamp() > start);
}
consumer = getConsumer(nsAddr, group, topic, "*", new RMQNormalListener());
consumeMessagesAndCheck(producer, consumer, topic, queueNum, msgEachQueue, 1, brokers.size());