Merge pull request alibaba#1705 from duhenglucky/pull_consumer_offset

[ISSUE alibaba#1706]feat(pull_consumer) refactor the consumer offset update logic
CST11021 · Jan 8, 2020 · 366e6e7 · 366e6e7
2 parents 1aa2355 + ad149dd
commit 366e6e7
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 29 deletions.
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java b/client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java
@@ -269,7 +269,7 @@ public void registerTopicMessageQueueChangeListener(String topic,
 
     @Override
     public void commitSync() {
-        this.defaultLitePullConsumerImpl.commitSync();
+        this.defaultLitePullConsumerImpl.commitAll();
     }
 
     @Override

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
@@ -177,6 +177,10 @@ public void removeAssignedMessageQueue(String topic) {
         }
     }
 
+    public Set<MessageQueue> getAssignedMessageQueues() {
+        return this.assignedMessageQueueState.keySet();
+    }
+
     private class MessageQueueState {
         private MessageQueue messageQueue;
         private ProcessQueue processQueue;

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
@@ -590,37 +590,14 @@ private void removePullTask(final String topic) {
         }
     }
 
-    public synchronized void commitSync() {
+    public synchronized void commitAll() {
         try {
             for (MessageQueue messageQueue : assignedMessageQueue.messageQueues()) {
                 long consumerOffset = assignedMessageQueue.getConsumerOffset(messageQueue);
                 if (consumerOffset != -1) {
                     ProcessQueue processQueue = assignedMessageQueue.getProcessQueue(messageQueue);
-                    long preConsumerOffset = this.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY);
-                    if (processQueue != null && !processQueue.isDropped() && consumerOffset != preConsumerOffset) {
+                    if (processQueue != null && !processQueue.isDropped()) {
                         updateConsumeOffset(messageQueue, consumerOffset);
-                        updateConsumeOffsetToBroker(messageQueue, consumerOffset, false);
-                    }
-                }
-            }
-            if (defaultLitePullConsumer.getMessageModel() == MessageModel.BROADCASTING) {
-                offsetStore.persistAll(assignedMessageQueue.messageQueues());
-            }
-        } catch (Exception e) {
-            log.error("An error occurred when update consume offset synchronously.", e);
-        }
-    }
-
-    private synchronized void commitAll() {
-        try {
-            for (MessageQueue messageQueue : assignedMessageQueue.messageQueues()) {
-                long consumerOffset = assignedMessageQueue.getConsumerOffset(messageQueue);
-                if (consumerOffset != -1) {
-                    ProcessQueue processQueue = assignedMessageQueue.getProcessQueue(messageQueue);
-                    long preConsumerOffset = this.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY);
-                    if (processQueue != null && !processQueue.isDropped() && consumerOffset != preConsumerOffset) {
-                        updateConsumeOffset(messageQueue, consumerOffset);
-                        updateConsumeOffsetToBroker(messageQueue, consumerOffset, true);
                     }
                 }
             }
@@ -927,11 +904,16 @@ public void persistConsumerOffset() {
         try {
             checkServiceState();
             Set<MessageQueue> mqs = new HashSet<MessageQueue>();
-            Set<MessageQueue> allocateMq = this.rebalanceImpl.getProcessQueueTable().keySet();
-            mqs.addAll(allocateMq);
+            if (this.subscriptionType == SubscriptionType.SUBSCRIBE) {
+                Set<MessageQueue> allocateMq = this.rebalanceImpl.getProcessQueueTable().keySet();
+                mqs.addAll(allocateMq);
+            } else if (this.subscriptionType == SubscriptionType.ASSIGN) {
+                Set<MessageQueue> assignedMessageQueue = this.assignedMessageQueue.getAssignedMessageQueues();
+                mqs.addAll(assignedMessageQueue);
+            }
             this.offsetStore.persistAll(mqs);
         } catch (Exception e) {
-            log.error("group: " + this.defaultLitePullConsumer.getConsumerGroup() + " persistConsumerOffset exception", e);
+            log.error("Persist consumer offset error for group: {} ", this.defaultLitePullConsumer.getConsumerGroup(), e);
         }
     }