Skip to content

Commit 2569b68

Browse files
yx9oyx9o
committed
Add test
1 parent 0e709d4 commit 2569b68

3 files changed

Lines changed: 89 additions & 6 deletions

File tree

‎broker/src/test/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessorTest.java‎

Lines changed: 78 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,18 +18,15 @@
1818

1919
import io.netty.channel.Channel;
2020
import io.netty.channel.ChannelHandlerContext;
21-
import java.lang.reflect.Field;
22-
import java.util.concurrent.CompletableFuture;
23-
import java.util.concurrent.ConcurrentHashMap;
2421
import org.apache.rocketmq.broker.BrokerController;
2522
import org.apache.rocketmq.broker.client.ClientChannelInfo;
2623
import org.apache.rocketmq.broker.client.net.Broker2Client;
2724
import org.apache.rocketmq.broker.failover.EscapeBridge;
2825
import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
29-
import org.apache.rocketmq.store.stats.BrokerStatsManager;
26+
import org.apache.rocketmq.broker.metrics.PopMetricsManager;
27+
import org.apache.rocketmq.broker.topic.TopicConfigManager;
3028
import org.apache.rocketmq.common.BrokerConfig;
3129
import org.apache.rocketmq.common.TopicConfig;
32-
import org.apache.rocketmq.broker.topic.TopicConfigManager;
3330
import org.apache.rocketmq.common.message.MessageConst;
3431
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
3532
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
@@ -41,24 +38,35 @@
4138
import org.apache.rocketmq.remoting.protocol.ResponseCode;
4239
import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeRequestHeader;
4340
import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
41+
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData;
4442
import org.apache.rocketmq.store.AppendMessageResult;
4543
import org.apache.rocketmq.store.AppendMessageStatus;
4644
import org.apache.rocketmq.store.DefaultMessageStore;
45+
import org.apache.rocketmq.store.MessageStore;
4746
import org.apache.rocketmq.store.PutMessageResult;
4847
import org.apache.rocketmq.store.PutMessageStatus;
4948
import org.apache.rocketmq.store.config.MessageStoreConfig;
5049
import org.apache.rocketmq.store.exception.ConsumeQueueException;
50+
import org.apache.rocketmq.store.stats.BrokerStatsManager;
5151
import org.junit.Before;
5252
import org.junit.Test;
5353
import org.junit.runner.RunWith;
5454
import org.mockito.Mock;
5555
import org.mockito.Spy;
5656
import org.mockito.junit.MockitoJUnitRunner;
5757

58+
import java.lang.reflect.Field;
59+
import java.util.concurrent.CompletableFuture;
60+
import java.util.concurrent.ConcurrentHashMap;
61+
62+
import static org.apache.rocketmq.broker.processor.PullMessageProcessorTest.createConsumerData;
5863
import static org.assertj.core.api.Assertions.assertThat;
64+
import static org.junit.Assert.assertEquals;
65+
import static org.junit.Assert.assertNotNull;
5966
import static org.mockito.ArgumentMatchers.any;
6067
import static org.mockito.ArgumentMatchers.anyInt;
6168
import static org.mockito.ArgumentMatchers.anyString;
69+
import static org.mockito.Mockito.doNothing;
6270
import static org.mockito.Mockito.mock;
6371
import static org.mockito.Mockito.when;
6472

