fix

Change-Id: I9274ca9ede5e6fb206dba362b653224447c79a40

Author: qianye1001

broker/src/main/java/org/apache/rocketmq/broker/processor/NotificationProcessor.java

@@ -20,6 +20,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import java.util.Map;
 import java.util.Random;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.filter.ConsumerFilterData;
 import org.apache.rocketmq.broker.filter.ConsumerFilterManager;
@@ -151,7 +152,9 @@ public RemotingCommand processRequest(final ChannelHandlerContext ctx,
 
         SubscriptionData subscriptionData = null;
         ExpressionMessageFilter messageFilter = null;
-        if (brokerConfig.isUseMessageFilterForNotification() && requestHeader.getExp() != null && !requestHeader.getExp().isEmpty()) {
+        if (brokerConfig.isUseMessageFilterForNotification() &&
+            StringUtils.isNotEmpty(requestHeader.getExpType()) &&
+            StringUtils.isNotEmpty(requestHeader.getExp())) {
             try {
                 // origin topic
                 subscriptionData = FilterAPI.build(
@@ -254,7 +257,9 @@ private boolean hasMsgFromQueue(String targetTopic, NotificationRequestHeader re
                 }
                 ReferredIterator<CqUnit> iterator = null;
                 try {
-                    iterator = queue.iterateFrom(offset, maxFilterMessageNum);
+                    // In order to take into account both the file CQ and the Rocksdb CQ,
+                    // the count passed here is 32.
+                    iterator = queue.iterateFrom(offset, 32);
                     if (iterator != null) {
                         while (iterator.hasNext()) {
                             CqUnit cqUnit = iterator.next();