[ISSUE #9196] Broker return pop stats when receive notification (#9197)

Author: qianye1001

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

@@ -161,8 +161,11 @@ public RemotingCommand processRequest(final ChannelHandlerContext ctx,
         }
 
         if (!hasMsg) {
-            if (popLongPollingService.polling(ctx, request, new PollingHeader(requestHeader)) == PollingResult.POLLING_SUC) {
+            PollingResult pollingResult = popLongPollingService.polling(ctx, request, new PollingHeader(requestHeader));
+            if (pollingResult == PollingResult.POLLING_SUC) {
                 return null;
+            } else if (pollingResult == PollingResult.POLLING_FULL) {
+                responseHeader.setPollingFull(true);
             }
         }
         response.setCode(ResponseCode.SUCCESS);

client/src/main/java/org/apache/rocketmq/client/consumer/NotifyResult.java

@@ -0,0 +1,45 @@
+/*
+ * 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.client.consumer;
+
+public class NotifyResult {
+    private boolean hasMsg;
+    private boolean pollingFull;
+
+    public boolean isHasMsg() {
+        return hasMsg;
+    }
+
+    public boolean isPollingFull() {
+        return pollingFull;
+    }
+
+    public void setHasMsg(boolean hasMsg) {
+        this.hasMsg = hasMsg;
+    }
+
+    public void setPollingFull(boolean pollingFull) {
+        this.pollingFull = pollingFull;
+    }
+
+    @Override public String toString() {
+        return "NotifyResult{" +
+            "hasMsg=" + hasMsg +
+            ", pollingFull=" + pollingFull +
+            '}';
+    }
+}

client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java

@@ -24,6 +24,7 @@
 import org.apache.rocketmq.client.ClientConfig;
 import org.apache.rocketmq.client.consumer.AckCallback;
 import org.apache.rocketmq.client.consumer.AckResult;
+import org.apache.rocketmq.client.consumer.NotifyResult;
 import org.apache.rocketmq.client.consumer.PopCallback;
 import org.apache.rocketmq.client.consumer.PopResult;
 import org.apache.rocketmq.client.consumer.PullCallback;
@@ -620,14 +621,23 @@ public CompletableFuture<Void> unlockBatchMQOneway(String brokerAddr,
     }
 
     public CompletableFuture<Boolean> notification(String brokerAddr, NotificationRequestHeader requestHeader,
+        long timeoutMillis) {
+        return notificationWithPollingStats(brokerAddr, requestHeader, timeoutMillis).thenApply(NotifyResult::isHasMsg);
+    }
+
+    public CompletableFuture<NotifyResult> notificationWithPollingStats(String brokerAddr,
+        NotificationRequestHeader requestHeader,
         long timeoutMillis) {
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.NOTIFICATION, requestHeader);
         return this.getRemotingClient().invoke(brokerAddr, request, timeoutMillis).thenCompose(response -> {
-            CompletableFuture<Boolean> future0 = new CompletableFuture<>();
+            CompletableFuture<NotifyResult> future0 = new CompletableFuture<>();
             if (response.getCode() == ResponseCode.SUCCESS) {
                 try {
                     NotificationResponseHeader responseHeader = (NotificationResponseHeader) response.decodeCommandCustomHeader(NotificationResponseHeader.class);
-                    future0.complete(responseHeader.isHasMsg());
+                    NotifyResult notifyResult = new NotifyResult();
+                    notifyResult.setHasMsg(responseHeader.isHasMsg());
+                    notifyResult.setPollingFull(responseHeader.isPollingFull());
+                    future0.complete(notifyResult);
                 } catch (Throwable t) {
                     future0.completeExceptionally(t);
                 }

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/NotificationResponseHeader.java

@@ -26,10 +26,20 @@ public class NotificationResponseHeader implements CommandCustomHeader {
     @CFNotNull
     private boolean hasMsg = false;
 
+    private boolean pollingFull = false;
+
     public boolean isHasMsg() {
         return hasMsg;
     }
 
+    public boolean isPollingFull() {
+        return pollingFull;
+    }
+
+    public void setPollingFull(boolean pollingFull) {
+        this.pollingFull = pollingFull;
+    }
+
     public void setHasMsg(boolean hasMsg) {
         this.hasMsg = hasMsg;
     }