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

add tests for computePullFromWhereWithException (#3844)

Browse Source
This commit is contained in:
yuz10
2022-02-13 10:02:40 +08:00
committed by GitHub
parent 8b78f6114f
commit 48e8c1955b
2 changed files with 171 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/test/java/org/apache/rocketmq/client/impl/consumer/RebalanceLitePullImplTest.java
+100
View File
@@ -0,0 +1,100 @@
/*
* 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.client.impl.consumer;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.store.OffsetStore;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.MQAdminImpl;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class RebalanceLitePullImplTest {
private MessageQueue mq = new MessageQueue("topic1", "broker1", 0);
private MessageQueue retryMq = new MessageQueue(MixAll.RETRY_GROUP_TOPIC_PREFIX + "group", "broker1", 0);
private DefaultLitePullConsumerImpl consumerImpl = mock(DefaultLitePullConsumerImpl.class);
private RebalanceLitePullImpl rebalanceImpl = new RebalanceLitePullImpl(consumerImpl);
private OffsetStore offsetStore = mock(OffsetStore.class);
private DefaultLitePullConsumer consumer = new DefaultLitePullConsumer();
private MQClientInstance client = mock(MQClientInstance.class);
private MQAdminImpl admin = mock(MQAdminImpl.class);
public RebalanceLitePullImplTest() {
when(consumerImpl.getDefaultLitePullConsumer()).thenReturn(consumer);
when(consumerImpl.getOffsetStore()).thenReturn(offsetStore);
rebalanceImpl.setmQClientFactory(client);
when(client.getMQAdminImpl()).thenReturn(admin);
}
@Test
public void testComputePullFromWhereWithException_ne_minus1() throws MQClientException {
for (ConsumeFromWhere where : new ConsumeFromWhere[]{
ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET,
ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET,
ConsumeFromWhere.CONSUME_FROM_TIMESTAMP}) {
consumer.setConsumeFromWhere(where);
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(0L);
assertEquals(0, rebalanceImpl.computePullFromWhereWithException(mq));
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(-2L);
assertEquals(-1, rebalanceImpl.computePullFromWhereWithException(mq));
}
}
@Test
public void testComputePullFromWhereWithException_eq_minus1_last() throws MQClientException {
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(-1L);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
when(admin.maxOffset(any(MessageQueue.class))).thenReturn(12345L);
assertEquals(12345L, rebalanceImpl.computePullFromWhereWithException(mq));
assertEquals(0L, rebalanceImpl.computePullFromWhereWithException(retryMq));
}
@Test
public void testComputePullFromWhereWithException_eq_minus1_first() throws MQClientException {
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(-1L);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
assertEquals(0, rebalanceImpl.computePullFromWhereWithException(mq));
}
@Test
public void testComputePullFromWhereWithException_eq_minus1_timestamp() throws MQClientException {
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(-1L);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
when(admin.searchOffset(any(MessageQueue.class), anyLong())).thenReturn(12345L);
when(admin.maxOffset(any(MessageQueue.class))).thenReturn(23456L);
assertEquals(12345L, rebalanceImpl.computePullFromWhereWithException(mq));
assertEquals(23456L, rebalanceImpl.computePullFromWhereWithException(retryMq));
}
}
client/src/test/java/org/apache/rocketmq/client/impl/consumer/RebalancePushImplTest.java
+71 -3
View File
@@ -19,11 +19,16 @@ package org.apache.rocketmq.client.impl.consumer;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.consumer.store.OffsetStore;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.MQAdminImpl;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
@@ -36,9 +41,12 @@ import org.mockito.junit.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
@@ -47,10 +55,23 @@ public class RebalancePushImplTest {
private DefaultMQPushConsumerImpl defaultMQPushConsumer = new DefaultMQPushConsumerImpl(new DefaultMQPushConsumer("RebalancePushImplTest"), null);
@Mock
private MQClientInstance mqClientInstance;
@Mock
private OffsetStore offsetStore;
private OffsetStore offsetStore = mock(OffsetStore.class);
private String consumerGroup = "CID_RebalancePushImplTest";
private String topic = "TopicA";
private MessageQueue mq = new MessageQueue("topic1", "broker1", 0);
private MessageQueue retryMq = new MessageQueue(MixAll.RETRY_GROUP_TOPIC_PREFIX + "group", "broker1", 0);
private DefaultMQPushConsumerImpl consumerImpl = mock(DefaultMQPushConsumerImpl.class);
private RebalancePushImpl rebalanceImpl = new RebalancePushImpl(consumerImpl);
private DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
private MQClientInstance client = mock(MQClientInstance.class);
private MQAdminImpl admin = mock(MQAdminImpl.class);
public RebalancePushImplTest() {
when(consumerImpl.getDefaultMQPushConsumer()).thenReturn(consumer);
when(consumerImpl.getOffsetStore()).thenReturn(offsetStore);
rebalanceImpl.setmQClientFactory(client);
when(client.getMQAdminImpl()).thenReturn(admin);
}
@Test
public void testMessageQueueChanged_CountThreshold() {
@@ -160,4 +181,51 @@ public class RebalancePushImplTest {
assertThat(defaultMQPushConsumer.consumerRunningInfo().getProperties().get("pullThresholdSizeForTopic")).isEqualTo("1024");
assertThat(defaultMQPushConsumer.consumerRunningInfo().getProperties().get("pullThresholdForTopic")).isEqualTo("1024");
}
}
@Test
public void testComputePullFromWhereWithException_ne_minus1() throws MQClientException {
for (ConsumeFromWhere where : new ConsumeFromWhere[]{
ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET,
ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET,
ConsumeFromWhere.CONSUME_FROM_TIMESTAMP}) {
consumer.setConsumeFromWhere(where);
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(0L);
assertEquals(0, rebalanceImpl.computePullFromWhereWithException(mq));
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(-2L);
assertEquals(-1, rebalanceImpl.computePullFromWhereWithException(mq));
}
}
@Test
public void testComputePullFromWhereWithException_eq_minus1_last() throws MQClientException {
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(-1L);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
when(admin.maxOffset(any(MessageQueue.class))).thenReturn(12345L);
assertEquals(12345L, rebalanceImpl.computePullFromWhereWithException(mq));
assertEquals(0L, rebalanceImpl.computePullFromWhereWithException(retryMq));
}
@Test
public void testComputePullFromWhereWithException_eq_minus1_first() throws MQClientException {
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(-1L);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
assertEquals(0, rebalanceImpl.computePullFromWhereWithException(mq));
}
@Test
public void testComputePullFromWhereWithException_eq_minus1_timestamp() throws MQClientException {
when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(-1L);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
when(admin.searchOffset(any(MessageQueue.class), anyLong())).thenReturn(12345L);
when(admin.maxOffset(any(MessageQueue.class))).thenReturn(23456L);
assertEquals(12345L, rebalanceImpl.computePullFromWhereWithException(mq));
assertEquals(23456L, rebalanceImpl.computePullFromWhereWithException(retryMq));
}
}