package com.alibaba.rocketmq.client.consumer.cacheable;

import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.client.producer.concurrent.DefaultLocalMessageStore;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.StashableMessage;
import org.slf4j.Logger;

import java.util.concurrent.LinkedBlockingQueue;

public class ProcessMessageTask implements Runnable {

    private static final Logger LOGGER = ClientLogger.getLog();

    private static final String NEXT_TIME_KEY = "next_time";

    private MessageExt message;

    private MessageHandler messageHandler;

    private DefaultLocalMessageStore localMessageStore;

    private LinkedBlockingQueue<MessageExt> messageQueue;

    public ProcessMessageTask(MessageExt message, //Message to process.
                              MessageHandler messageHandler, //Message handler.
                              DefaultLocalMessageStore localMessageStore, //Local message store.
                              LinkedBlockingQueue<MessageExt> messageQueue //Buffered message queue.
    ) {
        this.message = message;
        this.messageHandler = messageHandler;
        this.localMessageStore = localMessageStore;
        this.messageQueue = messageQueue;
    }

    @Override
    public void run() {
        try {
            int result = messageHandler.handle(message);
            //Remove the message from in-progress queue.
            if (null != messageQueue && messageQueue.contains(message)) {
                messageQueue.remove(message);
                LOGGER.info("Now " + messageQueue.size() + " messages in queue.");
            }

            if (result > 0) {
                StashableMessage stashableMessage = message.buildStashableMessage();
                stashableMessage.putUserProperty(NEXT_TIME_KEY, String.valueOf(System.currentTimeMillis() + result));
                LOGGER.info("Stashing message[msgId=" + message.getMsgId() + "] for later retry in " + result
                        + " milliseconds.");
                localMessageStore.stash(stashableMessage);
                LOGGER.info("Message stashed.");
            }
        } catch (Exception e) {
            LOGGER.error("ProcessMessageTask error", e);
        }
    }

}

@@ -374,16 +374,63 @@
 ngQueue;
+%0Aimport java.util.concurrent.atomic.AtomicLong;
 %0A%0Apublic
@@ -794,24 +794,160 @@
 sageQueue;%0A%0A
+    private static final AtomicLong LOG_COUNTER = new AtomicLong(0L);%0A%0A    private static final long LOG_PERFORMANCE_INTERVAL = 1000L;%0A%0A
     public P
@@ -1527,64 +1527,452 @@
 y %7B%0A
-            int result = messageHandler.handle(message);
+%0A            boolean logPerformance = LOG_COUNTER.incrementAndGet() %25 LOG_PERFORMANCE_INTERVAL == 0;%0A%0A            long start = 0L;%0A            if (logPerformance) %7B%0A                start = System.currentTimeMillis();%0A            %7D%0A%0A            int result = messageHandler.handle(message);%0A%0A            if (logPerformance) %7B%0A                LOGGER.info(%22Business processing takes %22 + (System.currentTimeMillis() - start) + %22 ms%22);%0A            %7D%0A
 %0A   

Add back logs, with frequency control.