fix

Change-Id: I19b1c387bb3f03f5acade1bef28daaea0369613a

Author: qianye1001

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

@@ -47,7 +47,6 @@
 import org.apache.rocketmq.remoting.protocol.header.NotificationResponseHeader;
 import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
 import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
-import org.apache.rocketmq.store.ConsumeQueue;
 import org.apache.rocketmq.store.MessageFilter;
 import org.apache.rocketmq.store.exception.ConsumeQueueException;
 import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
@@ -244,7 +243,7 @@ private boolean hasMsgFromQueue(String targetTopic, NotificationRequestHeader re
         long offset = getPopOffset(targetTopic, requestHeader.getConsumerGroup(), queueId);
         try {
             long restNum = this.brokerController.getMessageStore().getMaxOffsetInQueue(targetTopic, queueId) - offset;
-            int maxFilterMessageNum = brokerController.getMessageStoreConfig().getMaxFilterMessageSize() / ConsumeQueue.CQ_STORE_UNIT_SIZE;
+            int maxFilterMessageNum = this.brokerController.getBrokerConfig().getMaxMessageFilterNumForNotification();
             boolean needFilter = restNum < maxFilterMessageNum &&
                 subscriptionData != null &&
                 messageFilter != null &&
@@ -267,6 +266,7 @@ private boolean hasMsgFromQueue(String targetTopic, NotificationRequestHeader re
                                 return true;
                             }
                         }
+                        return false;
                     }
                 } finally {
                     if (iterator != null) {

common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java

@@ -256,6 +256,7 @@ public class BrokerConfig extends BrokerIdentity {
     private boolean realTimeNotifyConsumerChange = true;
 
     private boolean useMessageFilterForNotification = true;
+    private int maxMessageFilterNumForNotification = 64;
 
     private boolean litePullMessageEnable = true;
 
@@ -2416,4 +2417,12 @@ public boolean isUseMessageFilterForNotification() {
     public void setUseMessageFilterForNotification(boolean useMessageFilterForNotification) {
         this.useMessageFilterForNotification = useMessageFilterForNotification;
     }
+
+    public int getMaxMessageFilterNumForNotification() {
+        return maxMessageFilterNumForNotification;
+    }
+
+    public void setMaxMessageFilterNumForNotification(int maxMessageFilterNumForNotification) {
+        this.maxMessageFilterNumForNotification = maxMessageFilterNumForNotification;
+    }
 }