[ISSUE #8674] fix():The correct consumption pattern when the heartbeat is reported

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java

@@ -131,6 +131,7 @@ public class DefaultMQPushConsumerImpl implements MQConsumerInner {
     private ConsumeMessageService consumeMessagePopService;
     private long queueFlowControlTimes = 0;
     private long queueMaxSpanFlowControlTimes = 0;
+    private boolean existPop = false;
 
     //10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     private final int[] popDelayLevel = new int[] {10, 30, 60, 120, 180, 240, 300, 360, 420, 480, 540, 600, 1200, 1800, 3600, 7200};
@@ -1370,6 +1371,9 @@ public MessageModel messageModel() {
 
     @Override
     public ConsumeType consumeType() {
+        if (!defaultMQPushConsumer.isClientRebalance() && isExistPop()) {
+            return ConsumeType.CONSUME_POP;
+        }
         return ConsumeType.CONSUME_PASSIVELY;
     }
 
@@ -1595,4 +1599,12 @@ public MessageQueueListener getMessageQueueListener() {
         }
         return defaultMQPushConsumer.getMessageQueueListener();
     }
+
+    public void setExistPop(boolean existPop) {
+        this.existPop = existPop;
+    }
+
+    public boolean isExistPop() {
+        return existPop;
+    }
 }

client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java

@@ -384,6 +384,7 @@ private boolean rebalanceByTopic(final String topic, final boolean isOrder) {
     private boolean getRebalanceResultFromBroker(final String topic, final boolean isOrder) {
         String strategyName = this.allocateMessageQueueStrategy.getName();
         Set<MessageQueueAssignment> messageQueueAssignments;
+        boolean existPop = false;
         try {
             messageQueueAssignments = this.mQClientFactory.queryAssignment(topic, consumerGroup,
                 strategyName, messageModel, QUERY_ASSIGNMENT_TIMEOUT);
@@ -401,7 +402,15 @@ private boolean getRebalanceResultFromBroker(final String topic, final boolean i
             if (messageQueueAssignment.getMessageQueue() != null) {
                 mqSet.add(messageQueueAssignment.getMessageQueue());
             }
+            if (!existPop && ConsumeType.CONSUME_POP.getTypeCN().equals(messageQueueAssignment.getMode().getName())) {
+                existPop = true;
+            }
+        }
+        if (this.consumeType() == ConsumeType.CONSUME_PASSIVELY) {
+            RebalancePushImpl rebalancePush = (RebalancePushImpl) this;
+            rebalancePush.getDefaultMQPushConsumerImpl().setExistPop(existPop);
         }
+
         Set<MessageQueue> mqAll = null;
         boolean changed = this.updateMessageQueueAssignment(topic, messageQueueAssignments, isOrder);
         if (changed) {

client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalancePushImpl.java

@@ -288,4 +288,8 @@ public ProcessQueue createProcessQueue() {
     public PopProcessQueue createPopProcessQueue() {
         return new PopProcessQueue();
     }
+
+    public DefaultMQPushConsumerImpl getDefaultMQPushConsumerImpl() {
+        return defaultMQPushConsumerImpl;
+    }
 }