From 80bc36cf304e5e938ff1dbe6c20bcf7664daf2d9 Mon Sep 17 00:00:00 2001 From: hill007299 Date: Tue, 9 Mar 2021 11:22:37 +0800 Subject: [PATCH] [RIP-19] Pop Consuming (integration test) --- .../test/base/IntegrationTestBase.java | 12 +-- .../client/consumer/pop/PopSubCheckIT.java | 89 +++++++++++++++++++ .../smoke/NormalMessageSendAndRecvIT.java | 6 ++ 3 files changed, 101 insertions(+), 6 deletions(-) create mode 100644 test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/PopSubCheckIT.java diff --git a/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java b/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java index c484e87c11..a32da0320f 100644 --- a/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java +++ b/test/src/test/java/org/apache/rocketmq/test/base/IntegrationTestBase.java @@ -17,6 +17,7 @@ package org.apache.rocketmq.test.base; +import com.google.common.truth.Truth; import java.io.File; import java.util.ArrayList; import java.util.List; @@ -26,16 +27,15 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.common.namesrv.NamesrvConfig; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; -import org.apache.rocketmq.common.namesrv.NamesrvConfig; import org.apache.rocketmq.namesrv.NamesrvController; import org.apache.rocketmq.remoting.netty.NettyClientConfig; import org.apache.rocketmq.remoting.netty.NettyServerConfig; import org.apache.rocketmq.store.config.MessageStoreConfig; import org.apache.rocketmq.test.util.MQAdmin; import org.apache.rocketmq.test.util.TestUtils; -import org.junit.Assert; public class IntegrationTestBase { public static InternalLogger logger = InternalLoggerFactory.getLogger(IntegrationTestBase.class); @@ -113,7 +113,7 @@ public class IntegrationTestBase { nameServerNettyServerConfig.setListenPort(nextPort()); NamesrvController namesrvController = new NamesrvController(namesrvConfig, nameServerNettyServerConfig); try { - Assert.assertTrue(namesrvController.initialize()); + Truth.assertThat(namesrvController.initialize()).isTrue(); logger.info("Name Server Start:{}", nameServerNettyServerConfig.getListenPort()); namesrvController.start(); } catch (Exception e) { @@ -149,7 +149,7 @@ public class IntegrationTestBase { storeConfig.setHaListenPort(nextPort()); BrokerController brokerController = new BrokerController(brokerConfig, nettyServerConfig, nettyClientConfig, storeConfig); try { - Assert.assertTrue(brokerController.initialize()); + Truth.assertThat(brokerController.initialize()).isTrue(); logger.info("Broker Start name:{} addr:{}", brokerConfig.getBrokerName(), brokerController.getBrokerAddr()); brokerController.start(); } catch (Throwable t) { @@ -169,8 +169,8 @@ public class IntegrationTestBase { if (createResult) { break; } else if (System.currentTimeMillis() - startTime > topicCreateTime) { - Assert.fail(String.format("topic[%s] is created failed after:%d ms", topic, - System.currentTimeMillis() - startTime)); + Truth.assertWithMessage(String.format("topic[%s] is created failed after:%d ms", topic, + System.currentTimeMillis() - startTime)).fail(); break; } else { TestUtils.waitForMoment(500); diff --git a/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/PopSubCheckIT.java b/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/PopSubCheckIT.java new file mode 100644 index 0000000000..74a9e04ce2 --- /dev/null +++ b/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/PopSubCheckIT.java @@ -0,0 +1,89 @@ +/* + * 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.test.client.consumer.pop; + +import org.apache.rocketmq.common.message.MessageClientExt; +import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageRequestMode; +import org.apache.rocketmq.logging.inner.Logger; +import org.apache.rocketmq.test.base.BaseConf; +import org.apache.rocketmq.test.client.rmq.RMQNormalConsumer; +import org.apache.rocketmq.test.client.rmq.RMQNormalProducer; +import org.apache.rocketmq.test.listener.rmq.concurrent.RMQNormalListener; +import org.apache.rocketmq.test.util.RandomUtil; +import org.apache.rocketmq.test.util.VerifyUtils; +import org.apache.rocketmq.tools.admin.DefaultMQAdminExt; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import static com.google.common.truth.Truth.assertThat; + +public class PopSubCheckIT extends BaseConf { + private static Logger logger = Logger.getLogger(PopSubCheckIT.class); + private String group; + + private DefaultMQAdminExt defaultMQAdminExt; + + @Before + public void setUp() throws Exception { + group = initConsumerGroup(); + + defaultMQAdminExt = new DefaultMQAdminExt(); + defaultMQAdminExt.setInstanceName(RandomUtil.getStringByUUID()); + defaultMQAdminExt.start(); + } + + @After + public void tearDown() { + defaultMQAdminExt.shutdown(); + super.shutdown(); + } + + @Test + public void testNormalPopAck() throws Exception { + String topic = initTopic(); + logger.info(String.format("use topic: %s; group: %s !", topic, group)); + + RMQNormalProducer producer = getProducer(nsAddr, topic); + producer.getProducer().setCompressMsgBodyOverHowmuch(Integer.MAX_VALUE); + + RMQNormalConsumer consumer = getConsumer(nsAddr, group, topic, "*", new RMQNormalListener()); + + for (String brokerAddr : new String[]{brokerController1.getBrokerAddr(), brokerController2.getBrokerAddr()}) { + defaultMQAdminExt.setMessageRequestMode(brokerAddr, topic, group, MessageRequestMode.POP, 8, 60_000); + } + + int msgNum = 1; + producer.send(msgNum); + Assert.assertEquals("Not all sent succeeded", msgNum, producer.getAllUndupMsgBody().size()); + logger.info(producer.getFirstMsg()); + + consumer.getListener().waitForMessageConsume(msgNum, 30_000); + assertThat(VerifyUtils.getFilterdMessage(producer.getAllMsgBody(), consumer.getListener().getAllMsgBody())) + .containsExactlyElementsIn(producer.getAllMsgBody()); + for (Object o : consumer.getListener().getAllOriginMsg()) { + MessageClientExt msg = (MessageClientExt) o; + assertThat(msg.getProperty(MessageConst.PROPERTY_POP_CK)).isNotEmpty(); + } + + consumer.getListener().waitForMessageConsume(msgNum, 3_000 * 9); + assertThat(consumer.getListener().getAllOriginMsg().size()).isEqualTo(msgNum); + } +} diff --git a/test/src/test/java/org/apache/rocketmq/test/smoke/NormalMessageSendAndRecvIT.java b/test/src/test/java/org/apache/rocketmq/test/smoke/NormalMessageSendAndRecvIT.java index c7886554a7..81dc864d17 100644 --- a/test/src/test/java/org/apache/rocketmq/test/smoke/NormalMessageSendAndRecvIT.java +++ b/test/src/test/java/org/apache/rocketmq/test/smoke/NormalMessageSendAndRecvIT.java @@ -18,6 +18,8 @@ package org.apache.rocketmq.test.smoke; import org.apache.log4j.Logger; +import org.apache.rocketmq.common.message.MessageClientExt; +import org.apache.rocketmq.common.message.MessageConst; import org.apache.rocketmq.test.base.BaseConf; import org.apache.rocketmq.test.client.rmq.RMQNormalConsumer; import org.apache.rocketmq.test.client.rmq.RMQNormalProducer; @@ -58,5 +60,9 @@ public class NormalMessageSendAndRecvIT extends BaseConf { assertThat(VerifyUtils.getFilterdMessage(producer.getAllMsgBody(), consumer.getListener().getAllMsgBody())) .containsExactlyElementsIn(producer.getAllMsgBody()); + for (Object o : consumer.getListener().getAllOriginMsg()) { + MessageClientExt msg = (MessageClientExt) o; + assertThat(msg.getProperty(MessageConst.PROPERTY_POP_CK)).isNull(); + } } }