[ISSUE apache#2732] Fix message loss problem when rebalance with LitePullConsumer (apache#2832)

* [ISSUE apache#2732] Fix message loss problem when rebalance with LitePullConsumer

* Fix message loss problem when rebalance with LitePullConsumer, update 2

Author: areyouok

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

@@ -83,9 +83,12 @@ public long getPullOffset(MessageQueue messageQueue) {
         return -1;
     }
 
-    public void updatePullOffset(MessageQueue messageQueue, long offset) {
+    public void updatePullOffset(MessageQueue messageQueue, long offset, ProcessQueue processQueue) {
         MessageQueueState messageQueueState = assignedMessageQueueState.get(messageQueue);
         if (messageQueueState != null) {
+            if (messageQueueState.getProcessQueue() != processQueue) {
+                return;
+            }
             messageQueueState.setPullOffset(offset);
         }
     }

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

@@ -612,9 +612,9 @@ public synchronized void commitAll() {
         }
     }
 
-    private void updatePullOffset(MessageQueue messageQueue, long nextPullOffset) {
+    private void updatePullOffset(MessageQueue messageQueue, long nextPullOffset, ProcessQueue processQueue) {
         if (assignedMessageQueue.getSeekOffset(messageQueue) == -1) {
-            assignedMessageQueue.updatePullOffset(messageQueue, nextPullOffset);
+            assignedMessageQueue.updatePullOffset(messageQueue, nextPullOffset, processQueue);
         }
     }
 
@@ -740,6 +740,9 @@ public void run() {
                 }
 
                 long offset = nextPullOffset(messageQueue);
+                if (this.isCancelled() || processQueue.isDropped()) {
+                    return;
+                }
                 long pullDelayTimeMills = 0;
                 try {
                     SubscriptionData subscriptionData;
@@ -752,7 +755,9 @@ public void run() {
                     }
 
                     PullResult pullResult = pull(messageQueue, subscriptionData, offset, defaultLitePullConsumer.getPullBatchSize());
-
+                    if (this.isCancelled() || processQueue.isDropped()) {
+                        return;
+                    }
                     switch (pullResult.getPullStatus()) {
                         case FOUND:
                             final Object objLock = messageQueueLock.fetchLockObject(messageQueue);
@@ -769,7 +774,7 @@ public void run() {
                         default:
                             break;
                     }
-                    updatePullOffset(messageQueue, pullResult.getNextBeginOffset());
+                    updatePullOffset(messageQueue, pullResult.getNextBeginOffset(), processQueue);
                 } catch (Throwable e) {
                     pullDelayTimeMills = pullTimeDelayMillsWhenException;
                     log.error("An error occurred in pull message process.", e);