From 35ac588fb55ebf6ac0d72e7c7ad8272898d1df0f Mon Sep 17 00:00:00 2001 From: Zhanhui Li Date: Mon, 5 Sep 2022 14:27:01 +0800 Subject: [PATCH] [ISSUE #4979] Fix flaky integration test NormalMessageSendAndRecvIT (#4980) * 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 --- .../broker/processor/PopReviveService.java | 16 ++++--- ...e.java => BatchUnregistrationService.java} | 26 ++++++------ .../namesrv/routeinfo/RouteInfoManager.java | 4 +- .../remoting/netty/NettyRemotingAbstract.java | 2 +- .../remoting/netty/NettyRemotingServer.java | 2 +- .../smoke/NormalMessageSendAndRecvIT.java | 42 ++++++++++++++----- 6 files changed, 58 insertions(+), 34 deletions(-) rename namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/{BatchUnRegisterService.java => BatchUnregistrationService.java} (70%) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java index 1a6c52ec33..8708b48a6a 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopReviveService.java @@ -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; diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BatchUnRegisterService.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BatchUnregistrationService.java similarity index 70% rename from namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BatchUnRegisterService.java rename to namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BatchUnregistrationService.java index 9e84272cb5..5a4def3055 100644 --- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BatchUnRegisterService.java +++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BatchUnregistrationService.java @@ -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 unRegisterQueue; + private BlockingQueue 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 unRegisterRequests = new HashSet<>(); - unRegisterQueue.drainTo(unRegisterRequests); + final UnRegisterBrokerRequestHeader request = unregistrationQueue.take(); + Set 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(); } } diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java index 3facb3cf83..ace963ca9c 100644 --- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java +++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java @@ -76,7 +76,7 @@ public class RouteInfoManager { private final Map/* Filter Server */> filterServerTable; private final Map> 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; } diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java index b5569bb972..b287387a74 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java @@ -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 { diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java index 06bbae120c..bdcfdc95ea 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java @@ -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() { 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 7fe49b1fad..8e6eec3717 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 @@ -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> messageQueueList = new AtomicReference<>(); + AtomicReference 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 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()); } }