From 8720568106dde9236eb30afa5788ceccf31abdd7 Mon Sep 17 00:00:00 2001
From: ssssssnake <157384081@qq.com>
Date: Tue, 19 Mar 2019 14:14:28 +0800
Subject: [PATCH] Fix null pointer problem when consumption start time is null

---
 .../rocketmq/client/impl/consumer/ProcessQueue.java  | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java
index 0a52817ced7..2886a3c9532 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ProcessQueue.java
@@ -26,6 +26,8 @@
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.log.ClientLogger;
 import org.apache.rocketmq.logging.InternalLogger;
@@ -84,10 +86,14 @@ public void cleanExpiredMsg(DefaultMQPushConsumer pushConsumer) {
             try {
                 this.lockTreeMap.readLock().lockInterruptibly();
                 try {
-                    if (!msgTreeMap.isEmpty() && System.currentTimeMillis() - Long.parseLong(MessageAccessor.getConsumeStartTimeStamp(msgTreeMap.firstEntry().getValue())) > pushConsumer.getConsumeTimeout() * 60 * 1000) {
-                        msg = msgTreeMap.firstEntry().getValue();
+                    if (!msgTreeMap.isEmpty()) {
+                        String consumeStartTimeStamp = MessageAccessor.getConsumeStartTimeStamp(msgTreeMap.firstEntry().getValue());
+                        if (StringUtils.isNotEmpty(consumeStartTimeStamp) && System.currentTimeMillis() - Long.parseLong(consumeStartTimeStamp) > pushConsumer.getConsumeTimeout() * 60 * 1000) {
+                            msg = msgTreeMap.firstEntry().getValue();
+                        } else {
+                            break;
+                        }
                     } else {
-
                         break;
                     }
                 } finally {