@@ -118,6 +126,17 @@ public void init() throws IllegalAccessException, NoSuchFieldException {
118126
PopBufferMergeService popBufferMergeService = mock(PopBufferMergeService.class);
119127
when(brokerController.getPopMessageProcessor()).thenReturn(popMessageProcessor);
120128
when(popMessageProcessor.getPopBufferMergeService()).thenReturn(popBufferMergeService);
129+
130+
ConsumerData consumerData = createConsumerData(group, topic);
131+
clientInfo = new ClientChannelInfo(channel, "127.0.0.1", LanguageCode.JAVA, 0);
132+
brokerController.getConsumerManager().registerConsumer(
133+
consumerData.getGroupName(),
134+
clientInfo,
135+
consumerData.getConsumeType(),
136+
consumerData.getMessageModel(),
137+
consumerData.getConsumeFromWhere(),
138+
consumerData.getSubscriptionDataSet(),
139+
false);
121140

122141
clientInfo = new ClientChannelInfo(channel, "127.0.0.1", LanguageCode.JAVA, 0);
123142
changeInvisibleTimeProcessor = new ChangeInvisibleTimeProcessor(brokerController);
@@ -177,4 +196,58 @@ public void testProcessRequest_NoMessage() throws RemotingCommandException, Cons
177196
assertThat(responseToReturn.getCode()).isEqualTo(ResponseCode.NO_MESSAGE);
178197
assertThat(responseToReturn.getOpaque()).isEqualTo(request.getOpaque());
179198
}
199+
200+
@Test
201+
public void testProcessRequestAsync_JsonParsing() throws Exception {
202+
Channel mockChannel = mock(Channel.class);
203+
RemotingCommand mockRequest = mock(RemotingCommand.class);
204+
BrokerController mockBrokerController = mock(BrokerController.class);
205+
TopicConfigManager mockTopicConfigManager = mock(TopicConfigManager.class);
206+
MessageStore mockMessageStore = mock(MessageStore.class);
207+
BrokerConfig mockBrokerConfig = mock(BrokerConfig.class);
208+
BrokerStatsManager mockBrokerStatsManager = mock(BrokerStatsManager.class);
209+
PopMessageProcessor mockPopMessageProcessor = mock(PopMessageProcessor.class);
210+
PopBufferMergeService mockPopBufferMergeService = mock(PopBufferMergeService.class);
211+
BrokerMetricsManager brokerMetricsManager = mock(BrokerMetricsManager.class);
212+
PopMetricsManager popMetricsManager = mock(PopMetricsManager.class);
213+
214+
when(brokerMetricsManager.getPopMetricsManager()).thenReturn(popMetricsManager);
215+
when(mockBrokerController.getBrokerMetricsManager()).thenReturn(brokerMetricsManager);
216+
doNothing().when(popMetricsManager).incPopReviveCkPutCount(any(), any());
217+
when(brokerMetricsManager.getPopMetricsManager()).thenReturn(popMetricsManager);
218+
when(mockBrokerController.getTopicConfigManager()).thenReturn(mockTopicConfigManager);
219+
when(mockBrokerController.getMessageStore()).thenReturn(mockMessageStore);
220+
when(mockBrokerController.getBrokerConfig()).thenReturn(mockBrokerConfig);
221+
when(mockBrokerController.getBrokerStatsManager()).thenReturn(mockBrokerStatsManager);
222+
when(mockBrokerController.getPopMessageProcessor()).thenReturn(mockPopMessageProcessor);
223+
when(mockPopMessageProcessor.getPopBufferMergeService()).thenReturn(mockPopBufferMergeService);
224+
when(mockPopBufferMergeService.addAk(anyInt(), any())).thenReturn(false);
225+
when(mockBrokerController.getEscapeBridge()).thenReturn(escapeBridge);
226+
PutMessageResult mockPutMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, null, true);
227+
when(mockBrokerController.getEscapeBridge().asyncPutMessageToSpecificQueue(any()))
228+
.thenReturn(CompletableFuture.completedFuture(mockPutMessageResult));
229+
230+
TopicConfig topicConfig = new TopicConfig();
231+
topicConfig.setReadQueueNums(4);
232+
when(mockTopicConfigManager.selectTopicConfig(anyString())).thenReturn(topicConfig);
233+
when(mockMessageStore.getMinOffsetInQueue(anyString(), anyInt())).thenReturn(0L);
234+
when(mockMessageStore.getMaxOffsetInQueue(anyString(), anyInt())).thenReturn(10L);
235+
when(mockBrokerConfig.isPopConsumerKVServiceEnable()).thenReturn(false);
236+
237+
ChangeInvisibleTimeRequestHeader requestHeader = new ChangeInvisibleTimeRequestHeader();
238+
requestHeader.setTopic("TestTopic");
239+
requestHeader.setQueueId(1);
240+
requestHeader.setOffset(5L);
241+
requestHeader.setConsumerGroup("TestGroup");
242+
requestHeader.setExtraInfo("0 10000 10000 0 TestBroker 1");
243+
requestHeader.setInvisibleTime(60000L);
244+
when(mockRequest.decodeCommandCustomHeader(ChangeInvisibleTimeRequestHeader.class)).thenReturn(requestHeader);
245+
246+
ChangeInvisibleTimeProcessor processor = new ChangeInvisibleTimeProcessor(mockBrokerController);
247+
CompletableFuture<RemotingCommand> futureResponse = processor.processRequestAsync(mockChannel, mockRequest, true);
248+
249+
RemotingCommand response = futureResponse.get();
250+
assertNotNull(response);
251+
assertEquals(ResponseCode.SUCCESS, response.getCode());
252+
}
180253
}

