From 0a329ba627fcffa7fb976501dfc5a0756a4f6039 Mon Sep 17 00:00:00 2001 From: Zhouxiang Zhan Date: Mon, 8 May 2023 11:49:02 +0800 Subject: [PATCH] [ISSUE #6699] Make NotificationProcessor use PopLongPollingService (#6700) * Refector PopLongPollingService * Use enum PollingResult to replace int value * Move PopLongPollingService to upper level * Move wakeUp and polling method into PopLongPollingService * NotificationProcessor use PopLongPollingService * Add BORN_TIME check * add NotificationIT --- .../rocketmq/broker/BrokerController.java | 6 +- .../broker/longpolling/PollingHeader.java | 65 ++++ .../broker/longpolling/PollingResult.java | 25 ++ .../longpolling/PopLongPollingService.java | 333 +++++++++++++++++ .../broker/longpolling/PopRequest.java | 21 +- .../processor/NotificationProcessor.java | 147 +------- .../broker/processor/PopMessageProcessor.java | 335 ++---------------- .../client/impl/mqclient/MQClientAPIExt.java | 21 +- .../remoting/protocol/RemotingCommand.java | 4 + .../test/client/rmq/RMQPopClient.java | 20 +- .../client/consumer/pop/NotificationIT.java | 74 ++++ 11 files changed, 590 insertions(+), 461 deletions(-) create mode 100644 broker/src/main/java/org/apache/rocketmq/broker/longpolling/PollingHeader.java create mode 100644 broker/src/main/java/org/apache/rocketmq/broker/longpolling/PollingResult.java create mode 100644 broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java create mode 100644 test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/NotificationIT.java diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java index e7b33e796c..fc76e67b63 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java @@ -1288,7 +1288,7 @@ public class BrokerController { } if (this.notificationProcessor != null) { - this.notificationProcessor.shutdown(); + this.notificationProcessor.getPopLongPollingService().shutdown(); } if (this.consumerIdsChangeListener != null) { @@ -1507,6 +1507,10 @@ public class BrokerController { this.ackMessageProcessor.startPopReviveService(); } + if (this.notificationProcessor != null) { + this.notificationProcessor.getPopLongPollingService().start(); + } + if (this.topicQueueMappingCleanService != null) { this.topicQueueMappingCleanService.start(); } diff --git a/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PollingHeader.java b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PollingHeader.java new file mode 100644 index 0000000000..9f6774a0f3 --- /dev/null +++ b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PollingHeader.java @@ -0,0 +1,65 @@ +/* + * 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.broker.longpolling; + +import org.apache.rocketmq.remoting.protocol.header.NotificationRequestHeader; +import org.apache.rocketmq.remoting.protocol.header.PopMessageRequestHeader; + +public class PollingHeader { + private final String consumerGroup; + private final String topic; + private final int queueId; + private final long bornTime; + private final long pollTime; + + public PollingHeader(PopMessageRequestHeader requestHeader) { + this.consumerGroup = requestHeader.getConsumerGroup(); + this.topic = requestHeader.getTopic(); + this.queueId = requestHeader.getQueueId(); + this.bornTime = requestHeader.getBornTime(); + this.pollTime = requestHeader.getPollTime(); + } + + public PollingHeader(NotificationRequestHeader requestHeader) { + this.consumerGroup = requestHeader.getConsumerGroup(); + this.topic = requestHeader.getTopic(); + this.queueId = requestHeader.getQueueId(); + this.bornTime = requestHeader.getBornTime(); + this.pollTime = requestHeader.getPollTime(); + } + + public String getConsumerGroup() { + return consumerGroup; + } + + public String getTopic() { + return topic; + } + + public int getQueueId() { + return queueId; + } + + public long getBornTime() { + return bornTime; + } + + public long getPollTime() { + return pollTime; + } +} diff --git a/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PollingResult.java b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PollingResult.java new file mode 100644 index 0000000000..6b7c4fa4a8 --- /dev/null +++ b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PollingResult.java @@ -0,0 +1,25 @@ +/* + * 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.broker.longpolling; + +public enum PollingResult { + POLLING_SUC, + POLLING_FULL, + POLLING_TIMEOUT, + NOT_POLLING; +} diff --git a/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java new file mode 100644 index 0000000000..113c91297e --- /dev/null +++ b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopLongPollingService.java @@ -0,0 +1,333 @@ +/* + * 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.broker.longpolling; + +import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap; +import io.netty.channel.ChannelHandlerContext; +import java.util.Iterator; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentSkipListSet; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.rocketmq.broker.BrokerController; +import org.apache.rocketmq.common.KeyBuilder; +import org.apache.rocketmq.common.PopAckConstants; +import org.apache.rocketmq.common.ServiceThread; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; +import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract; +import org.apache.rocketmq.remoting.netty.NettyRequestProcessor; +import org.apache.rocketmq.remoting.netty.RequestTask; +import org.apache.rocketmq.remoting.protocol.RemotingCommand; + +import static org.apache.rocketmq.broker.longpolling.PollingResult.NOT_POLLING; +import static org.apache.rocketmq.broker.longpolling.PollingResult.POLLING_FULL; +import static org.apache.rocketmq.broker.longpolling.PollingResult.POLLING_SUC; +import static org.apache.rocketmq.broker.longpolling.PollingResult.POLLING_TIMEOUT; + +public class PopLongPollingService extends ServiceThread { + private static final Logger POP_LOGGER = + LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LOGGER_NAME); + private final BrokerController brokerController; + private final NettyRequestProcessor processor; + private final ConcurrentHashMap> topicCidMap; + private final ConcurrentLinkedHashMap> pollingMap; + private long lastCleanTime = 0; + + private final AtomicLong totalPollingNum = new AtomicLong(0); + + public PopLongPollingService(BrokerController brokerController, NettyRequestProcessor processor) { + this.brokerController = brokerController; + this.processor = processor; + // 100000 topic default, 100000 lru topic + cid + qid + this.topicCidMap = new ConcurrentHashMap<>(brokerController.getBrokerConfig().getPopPollingMapSize()); + this.pollingMap = new ConcurrentLinkedHashMap.Builder>() + .maximumWeightedCapacity(this.brokerController.getBrokerConfig().getPopPollingMapSize()).build(); + } + + @Override + public String getServiceName() { + if (brokerController.getBrokerConfig().isInBrokerContainer()) { + return brokerController.getBrokerIdentity().getIdentifier() + PopLongPollingService.class.getSimpleName(); + } + return PopLongPollingService.class.getSimpleName(); + } + + @Override + public void run() { + int i = 0; + while (!this.stopped) { + try { + this.waitForRunning(20); + i++; + if (pollingMap.isEmpty()) { + continue; + } + long tmpTotalPollingNum = 0; + for (Map.Entry> entry : pollingMap.entrySet()) { + String key = entry.getKey(); + ConcurrentSkipListSet popQ = entry.getValue(); + if (popQ == null) { + continue; + } + PopRequest first; + do { + first = popQ.pollFirst(); + if (first == null) { + break; + } + if (!first.isTimeout()) { + if (popQ.add(first)) { + break; + } else { + POP_LOGGER.info("polling, add fail again: {}", first); + } + } + if (brokerController.getBrokerConfig().isEnablePopLog()) { + POP_LOGGER.info("timeout , wakeUp polling : {}", first); + } + totalPollingNum.decrementAndGet(); + wakeUp(first); + } + while (true); + if (i >= 100) { + long tmpPollingNum = popQ.size(); + tmpTotalPollingNum = tmpTotalPollingNum + tmpPollingNum; + if (tmpPollingNum > 100) { + POP_LOGGER.info("polling queue {} , size={} ", key, tmpPollingNum); + } + } + } + + if (i >= 100) { + POP_LOGGER.info("pollingMapSize={},tmpTotalSize={},atomicTotalSize={},diffSize={}", + pollingMap.size(), tmpTotalPollingNum, totalPollingNum.get(), + Math.abs(totalPollingNum.get() - tmpTotalPollingNum)); + totalPollingNum.set(tmpTotalPollingNum); + i = 0; + } + + // clean unused + if (lastCleanTime == 0 || System.currentTimeMillis() - lastCleanTime > 5 * 60 * 1000) { + cleanUnusedResource(); + } + } catch (Throwable e) { + POP_LOGGER.error("checkPolling error", e); + } + } + // clean all; + try { + for (Map.Entry> entry : pollingMap.entrySet()) { + ConcurrentSkipListSet popQ = entry.getValue(); + PopRequest first; + while ((first = popQ.pollFirst()) != null) { + wakeUp(first); + } + } + } catch (Throwable e) { + } + } + + public void notifyMessageArriving(final String topic, final int queueId) { + ConcurrentHashMap cids = topicCidMap.get(topic); + if (cids == null) { + return; + } + for (Map.Entry cid : cids.entrySet()) { + if (queueId >= 0) { + notifyMessageArriving(topic, cid.getKey(), -1); + } + notifyMessageArriving(topic, cid.getKey(), queueId); + } + } + + public boolean notifyMessageArriving(final String topic, final String cid, final int queueId) { + ConcurrentSkipListSet remotingCommands = pollingMap.get(KeyBuilder.buildPollingKey(topic, cid, queueId)); + if (remotingCommands == null || remotingCommands.isEmpty()) { + return false; + } + PopRequest popRequest = remotingCommands.pollFirst(); + //clean inactive channel + while (popRequest != null && !popRequest.getChannel().isActive()) { + totalPollingNum.decrementAndGet(); + popRequest = remotingCommands.pollFirst(); + } + + if (popRequest == null) { + return false; + } + totalPollingNum.decrementAndGet(); + if (brokerController.getBrokerConfig().isEnablePopLog()) { + POP_LOGGER.info("lock release , new msg arrive , wakeUp : {}", popRequest); + } + return wakeUp(popRequest); + } + + public boolean wakeUp(final PopRequest request) { + if (request == null || !request.complete()) { + return false; + } + if (!request.getCtx().channel().isActive()) { + return false; + } + Runnable run = () -> { + try { + final RemotingCommand response = processor.processRequest(request.getCtx(), request.getRemotingCommand()); + if (response != null) { + response.setOpaque(request.getRemotingCommand().getOpaque()); + response.markResponseType(); + NettyRemotingAbstract.writeResponse(request.getChannel(), request.getRemotingCommand(), response, future -> { + if (!future.isSuccess()) { + POP_LOGGER.error("ProcessRequestWrapper response to {} failed", request.getChannel().remoteAddress(), future.cause()); + POP_LOGGER.error(request.toString()); + POP_LOGGER.error(response.toString()); + } + }); + } + } catch (Exception e1) { + POP_LOGGER.error("ExecuteRequestWhenWakeup run", e1); + } + }; + this.brokerController.getPullMessageExecutor().submit(new RequestTask(run, request.getChannel(), request.getRemotingCommand())); + return true; + } + + /** + * @param ctx + * @param remotingCommand + * @param requestHeader + * @return + */ + public PollingResult polling(final ChannelHandlerContext ctx, RemotingCommand remotingCommand, + final PollingHeader requestHeader) { + if (requestHeader.getPollTime() <= 0 || this.isStopped()) { + return NOT_POLLING; + } + ConcurrentHashMap cids = topicCidMap.get(requestHeader.getTopic()); + if (cids == null) { + cids = new ConcurrentHashMap<>(); + ConcurrentHashMap old = topicCidMap.putIfAbsent(requestHeader.getTopic(), cids); + if (old != null) { + cids = old; + } + } + cids.putIfAbsent(requestHeader.getConsumerGroup(), Byte.MIN_VALUE); + long expired = requestHeader.getBornTime() + requestHeader.getPollTime(); + final PopRequest request = new PopRequest(remotingCommand, ctx, expired); + boolean isFull = totalPollingNum.get() >= this.brokerController.getBrokerConfig().getMaxPopPollingSize(); + if (isFull) { + POP_LOGGER.info("polling {}, result POLLING_FULL, total:{}", remotingCommand, totalPollingNum.get()); + return POLLING_FULL; + } + boolean isTimeout = request.isTimeout(); + if (isTimeout) { + if (brokerController.getBrokerConfig().isEnablePopLog()) { + POP_LOGGER.info("polling {}, result POLLING_TIMEOUT", remotingCommand); + } + return POLLING_TIMEOUT; + } + String key = KeyBuilder.buildPollingKey(requestHeader.getTopic(), requestHeader.getConsumerGroup(), + requestHeader.getQueueId()); + ConcurrentSkipListSet queue = pollingMap.get(key); + if (queue == null) { + queue = new ConcurrentSkipListSet<>(PopRequest.COMPARATOR); + ConcurrentSkipListSet old = pollingMap.putIfAbsent(key, queue); + if (old != null) { + queue = old; + } + } else { + // check size + int size = queue.size(); + if (size > brokerController.getBrokerConfig().getPopPollingSize()) { + POP_LOGGER.info("polling {}, result POLLING_FULL, singleSize:{}", remotingCommand, size); + return POLLING_FULL; + } + } + if (queue.add(request)) { + remotingCommand.setSuspended(true); + totalPollingNum.incrementAndGet(); + if (brokerController.getBrokerConfig().isEnablePopLog()) { + POP_LOGGER.info("polling {}, result POLLING_SUC", remotingCommand); + } + return POLLING_SUC; + } else { + POP_LOGGER.info("polling {}, result POLLING_FULL, add fail, {}", request, queue); + return POLLING_FULL; + } + } + + public ConcurrentLinkedHashMap> getPollingMap() { + return pollingMap; + } + + private void cleanUnusedResource() { + try { + { + Iterator>> topicCidMapIter = topicCidMap.entrySet().iterator(); + while (topicCidMapIter.hasNext()) { + Map.Entry> entry = topicCidMapIter.next(); + String topic = entry.getKey(); + if (brokerController.getTopicConfigManager().selectTopicConfig(topic) == null) { + POP_LOGGER.info("remove not exit topic {} in topicCidMap!", topic); + topicCidMapIter.remove(); + continue; + } + Iterator> cidMapIter = entry.getValue().entrySet().iterator(); + while (cidMapIter.hasNext()) { + Map.Entry cidEntry = cidMapIter.next(); + String cid = cidEntry.getKey(); + if (!brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().containsKey(cid)) { + POP_LOGGER.info("remove not exit sub {} of topic {} in topicCidMap!", cid, topic); + cidMapIter.remove(); + } + } + } + } + + { + Iterator>> pollingMapIter = pollingMap.entrySet().iterator(); + while (pollingMapIter.hasNext()) { + Map.Entry> entry = pollingMapIter.next(); + if (entry.getKey() == null) { + continue; + } + String[] keyArray = entry.getKey().split(PopAckConstants.SPLIT); + if (keyArray.length != 3) { + continue; + } + String topic = keyArray[0]; + String cid = keyArray[1]; + if (brokerController.getTopicConfigManager().selectTopicConfig(topic) == null) { + POP_LOGGER.info("remove not exit topic {} in pollingMap!", topic); + pollingMapIter.remove(); + continue; + } + if (!brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().containsKey(cid)) { + POP_LOGGER.info("remove not exit sub {} of topic {} in pollingMap!", cid, topic); + pollingMapIter.remove(); + } + } + } + } catch (Throwable e) { + POP_LOGGER.error("cleanUnusedResource", e); + } + + lastCleanTime = System.currentTimeMillis(); + } +} diff --git a/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopRequest.java b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopRequest.java index a6546e9127..a45bcce9f6 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopRequest.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/longpolling/PopRequest.java @@ -16,6 +16,7 @@ */ package org.apache.rocketmq.broker.longpolling; +import io.netty.channel.ChannelHandlerContext; import java.util.Comparator; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -27,20 +28,24 @@ import io.netty.channel.Channel; public class PopRequest { private static final AtomicLong COUNTER = new AtomicLong(Long.MIN_VALUE); - private RemotingCommand remotingCommand; - private Channel channel; - private long expired; - private AtomicBoolean complete = new AtomicBoolean(false); + private final RemotingCommand remotingCommand; + private final ChannelHandlerContext ctx; + private final long expired; + private final AtomicBoolean complete = new AtomicBoolean(false); private final long op = COUNTER.getAndIncrement(); - public PopRequest(RemotingCommand remotingCommand, Channel channel, long expired) { - this.channel = channel; + public PopRequest(RemotingCommand remotingCommand, ChannelHandlerContext ctx, long expired) { + this.ctx = ctx; this.remotingCommand = remotingCommand; this.expired = expired; } public Channel getChannel() { - return channel; + return ctx.channel(); + } + + public ChannelHandlerContext getCtx() { + return ctx; } public RemotingCommand getRemotingCommand() { @@ -63,7 +68,7 @@ public class PopRequest { public String toString() { final StringBuilder sb = new StringBuilder("PopRequest{"); sb.append("cmd=").append(remotingCommand); - sb.append(", channel=").append(channel); + sb.append(", ctx=").append(ctx); sb.append(", expired=").append(expired); sb.append(", complete=").append(complete); sb.append(", op=").append(op); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/NotificationProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/NotificationProcessor.java index 4be77468f1..d07aadfdbf 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/NotificationProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/NotificationProcessor.java @@ -16,17 +16,14 @@ */ package org.apache.rocketmq.broker.processor; -import com.googlecode.concurrentlinkedhashmap.ConcurrentLinkedHashMap; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; +import java.util.Objects; import java.util.Random; -import java.util.concurrent.ArrayBlockingQueue; import org.apache.rocketmq.broker.BrokerController; -import org.apache.rocketmq.broker.longpolling.NotificationRequest; -import org.apache.rocketmq.common.AbstractBrokerRunnable; +import org.apache.rocketmq.broker.longpolling.PollingHeader; +import org.apache.rocketmq.broker.longpolling.PollingResult; +import org.apache.rocketmq.broker.longpolling.PopLongPollingService; import org.apache.rocketmq.common.KeyBuilder; import org.apache.rocketmq.common.TopicConfig; import org.apache.rocketmq.common.constant.LoggerName; @@ -36,9 +33,7 @@ import org.apache.rocketmq.logging.org.slf4j.Logger; import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; import org.apache.rocketmq.remoting.common.RemotingHelper; import org.apache.rocketmq.remoting.exception.RemotingCommandException; -import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract; import org.apache.rocketmq.remoting.netty.NettyRequestProcessor; -import org.apache.rocketmq.remoting.netty.RequestTask; import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.apache.rocketmq.remoting.protocol.ResponseCode; import org.apache.rocketmq.remoting.protocol.header.NotificationRequestHeader; @@ -48,61 +43,13 @@ import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfi public class NotificationProcessor implements NettyRequestProcessor { private static final Logger POP_LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LOGGER_NAME); private final BrokerController brokerController; - private Random random = new Random(System.currentTimeMillis()); + private final Random random = new Random(System.currentTimeMillis()); + private final PopLongPollingService popLongPollingService; private static final String BORN_TIME = "bornTime"; - private ConcurrentLinkedHashMap> pollingMap = new ConcurrentLinkedHashMap.Builder>().maximumWeightedCapacity(100000).build(); - private Thread checkNotificationPollingThread; public NotificationProcessor(final BrokerController brokerController) { this.brokerController = brokerController; - this.checkNotificationPollingThread = new Thread(new AbstractBrokerRunnable(brokerController.getBrokerConfig()) { - @Override - public void run0() { - while (true) { - if (Thread.currentThread().isInterrupted()) { - break; - } - try { - Thread.sleep(200L); - Collection> pops = pollingMap.values(); - for (ArrayBlockingQueue popQ : pops) { - NotificationRequest tmPopRequest = popQ.peek(); - while (tmPopRequest != null) { - if (tmPopRequest.isTimeout()) { - tmPopRequest = popQ.poll(); - if (tmPopRequest == null) { - break; - } - POP_LOGGER.info("timeout , wakeUp Notification : {}", tmPopRequest); - wakeUp(tmPopRequest); - tmPopRequest = popQ.peek(); - } else { - break; - } - } - } - } catch (InterruptedException e) { - break; - } catch (Exception e) { - POP_LOGGER.error("checkNotificationPolling error", e); - } - } - } - }); - this.checkNotificationPollingThread.setDaemon(true); - this.checkNotificationPollingThread.setName("checkNotificationPolling"); - this.checkNotificationPollingThread.start(); - } - - public void shutdown() { - this.checkNotificationPollingThread.interrupt(); - } - - @Override - public RemotingCommand processRequest(final ChannelHandlerContext ctx, - RemotingCommand request) throws RemotingCommandException { - request.addExtField(BORN_TIME, String.valueOf(System.currentTimeMillis())); - return this.processRequest(ctx.channel(), request); + this.popLongPollingService = new PopLongPollingService(brokerController, this); } @Override @@ -111,55 +58,18 @@ public class NotificationProcessor implements NettyRequestProcessor { } public void notifyMessageArriving(final String topic, final int queueId) { - notifyMessageArrivingForQueue(topic, -1); - if (queueId > 0) { - notifyMessageArrivingForQueue(topic, queueId); - } + popLongPollingService.notifyMessageArriving(topic, queueId); } - public void notifyMessageArrivingForQueue(final String topic, final int queueId) { - ArrayBlockingQueue remotingCommands = pollingMap.get(KeyBuilder.buildPollingNotificationKey(topic, queueId)); - if (remotingCommands != null) { - List c = new ArrayList<>(); - remotingCommands.drainTo(c); - for (NotificationRequest notificationRequest : c) { - POP_LOGGER.info("new msg arrive , wakeUp : {}", notificationRequest); - wakeUp(notificationRequest); - } + @Override + public RemotingCommand processRequest(final ChannelHandlerContext ctx, + RemotingCommand request) throws RemotingCommandException { + request.addExtFieldIfNotExist(BORN_TIME, String.valueOf(System.currentTimeMillis())); + if (Objects.equals(request.getExtFields().get(BORN_TIME), "0")) { + request.addExtField(BORN_TIME, String.valueOf(System.currentTimeMillis())); } - } + Channel channel = ctx.channel(); - private void wakeUp(final NotificationRequest request) { - if (request == null || !request.complete()) { - return; - } - if (!request.getChannel().isActive()) { - return; - } - Runnable run = () -> { - try { - final RemotingCommand response; - response = NotificationProcessor.this.processRequest(request.getChannel(), request.getRemotingCommand()); - if (response != null) { - response.setOpaque(request.getRemotingCommand().getOpaque()); - response.markResponseType(); - NettyRemotingAbstract.writeResponse(request.getChannel(), request.getRemotingCommand(), response, future -> { - if (!future.isSuccess()) { - POP_LOGGER.error("ProcessRequestWrapper response to {} failed", request.getChannel().remoteAddress(), future.cause()); - POP_LOGGER.error(request.toString()); - POP_LOGGER.error(response.toString()); - } - }); - } - } catch (RemotingCommandException e) { - POP_LOGGER.error("ExecuteRequestWhenWakeup run", e); - } - }; - this.brokerController.getPullMessageExecutor().submit(new RequestTask(run, request.getChannel(), request.getRemotingCommand())); - } - - private RemotingCommand processRequest(final Channel channel, RemotingCommand request) - throws RemotingCommandException { RemotingCommand response = RemotingCommand.createResponseCommand(NotificationResponseHeader.class); final NotificationResponseHeader responseHeader = (NotificationResponseHeader) response.readCustomHeader(); final NotificationRequestHeader requestHeader = @@ -254,7 +164,7 @@ public class NotificationProcessor implements NettyRequestProcessor { } if (!hasMsg) { - if (polling(channel, request, requestHeader)) { + if (popLongPollingService.polling(ctx, request, new PollingHeader(requestHeader)) == PollingResult.POLLING_SUC) { return null; } } @@ -287,28 +197,7 @@ public class NotificationProcessor implements NettyRequestProcessor { } } - private boolean polling(final Channel channel, RemotingCommand remotingCommand, - final NotificationRequestHeader requestHeader) { - if (requestHeader.getPollTime() <= 0) { - return false; - } - - long expired = requestHeader.getBornTime() + requestHeader.getPollTime(); - final NotificationRequest request = new NotificationRequest(remotingCommand, channel, expired); - boolean result = false; - if (!request.isTimeout()) { - String key = KeyBuilder.buildPollingNotificationKey(requestHeader.getTopic(), requestHeader.getQueueId()); - ArrayBlockingQueue queue = pollingMap.get(key); - if (queue == null) { - queue = new ArrayBlockingQueue<>(this.brokerController.getBrokerConfig().getPopPollingSize()); - pollingMap.put(key, queue); - result = queue.offer(request); - } else { - result = queue.offer(request); - } - } - POP_LOGGER.info("polling {}, result {}", remotingCommand, result); - return result; - + public PopLongPollingService getPopLongPollingService() { + return popLongPollingService; } } diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java index a89bbb1569..efa07c2eff 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java @@ -27,6 +27,7 @@ import java.nio.ByteBuffer; import java.util.Iterator; import java.util.List; import java.util.Map.Entry; +import java.util.Objects; import java.util.Random; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -38,6 +39,9 @@ import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.filter.ConsumerFilterData; import org.apache.rocketmq.broker.filter.ConsumerFilterManager; import org.apache.rocketmq.broker.filter.ExpressionMessageFilter; +import org.apache.rocketmq.broker.longpolling.PollingHeader; +import org.apache.rocketmq.broker.longpolling.PollingResult; +import org.apache.rocketmq.broker.longpolling.PopLongPollingService; import org.apache.rocketmq.broker.longpolling.PopRequest; import org.apache.rocketmq.broker.metrics.BrokerMetricsManager; import org.apache.rocketmq.broker.pagecache.ManyMessageTransfer; @@ -64,7 +68,6 @@ import org.apache.rocketmq.remoting.exception.RemotingCommandException; import org.apache.rocketmq.remoting.metrics.RemotingMetricsManager; import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract; import org.apache.rocketmq.remoting.netty.NettyRequestProcessor; -import org.apache.rocketmq.remoting.netty.RequestTask; import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.apache.rocketmq.remoting.protocol.ResponseCode; import org.apache.rocketmq.remoting.protocol.filter.FilterAPI; @@ -97,27 +100,15 @@ public class PopMessageProcessor implements NettyRequestProcessor { String reviveTopic; private static final String BORN_TIME = "bornTime"; - private static final int POLLING_SUC = 0; - private static final int POLLING_FULL = 1; - private static final int POLLING_TIMEOUT = 2; - private static final int NOT_POLLING = 3; - - private ConcurrentHashMap> topicCidMap; - private ConcurrentLinkedHashMap> pollingMap; - private AtomicLong totalPollingNum = new AtomicLong(0); - private PopLongPollingService popLongPollingService; - private PopBufferMergeService popBufferMergeService; - private QueueLockManager queueLockManager; - private AtomicLong ckMessageNumber; + private final PopLongPollingService popLongPollingService; + private final PopBufferMergeService popBufferMergeService; + private final QueueLockManager queueLockManager; + private final AtomicLong ckMessageNumber; public PopMessageProcessor(final BrokerController brokerController) { this.brokerController = brokerController; this.reviveTopic = PopAckConstants.buildClusterReviveTopic(this.brokerController.getBrokerConfig().getBrokerClusterName()); - // 100000 topic default, 100000 lru topic + cid + qid - this.topicCidMap = new ConcurrentHashMap<>(this.brokerController.getBrokerConfig().getPopPollingMapSize()); - this.pollingMap = new ConcurrentLinkedHashMap.Builder>() - .maximumWeightedCapacity(this.brokerController.getBrokerConfig().getPopPollingMapSize()).build(); - this.popLongPollingService = new PopLongPollingService(); + this.popLongPollingService = new PopLongPollingService(brokerController, this); this.queueLockManager = new QueueLockManager(); this.popBufferMergeService = new PopBufferMergeService(this.brokerController, this); this.ckMessageNumber = new AtomicLong(); @@ -155,20 +146,13 @@ public class PopMessageProcessor implements NettyRequestProcessor { + PopAckConstants.SPLIT + PopAckConstants.CK_TAG; } - @Override - public RemotingCommand processRequest(final ChannelHandlerContext ctx, - RemotingCommand request) throws RemotingCommandException { - request.addExtField(BORN_TIME, String.valueOf(System.currentTimeMillis())); - return this.processRequest(ctx.channel(), request); - } - @Override public boolean rejectRequest() { return false; } public ConcurrentLinkedHashMap> getPollingMap() { - return pollingMap; + return popLongPollingService.getPollingMap(); } public void notifyLongPollingRequestIfNeed(String topic, String group, int queueId) { @@ -177,10 +161,10 @@ public class PopMessageProcessor implements NettyRequestProcessor { long maxOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId); long offset = Math.max(popBufferOffset, consumerOffset); if (maxOffset > offset) { - boolean notifySuccess = this.brokerController.getPopMessageProcessor().notifyMessageArriving(topic, group, -1); + boolean notifySuccess = popLongPollingService.notifyMessageArriving(topic, group, -1); if (!notifySuccess) { // notify pop queue - notifySuccess = this.brokerController.getPopMessageProcessor().notifyMessageArriving(topic, group, queueId); + notifySuccess = popLongPollingService.notifyMessageArriving(topic, group, queueId); } this.brokerController.getNotificationProcessor().notifyMessageArriving(topic, queueId); if (this.brokerController.getBrokerConfig().isEnablePopLog()) { @@ -191,71 +175,22 @@ public class PopMessageProcessor implements NettyRequestProcessor { } public void notifyMessageArriving(final String topic, final int queueId) { - ConcurrentHashMap cids = topicCidMap.get(topic); - if (cids == null) { - return; - } - for (Entry cid : cids.entrySet()) { - if (queueId >= 0) { - notifyMessageArriving(topic, cid.getKey(), -1); - } - notifyMessageArriving(topic, cid.getKey(), queueId); - } + popLongPollingService.notifyMessageArriving(topic, queueId); } public boolean notifyMessageArriving(final String topic, final String cid, final int queueId) { - ConcurrentSkipListSet remotingCommands = pollingMap.get(KeyBuilder.buildPollingKey(topic, cid, queueId)); - if (remotingCommands == null || remotingCommands.isEmpty()) { - return false; - } - PopRequest popRequest = remotingCommands.pollFirst(); - //clean inactive channel - while (popRequest != null && !popRequest.getChannel().isActive()) { - totalPollingNum.decrementAndGet(); - popRequest = remotingCommands.pollFirst(); - } - - if (popRequest == null) { - return false; - } - totalPollingNum.decrementAndGet(); - if (brokerController.getBrokerConfig().isEnablePopLog()) { - POP_LOGGER.info("lock release , new msg arrive , wakeUp : {}", popRequest); - } - return wakeUp(popRequest); + return popLongPollingService.notifyMessageArriving(topic, cid, queueId); } - private boolean wakeUp(final PopRequest request) { - if (request == null || !request.complete()) { - return false; - } - if (!request.getChannel().isActive()) { - return false; - } - Runnable run = () -> { - try { - final RemotingCommand response = processRequest(request.getChannel(), request.getRemotingCommand()); - if (response != null) { - response.setOpaque(request.getRemotingCommand().getOpaque()); - response.markResponseType(); - NettyRemotingAbstract.writeResponse(request.getChannel(), request.getRemotingCommand(), response, future -> { - if (!future.isSuccess()) { - POP_LOGGER.error("ProcessRequestWrapper response to {} failed", request.getChannel().remoteAddress(), future.cause()); - POP_LOGGER.error(request.toString()); - POP_LOGGER.error(response.toString()); - } - }); - } - } catch (RemotingCommandException e1) { - POP_LOGGER.error("ExecuteRequestWhenWakeup run", e1); - } - }; - this.brokerController.getPullMessageExecutor().submit(new RequestTask(run, request.getChannel(), request.getRemotingCommand())); - return true; - } - - private RemotingCommand processRequest(final Channel channel, RemotingCommand request) + @Override + public RemotingCommand processRequest(final ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { + request.addExtFieldIfNotExist(BORN_TIME, String.valueOf(System.currentTimeMillis())); + if (Objects.equals(request.getExtFields().get(BORN_TIME), "0")) { + request.addExtField(BORN_TIME, String.valueOf(System.currentTimeMillis())); + } + Channel channel = ctx.channel(); + RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class); final PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader(); final PopMessageRequestHeader requestHeader = @@ -453,14 +388,14 @@ public class PopMessageProcessor implements NettyRequestProcessor { getMessageResult.setStatus(GetMessageStatus.FOUND); if (restNum > 0) { // all queue pop can not notify specified queue pop, and vice versa - notifyMessageArriving(requestHeader.getTopic(), requestHeader.getConsumerGroup(), + popLongPollingService.notifyMessageArriving(requestHeader.getTopic(), requestHeader.getConsumerGroup(), requestHeader.getQueueId()); } } else { - int pollingResult = polling(channel, request, requestHeader); - if (POLLING_SUC == pollingResult) { + PollingResult pollingResult = popLongPollingService.polling(ctx, request, new PollingHeader(requestHeader)); + if (PollingResult.POLLING_SUC == pollingResult) { return null; - } else if (POLLING_FULL == pollingResult) { + } else if (PollingResult.POLLING_FULL == pollingResult) { finalResponse.setCode(ResponseCode.POLLING_FULL); } else { finalResponse.setCode(ResponseCode.POLLING_TIMEOUT); @@ -723,70 +658,6 @@ public class PopMessageProcessor implements NettyRequestProcessor { } } - /** - * @param channel - * @param remotingCommand - * @param requestHeader - * @return - */ - private int polling(final Channel channel, RemotingCommand remotingCommand, - final PopMessageRequestHeader requestHeader) { - if (requestHeader.getPollTime() <= 0 || this.popLongPollingService.isStopped()) { - return NOT_POLLING; - } - ConcurrentHashMap cids = topicCidMap.get(requestHeader.getTopic()); - if (cids == null) { - cids = new ConcurrentHashMap<>(); - ConcurrentHashMap old = topicCidMap.putIfAbsent(requestHeader.getTopic(), cids); - if (old != null) { - cids = old; - } - } - cids.putIfAbsent(requestHeader.getConsumerGroup(), Byte.MIN_VALUE); - long expired = requestHeader.getBornTime() + requestHeader.getPollTime(); - final PopRequest request = new PopRequest(remotingCommand, channel, expired); - boolean isFull = totalPollingNum.get() >= this.brokerController.getBrokerConfig().getMaxPopPollingSize(); - if (isFull) { - POP_LOGGER.info("polling {}, result POLLING_FULL, total:{}", remotingCommand, totalPollingNum.get()); - return POLLING_FULL; - } - boolean isTimeout = request.isTimeout(); - if (isTimeout) { - if (brokerController.getBrokerConfig().isEnablePopLog()) { - POP_LOGGER.info("polling {}, result POLLING_TIMEOUT", remotingCommand); - } - return POLLING_TIMEOUT; - } - String key = KeyBuilder.buildPollingKey(requestHeader.getTopic(), requestHeader.getConsumerGroup(), - requestHeader.getQueueId()); - ConcurrentSkipListSet queue = pollingMap.get(key); - if (queue == null) { - queue = new ConcurrentSkipListSet<>(PopRequest.COMPARATOR); - ConcurrentSkipListSet old = pollingMap.putIfAbsent(key, queue); - if (old != null) { - queue = old; - } - } else { - // check size - int size = queue.size(); - if (size > brokerController.getBrokerConfig().getPopPollingSize()) { - POP_LOGGER.info("polling {}, result POLLING_FULL, singleSize:{}", remotingCommand, size); - return POLLING_FULL; - } - } - if (queue.add(request)) { - remotingCommand.setSuspended(true); - totalPollingNum.incrementAndGet(); - if (brokerController.getBrokerConfig().isEnablePopLog()) { - POP_LOGGER.info("polling {}, result POLLING_SUC", remotingCommand); - } - return POLLING_SUC; - } else { - POP_LOGGER.info("polling {}, result POLLING_FULL, add fail, {}", request, queue); - return POLLING_FULL; - } - } - public final MessageExtBrokerInner buildCkMsg(final PopCheckPoint ck, final int reviveQid) { MessageExtBrokerInner msgInner = new MessageExtBrokerInner(); @@ -869,154 +740,6 @@ public class PopMessageProcessor implements NettyRequestProcessor { return byteBuffer.array(); } - public class PopLongPollingService extends ServiceThread { - - private long lastCleanTime = 0; - - @Override - public String getServiceName() { - if (PopMessageProcessor.this.brokerController.getBrokerConfig().isInBrokerContainer()) { - return PopMessageProcessor.this.brokerController.getBrokerIdentity().getIdentifier() + PopLongPollingService.class.getSimpleName(); - } - return PopLongPollingService.class.getSimpleName(); - } - - private void cleanUnusedResource() { - try { - { - Iterator>> topicCidMapIter = topicCidMap.entrySet().iterator(); - while (topicCidMapIter.hasNext()) { - Entry> entry = topicCidMapIter.next(); - String topic = entry.getKey(); - if (brokerController.getTopicConfigManager().selectTopicConfig(topic) == null) { - POP_LOGGER.info("remove not exit topic {} in topicCidMap!", topic); - topicCidMapIter.remove(); - continue; - } - Iterator> cidMapIter = entry.getValue().entrySet().iterator(); - while (cidMapIter.hasNext()) { - Entry cidEntry = cidMapIter.next(); - String cid = cidEntry.getKey(); - if (!brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().containsKey(cid)) { - POP_LOGGER.info("remove not exit sub {} of topic {} in topicCidMap!", cid, topic); - cidMapIter.remove(); - } - } - } - } - - { - Iterator>> pollingMapIter = pollingMap.entrySet().iterator(); - while (pollingMapIter.hasNext()) { - Entry> entry = pollingMapIter.next(); - if (entry.getKey() == null) { - continue; - } - String[] keyArray = entry.getKey().split(PopAckConstants.SPLIT); - if (keyArray == null || keyArray.length != 3) { - continue; - } - String topic = keyArray[0]; - String cid = keyArray[1]; - if (brokerController.getTopicConfigManager().selectTopicConfig(topic) == null) { - POP_LOGGER.info("remove not exit topic {} in pollingMap!", topic); - pollingMapIter.remove(); - continue; - } - if (!brokerController.getSubscriptionGroupManager().getSubscriptionGroupTable().containsKey(cid)) { - POP_LOGGER.info("remove not exit sub {} of topic {} in pollingMap!", cid, topic); - pollingMapIter.remove(); - continue; - } - } - } - } catch (Throwable e) { - POP_LOGGER.error("cleanUnusedResource", e); - } - - lastCleanTime = System.currentTimeMillis(); - } - - @Override - public void run() { - int i = 0; - while (!this.stopped) { - try { - this.waitForRunning(20); - i++; - if (pollingMap.isEmpty()) { - continue; - } - long tmpTotalPollingNum = 0; - Iterator>> pollingMapIterator = pollingMap.entrySet().iterator(); - while (pollingMapIterator.hasNext()) { - Entry> entry = pollingMapIterator.next(); - String key = entry.getKey(); - ConcurrentSkipListSet popQ = entry.getValue(); - if (popQ == null) { - continue; - } - PopRequest first; - do { - first = popQ.pollFirst(); - if (first == null) { - break; - } - if (!first.isTimeout()) { - if (popQ.add(first)) { - break; - } else { - POP_LOGGER.info("polling, add fail again: {}", first); - } - } - if (brokerController.getBrokerConfig().isEnablePopLog()) { - POP_LOGGER.info("timeout , wakeUp polling : {}", first); - } - totalPollingNum.decrementAndGet(); - wakeUp(first); - } - while (true); - if (i >= 100) { - long tmpPollingNum = popQ.size(); - tmpTotalPollingNum = tmpTotalPollingNum + tmpPollingNum; - if (tmpPollingNum > 100) { - POP_LOGGER.info("polling queue {} , size={} ", key, tmpPollingNum); - } - } - } - - if (i >= 100) { - POP_LOGGER.info("pollingMapSize={},tmpTotalSize={},atomicTotalSize={},diffSize={}", - pollingMap.size(), tmpTotalPollingNum, totalPollingNum.get(), - Math.abs(totalPollingNum.get() - tmpTotalPollingNum)); - totalPollingNum.set(tmpTotalPollingNum); - i = 0; - } - - // clean unused - if (lastCleanTime == 0 || System.currentTimeMillis() - lastCleanTime > 5 * 60 * 1000) { - cleanUnusedResource(); - } - } catch (Throwable e) { - POP_LOGGER.error("checkPolling error", e); - } - } - // clean all; - try { - Iterator>> pollingMapIterator = pollingMap.entrySet().iterator(); - while (pollingMapIterator.hasNext()) { - Entry> entry = pollingMapIterator.next(); - ConcurrentSkipListSet popQ = entry.getValue(); - PopRequest first; - while ((first = popQ.pollFirst()) != null) { - wakeUp(first); - } - } - } catch (Throwable e) { - } - } - } - static class TimedLock { private final AtomicBoolean lock; private volatile long lockTime; @@ -1050,7 +773,7 @@ public class PopMessageProcessor implements NettyRequestProcessor { } public class QueueLockManager extends ServiceThread { - private ConcurrentHashMap expiredLocalCache = new ConcurrentHashMap<>(100000); + private final ConcurrentHashMap expiredLocalCache = new ConcurrentHashMap<>(100000); public String buildLockKey(String topic, String consumerGroup, int queueId) { return topic + PopAckConstants.SPLIT + consumerGroup + PopAckConstants.SPLIT + queueId; @@ -1082,8 +805,8 @@ public class PopMessageProcessor implements NettyRequestProcessor { /** * is not thread safe, may cause duplicate lock * - * @param usedExpireMillis - * @return + * @param usedExpireMillis the expired time in millisecond + * @return total numbers of TimedLock */ public int cleanUnusedLock(final long usedExpireMillis) { Iterator> iterator = expiredLocalCache.entrySet().iterator(); diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java b/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java index 842b1a6405..fb8f8d11fd 100644 --- a/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java +++ b/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java @@ -601,21 +601,16 @@ public class MQClientAPIExt extends MQClientAPIImpl { CompletableFuture future = new CompletableFuture<>(); RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.NOTIFICATION, requestHeader); try { - this.getRemotingClient().invokeAsync(brokerAddr, request, timeoutMillis, responseFuture -> { - RemotingCommand response = responseFuture.getResponseCommand(); - if (response != null) { - if (response.getCode() == ResponseCode.SUCCESS) { - try { - NotificationResponseHeader responseHeader = (NotificationResponseHeader) response.decodeCommandCustomHeader(NotificationResponseHeader.class); - future.complete(responseHeader.isHasMsg()); - } catch (Throwable t) { - future.completeExceptionally(t); - } - } else { - future.completeExceptionally(new MQBrokerException(response.getCode(), response.getRemark())); + this.getRemotingClient().invoke(brokerAddr, request, timeoutMillis).thenAccept(response -> { + if (response.getCode() == ResponseCode.SUCCESS) { + try { + NotificationResponseHeader responseHeader = (NotificationResponseHeader) response.decodeCommandCustomHeader(NotificationResponseHeader.class); + future.complete(responseHeader.isHasMsg()); + } catch (Throwable t) { + future.completeExceptionally(t); } } else { - future.completeExceptionally(processNullResponseErr(responseFuture)); + future.completeExceptionally(new MQBrokerException(response.getCode(), response.getRemark())); } }); } catch (Throwable t) { diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RemotingCommand.java b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RemotingCommand.java index ffbedf9bdb..a6ed022eae 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RemotingCommand.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/protocol/RemotingCommand.java @@ -602,6 +602,10 @@ public class RemotingCommand { extFields.put(key, value); } + public void addExtFieldIfNotExist(String key, String value) { + extFields.putIfAbsent(key, value); + } + @Override public String toString() { return "RemotingCommand [code=" + code + ", language=" + language + ", version=" + version + ", opaque=" + opaque + ", flag(B)=" diff --git a/test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQPopClient.java b/test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQPopClient.java index 85dfa7b494..74d8346819 100644 --- a/test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQPopClient.java +++ b/test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQPopClient.java @@ -24,12 +24,13 @@ import org.apache.rocketmq.client.consumer.AckResult; import org.apache.rocketmq.client.consumer.PopCallback; import org.apache.rocketmq.client.consumer.PopResult; import org.apache.rocketmq.client.impl.ClientRemotingProcessor; -import org.apache.rocketmq.client.impl.MQClientAPIImpl; +import org.apache.rocketmq.client.impl.mqclient.MQClientAPIExt; import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.remoting.netty.NettyClientConfig; import org.apache.rocketmq.remoting.protocol.header.AckMessageRequestHeader; import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeRequestHeader; import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil; +import org.apache.rocketmq.remoting.protocol.header.NotificationRequestHeader; import org.apache.rocketmq.remoting.protocol.header.PopMessageRequestHeader; import org.apache.rocketmq.test.clientinterface.MQConsumer; import org.apache.rocketmq.test.util.RandomUtil; @@ -38,7 +39,7 @@ public class RMQPopClient implements MQConsumer { private static final long DEFAULT_TIMEOUT = 3000; - private MQClientAPIImpl mqClientAPI; + private MQClientAPIExt mqClientAPI; @Override public void create() { @@ -52,8 +53,8 @@ public class RMQPopClient implements MQConsumer { NettyClientConfig nettyClientConfig = new NettyClientConfig(); nettyClientConfig.setUseTLS(useTLS); - this.mqClientAPI = new MQClientAPIImpl( - nettyClientConfig, new ClientRemotingProcessor(null), null, clientConfig); + this.mqClientAPI = new MQClientAPIExt( + clientConfig, nettyClientConfig, new ClientRemotingProcessor(null), null); } @Override @@ -168,4 +169,15 @@ public class RMQPopClient implements MQConsumer { } return future; } + + public CompletableFuture notification(String brokerAddr, String topic, + String consumerGroup, int queueId, long pollTime, long bornTime, long timeoutMillis) { + NotificationRequestHeader requestHeader = new NotificationRequestHeader(); + requestHeader.setConsumerGroup(consumerGroup); + requestHeader.setTopic(topic); + requestHeader.setQueueId(queueId); + requestHeader.setPollTime(pollTime); + requestHeader.setBornTime(bornTime); + return this.mqClientAPI.notification(brokerAddr, requestHeader, timeoutMillis); + } } diff --git a/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/NotificationIT.java b/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/NotificationIT.java new file mode 100644 index 0000000000..af6f499cd4 --- /dev/null +++ b/test/src/test/java/org/apache/rocketmq/test/client/consumer/pop/NotificationIT.java @@ -0,0 +1,74 @@ +/* + * 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 java.util.concurrent.CompletableFuture; +import org.apache.rocketmq.common.attribute.CQType; +import org.apache.rocketmq.common.attribute.TopicMessageType; +import org.apache.rocketmq.common.constant.ConsumeInitMode; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.test.base.IntegrationTestBase; +import org.apache.rocketmq.test.client.rmq.RMQNormalProducer; +import org.apache.rocketmq.test.client.rmq.RMQPopClient; +import org.apache.rocketmq.test.message.MessageQueueMsg; +import org.apache.rocketmq.test.util.MQRandomUtils; +import org.assertj.core.util.Lists; +import org.junit.Before; +import org.junit.Test; + +import static org.assertj.core.api.AssertionsForClassTypes.assertThat; + +public class NotificationIT extends BasePop { + protected String topic; + protected String group; + protected RMQNormalProducer producer = null; + protected RMQPopClient client = null; + protected String brokerAddr; + protected MessageQueue messageQueue; + + @Before + public void setUp() { + brokerAddr = brokerController1.getBrokerAddr(); + topic = MQRandomUtils.getRandomTopic(); + group = initConsumerGroup(); + IntegrationTestBase.initTopic(topic, NAMESRV_ADDR, BROKER1_NAME, 8, CQType.SimpleCQ, TopicMessageType.NORMAL); + producer = getProducer(NAMESRV_ADDR, topic); + client = getRMQPopClient(); + messageQueue = new MessageQueue(topic, BROKER1_NAME, -1); + } + + @Test + public void testNotification() throws Exception { + long pollTime = 500; + CompletableFuture future1 = client.notification(brokerAddr, topic, group, messageQueue.getQueueId(), pollTime, System.currentTimeMillis(), 5000); + CompletableFuture future2 = client.notification(brokerAddr, topic, group, messageQueue.getQueueId(), pollTime, System.currentTimeMillis(), 5000); + sendMessage(1); + Boolean result1 = future1.get(); + assertThat(result1).isTrue(); + client.popMessageAsync(brokerAddr, messageQueue, 10000, 1, group, 1000, false, + ConsumeInitMode.MIN, false, null, null); + Boolean result2 = future2.get(); + assertThat(result2).isFalse(); + } + + protected void sendMessage(int num) { + MessageQueueMsg mqMsgs = new MessageQueueMsg(Lists.newArrayList(messageQueue), num); + producer.send(mqMsgs.getMsgsWithMQ()); + } + +}