opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #4979] Fix flaky integration test NormalMessageSendAndRecvIT (#4980)

Browse Source 
* Fix netty warning; Fix flaky integration test NormalMessageSendAndRecvIT; Fix naming typo

* Change log level

* Use awaitability to make everyone happy

* Do not submit tasks when executor has been shut down
This commit is contained in:
Zhanhui Li
2022-09-05 14:27:01 +08:00
committed by GitHub
parent 8357cff101
commit 35ac588fb5
6 changed files with 58 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java
+10 -6
View File
@@ -198,12 +198,11 @@ public class PopReviveService extends ServiceThread {
break;
case NO_MATCHED_MESSAGE:
pullStatus = PullStatus.NO_MATCHED_MSG;
POP_LOGGER.warn("no matched message. GetMessageStatus={}, topic={}, groupId={}, requestOffset={}",
POP_LOGGER.debug("no matched message. GetMessageStatus={}, topic={}, groupId={}, requestOffset={}",
getMessageResult.getStatus(), topic, group, offset);
break;
case NO_MESSAGE_IN_QUEUE:
pullStatus = PullStatus.NO_NEW_MSG;
POP_LOGGER.warn("no new message. GetMessageStatus={}, topic={}, groupId={}, requestOffset={}",
POP_LOGGER.debug("no new message. GetMessageStatus={}, topic={}, groupId={}, requestOffset={}",
getMessageResult.getStatus(), topic, group, offset);
break;
case MESSAGE_WAS_REMOVING:
@@ -225,7 +224,11 @@ public class PopReviveService extends ServiceThread {
getMessageResult.getMaxOffset(), foundList);
} else {
POP_LOGGER.error("get message from store return null. topic={}, groupId={}, requestOffset={}", topic, group, offset);
long maxQueueOffset = brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
if (maxQueueOffset > offset) {
POP_LOGGER.error("get message from store return null. topic={}, groupId={}, requestOffset={}, maxQueueOffset={}",
topic, group, offset, maxQueueOffset);
}
return null;
}
}
@@ -289,11 +292,12 @@ public class PopReviveService extends ServiceThread {
break;
}
noMsgCount++;
// Fixme: why sleep is useful here?
try {
Thread.sleep(100);
} catch (Throwable e) {
} catch (Throwable ignore) {
}
if (noMsgCount * 100 > 4 * PopAckConstants.SECOND) {
if (noMsgCount * 100L > 4 * PopAckConstants.SECOND) {
break;
} else {
continue;
namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BatchUnRegisterService.java → namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BatchUnregistrationService.java
+13 -13
View File
@@ -29,17 +29,17 @@ import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
/**
* BatchUnRegisterProcessor provides a mechanism to unregister broker in batch manner, which can speed up broker offline
* BatchUnregistrationService provides a mechanism to unregister brokers in batch manner, which speeds up broker-offline
* process.
*/
public class BatchUnRegisterService extends ServiceThread {
public class BatchUnregistrationService extends ServiceThread {
private final RouteInfoManager routeInfoManager;
private BlockingQueue<UnRegisterBrokerRequestHeader> unRegisterQueue;
private BlockingQueue<UnRegisterBrokerRequestHeader> unregistrationQueue;
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
public BatchUnRegisterService(RouteInfoManager routeInfoManager, NamesrvConfig namesrvConfig) {
public BatchUnregistrationService(RouteInfoManager routeInfoManager, NamesrvConfig namesrvConfig) {
this.routeInfoManager = routeInfoManager;
this.unRegisterQueue = new LinkedBlockingQueue<>(namesrvConfig.getUnRegisterBrokerQueueCapacity());
this.unregistrationQueue = new LinkedBlockingQueue<>(namesrvConfig.getUnRegisterBrokerQueueCapacity());
}
/**
@@ -49,26 +49,26 @@ public class BatchUnRegisterService extends ServiceThread {
* @return {@code true} if the request was added to this queue, else {@code false}
*/
public boolean submit(UnRegisterBrokerRequestHeader unRegisterRequest) {
return unRegisterQueue.offer(unRegisterRequest);
return unregistrationQueue.offer(unRegisterRequest);
}
@Override
public String getServiceName() {
return BatchUnRegisterService.class.getName();
return BatchUnregistrationService.class.getName();
}
@Override
public void run() {
while (!this.isStopped()) {
try {
final UnRegisterBrokerRequestHeader request = unRegisterQueue.take();
Set<UnRegisterBrokerRequestHeader> unRegisterRequests = new HashSet<>();
unRegisterQueue.drainTo(unRegisterRequests);
final UnRegisterBrokerRequestHeader request = unregistrationQueue.take();
Set<UnRegisterBrokerRequestHeader> unregistrationRequests = new HashSet<>();
unregistrationQueue.drainTo(unregistrationRequests);
// Add polled request
unRegisterRequests.add(request);
unregistrationRequests.add(request);
this.routeInfoManager.unRegisterBroker(unRegisterRequests);
this.routeInfoManager.unRegisterBroker(unregistrationRequests);
} catch (Throwable e) {
log.error("Handle unregister broker request failed", e);
}
@@ -77,6 +77,6 @@ public class BatchUnRegisterService extends ServiceThread {
// For test only
int queueLength() {
return this.unRegisterQueue.size();
return this.unregistrationQueue.size();
}
}
namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
+2 -2
View File
@@ -76,7 +76,7 @@ public class RouteInfoManager {
private final Map<BrokerAddrInfo/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
private final Map<String/* topic */, Map<String/*brokerName*/, TopicQueueMappingInfo>> topicQueueMappingInfoTable;
private final BatchUnRegisterService unRegisterService;
private final BatchUnregistrationService unRegisterService;
private final NamesrvController namesrvController;
private final NamesrvConfig namesrvConfig;
@@ -88,7 +88,7 @@ public class RouteInfoManager {
this.brokerLiveTable = new ConcurrentHashMap<>(256);
this.filterServerTable = new ConcurrentHashMap<>(256);
this.topicQueueMappingInfoTable = new ConcurrentHashMap<>(1024);
this.unRegisterService = new BatchUnRegisterService(this, namesrvConfig);
this.unRegisterService = new BatchUnregistrationService(this, namesrvConfig);
this.namesrvConfig = namesrvConfig;
this.namesrvController = namesrvController;
}
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
+1 -1
View File
@@ -318,7 +318,7 @@ public abstract class NettyRemotingAbstract {
private void executeInvokeCallback(final ResponseFuture responseFuture) {
boolean runInThisThread = false;
ExecutorService executor = this.getCallbackExecutor();
if (executor != null) {
if (executor != null && !executor.isShutdown()) {
try {
executor.submit(() -> {
try {
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
+1 -1
View File
@@ -219,7 +219,7 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
.channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.option(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.SO_KEEPALIVE, false)
.childOption(ChannelOption.SO_KEEPALIVE, false)
.childOption(ChannelOption.TCP_NODELAY, true)
.localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
.childHandler(new ChannelInitializer<SocketChannel>() {
test/src/test/java/org/apache/rocketmq/test/smoke/NormalMessageSendAndRecvIT.java
+31 -11
View File
@@ -17,8 +17,10 @@
package org.apache.rocketmq.test.smoke;
import java.time.Duration;
import java.util.List;
import org.apache.log4j.Logger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.message.MessageClientExt;
import org.apache.rocketmq.common.message.MessageConst;
@@ -33,11 +35,14 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.awaitility.Awaitility;
import static com.google.common.truth.Truth.assertThat;
public class NormalMessageSendAndRecvIT extends BaseConf {
private static Logger logger = Logger.getLogger(NormalMessageSendAndRecvIT.class);
private static final Logger logger = LoggerFactory.getLogger(NormalMessageSendAndRecvIT.class);
private RMQNormalConsumer consumer = null;
private RMQNormalProducer producer = null;
private String topic = null;
@@ -57,17 +62,32 @@ public class NormalMessageSendAndRecvIT extends BaseConf {
@After
public void tearDown() {
super.shutdown();
BaseConf.shutdown();
}
@Test
public void testSynSendMessage() throws Exception {
AtomicReference<List<MessageQueue>> messageQueueList = new AtomicReference<>();
AtomicReference<ConsumeStats> consumeStats = new AtomicReference<>();
Awaitility.await().atMost(Duration.ofSeconds(120))
.until(() -> {
try {
consumeStats.set(defaultMQAdminExt.examineConsumeStats(group));
messageQueueList.set(producer.getProducer().fetchPublishMessageQueues(topic));
return !messageQueueList.get().isEmpty() && null != consumeStats.get()
&& consumeStats.get().getOffsetTable().keySet().containsAll(messageQueueList.get());
} catch (MQClientException e) {
logger.debug("Exception raised while checking producer and consumer are started", e);
}
return false;
});
int msgSize = 10;
List<MessageQueue> messageQueueList = producer.getProducer().fetchPublishMessageQueues(topic);
for (MessageQueue messageQueue: messageQueueList) {
for (MessageQueue messageQueue : messageQueueList.get()) {
producer.send(msgSize, messageQueue);
}
Assert.assertEquals("Not all sent succeeded", msgSize * messageQueueList.size(), producer.getAllUndupMsgBody().size());
Assert.assertEquals("Not all sent succeeded", msgSize * messageQueueList.get().size(),
producer.getAllUndupMsgBody().size());
consumer.getListener().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
assertThat(VerifyUtils.getFilterdMessage(producer.getAllMsgBody(),
consumer.getListener().getAllMsgBody()))
@@ -79,12 +99,12 @@ public class NormalMessageSendAndRecvIT extends BaseConf {
}
//shutdown to persist the offset
consumer.getConsumer().shutdown();
ConsumeStats consumeStats = defaultMQAdminExt.examineConsumeStats(group);
consumeStats.set(defaultMQAdminExt.examineConsumeStats(group));
//+1 for the retry topic
for (MessageQueue messageQueue: messageQueueList) {
Assert.assertTrue(consumeStats.getOffsetTable().containsKey(messageQueue));
Assert.assertEquals(msgSize, consumeStats.getOffsetTable().get(messageQueue).getConsumerOffset());
Assert.assertEquals(msgSize, consumeStats.getOffsetTable().get(messageQueue).getBrokerOffset());
for (MessageQueue messageQueue : messageQueueList.get()) {
Assert.assertTrue(consumeStats.get().getOffsetTable().containsKey(messageQueue));
Assert.assertEquals(msgSize, consumeStats.get().getOffsetTable().get(messageQueue).getConsumerOffset());
Assert.assertEquals(msgSize, consumeStats.get().getOffsetTable().get(messageQueue).getBrokerOffset());
}
}