‎broker/src/test/java/org/apache/rocketmq/broker/processor/PopBufferMergeServiceTest.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121
import org.apache.rocketmq.broker.BrokerController;
2222
import org.apache.rocketmq.broker.client.ConsumerManager;
2323
import org.apache.rocketmq.broker.failover.EscapeBridge;
24+
import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
25+
import org.apache.rocketmq.broker.metrics.PopMetricsManager;
2426
import org.apache.rocketmq.broker.schedule.ScheduleMessageService;
2527
import org.apache.rocketmq.broker.topic.TopicConfigManager;
2628
import org.apache.rocketmq.common.BrokerConfig;
@@ -49,6 +51,7 @@
4951
import static org.junit.Assert.assertEquals;
5052
import static org.junit.Assert.assertFalse;
5153
import static org.mockito.ArgumentMatchers.any;
54+
import static org.mockito.Mockito.doNothing;
5255
import static org.mockito.Mockito.mock;
5356
import static org.mockito.Mockito.times;
5457
import static org.mockito.Mockito.verify;
@@ -201,6 +204,13 @@ public void testPutAckToStore() throws Exception {
201204
EscapeBridge escapeBridge = mock(EscapeBridge.class);
202205
when(brokerController.getEscapeBridge()).thenReturn(escapeBridge);
203206
when(brokerController.getBrokerConfig().isAppendAckAsync()).thenReturn(false);
207+
BrokerMetricsManager brokerMetricsManager = mock(BrokerMetricsManager.class);
208+
PopMetricsManager popMetricsManager = mock(PopMetricsManager.class);
209+
210+
when(brokerMetricsManager.getPopMetricsManager()).thenReturn(popMetricsManager);
211+
when(brokerController.getBrokerMetricsManager()).thenReturn(brokerMetricsManager);
212+
doNothing().when(popMetricsManager).incPopReviveCkPutCount(any(), any());
213+
when(brokerMetricsManager.getPopMetricsManager()).thenReturn(popMetricsManager);
204214

205215
when(escapeBridge.putMessageToSpecificQueue(any())).thenAnswer(invocation -> {
206216
MessageExtBrokerInner capturedMessage = invocation.getArgument(0);

‎controller/src/test/java/org/apache/rocketmq/controller/impl/heartbeat/RaftBrokerHeartBeatManagerTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -151,7 +151,7 @@ public void testScanNotActiveBrokerSuccess() throws Exception {
151151
controller = mock(JRaftController.class);
152152
FieldUtils.writeDeclaredField(raftBrokerHeartBeatManager, "controller", controller, true);
153153
when(controller.isLeaderState()).thenReturn(true);
154-
FieldUtils.writeDeclaredField(raftBrokerHeartBeatManager, "firstReceivedHeartbeatTime", System.currentTimeMillis() + 10000L, true);
154+
FieldUtils.writeDeclaredField(raftBrokerHeartBeatManager, "firstReceivedHeartbeatTime", 1000, true);
155155

156156
List<BrokerIdentityInfo> inactiveBrokers = new ArrayList<>();
157157
BrokerIdentityInfo brokerInfo = new BrokerIdentityInfo("testCluster", "testBroker", 1L);

0 commit comments

Comments
 (0)