[ISSUE #6196] Update lastConsumeTimestamp and lastPullTimestamp in De…

…faultLitePullConsumer (#6197) * Update lastConsumeTimestamp and lastPullTimestamp in DefaultLitePullConsumer * Add lastConsumeTimestamp and lastPullTimestamp in consumerRunningInfo for DefaultLitePullConsumer * Pass the check style
apache · Feb 28, 2023 · 88000da · 88000da
1 parent 33371e5
commit 88000da
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 11 deletions.
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/AssignedMessageQueue.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/AssignedMessageQueue.java
@@ -37,10 +37,6 @@ public void setRebalanceImpl(RebalanceImpl rebalanceImpl) {
         this.rebalanceImpl = rebalanceImpl;
     }
 
-    public Set<MessageQueue> messageQueues() {
-        return assignedMessageQueueState.keySet();
-    }
-
     public boolean isPaused(MessageQueue messageQueue) {
         MessageQueueState messageQueueState = assignedMessageQueueState.get(messageQueue);
         if (messageQueueState != null) {

diff --git a/...t/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java b/...t/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
@@ -69,6 +69,7 @@
 import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
 import org.apache.rocketmq.remoting.protocol.ResponseCode;
 import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.ProcessQueueInfo;
 import org.apache.rocketmq.remoting.protocol.filter.FilterAPI;
 import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
 import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
@@ -158,7 +159,8 @@ private enum SubscriptionType {
     private final ArrayList<ConsumeMessageHook> consumeMessageHookList = new ArrayList<>();
 
     // only for test purpose, will be modified by reflection in unit test.
-    @SuppressWarnings("FieldMayBeFinal") private static boolean doNotUpdateTopicSubscribeInfoWhenSubscriptionChanged = false;
+    @SuppressWarnings("FieldMayBeFinal")
+    private static boolean doNotUpdateTopicSubscribeInfoWhenSubscriptionChanged = false;
 
     public DefaultLitePullConsumerImpl(final DefaultLitePullConsumer defaultLitePullConsumer, final RPCHook rpcHook) {
         this.defaultLitePullConsumer = defaultLitePullConsumer;
@@ -394,7 +396,7 @@ private void operateAfterRunning() throws MQClientException {
         }
         // If assign function invoke before start function, then update pull task after initialization.
         if (subscriptionType == SubscriptionType.ASSIGN) {
-            updateAssignPullTask(assignedMessageQueue.messageQueues());
+            updateAssignPullTask(assignedMessageQueue.getAssignedMessageQueues());
         }
 
         for (String topic : topicMessageQueueChangeListenerMap.keySet()) {
@@ -484,12 +486,14 @@ private void updateTopicSubscribeInfoWhenSubscriptionChanged() {
 
     /**
      * subscribe data by customizing messageQueueListener
+     *
      * @param topic
      * @param subExpression
      * @param messageQueueListener
      * @throws MQClientException
      */
-    public synchronized void subscribe(String topic, String subExpression, MessageQueueListener messageQueueListener) throws MQClientException {
+    public synchronized void subscribe(String topic, String subExpression,
+        MessageQueueListener messageQueueListener) throws MQClientException {
         try {
             if (StringUtils.isEmpty(topic)) {
                 throw new IllegalArgumentException("Topic can not be null or empty.");
@@ -516,7 +520,6 @@ public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<Messa
         }
     }
 
-
     public synchronized void subscribe(String topic, String subExpression) throws MQClientException {
         try {
             if (topic == null || "".equals(topic)) {
@@ -637,6 +640,7 @@ public synchronized List<MessageExt> poll(long timeout) {
                     consumeMessageContext.setSuccess(true);
                     this.executeHookAfter(consumeMessageContext);
                 }
+                consumeRequest.getProcessQueue().setLastConsumeTimestamp(System.currentTimeMillis());
                 return messages;
             }
         } catch (InterruptedException ignore) {
@@ -655,7 +659,7 @@ public void resume(Collection<MessageQueue> messageQueues) {
     }
 
     public synchronized void seek(MessageQueue messageQueue, long offset) throws MQClientException {
-        if (!assignedMessageQueue.messageQueues().contains(messageQueue)) {
+        if (!assignedMessageQueue.getAssignedMessageQueues().contains(messageQueue)) {
             if (subscriptionType == SubscriptionType.SUBSCRIBE) {
                 throw new MQClientException("The message queue is not in assigned list, may be rebalancing, message queue: " + messageQueue, null);
             } else {
@@ -721,7 +725,7 @@ private void removePullTask(final String topic) {
     }
 
     public synchronized void commitAll() {
-        for (MessageQueue messageQueue : assignedMessageQueue.messageQueues()) {
+        for (MessageQueue messageQueue : assignedMessageQueue.getAssignedMessageQueues()) {
             try {
                 commit(messageQueue);
             } catch (Exception e) {
@@ -732,6 +736,7 @@ public synchronized void commitAll() {
 
     /**
      * Specify offset commit
+     *
      * @param messageQueues
      * @param persist
      */
@@ -760,6 +765,7 @@ public synchronized void commit(final Map<MessageQueue, Long> messageQueues, boo
 
     /**
      * Get the queue assigned in subscribe mode
+     *
      * @return
      */
     public synchronized Set<MessageQueue> assignment() {
@@ -895,6 +901,8 @@ public void run() {
                     return;
                 }
 
+                processQueue.setLastPullTimestamp(System.currentTimeMillis());
+
                 if ((long) consumeRequestCache.size() * defaultLitePullConsumer.getPullBatchSize() > defaultLitePullConsumer.getPullThresholdForAll()) {
                     scheduledThreadPoolExecutor.schedule(this, PULL_TIME_DELAY_MILLS_WHEN_CACHE_FLOW_CONTROL, TimeUnit.MILLISECONDS);
                     if ((consumeRequestFlowControlTimes++ % 1000) == 0) {
@@ -1172,6 +1180,15 @@ public ConsumerRunningInfo consumerRunningInfo() {
         info.setProperties(prop);
 
         info.getSubscriptionSet().addAll(this.subscriptions());
+
+        for (MessageQueue mq : this.assignedMessageQueue.getAssignedMessageQueues()) {
+            ProcessQueue pq = this.assignedMessageQueue.getProcessQueue(mq);
+            ProcessQueueInfo pqInfo = new ProcessQueueInfo();
+            pqInfo.setCommitOffset(this.offsetStore.readOffset(mq, ReadOffsetType.MEMORY_FIRST_THEN_STORE));
+            pq.fillProcessQueueInfo(pqInfo);
+            info.getMqTable().put(mq, pqInfo);
+        }
+
         return info;
     }
 
@@ -1234,7 +1251,7 @@ public AssignedMessageQueue getAssignedMessageQueue() {
     }
 
     public synchronized void registerTopicMessageQueueChangeListener(String topic,
-                                                                     TopicMessageQueueChangeListener listener) throws MQClientException {
+        TopicMessageQueueChangeListener listener) throws MQClientException {
         if (topic == null || listener == null) {
             throw new MQClientException("Topic or listener is null", null);
         }