opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

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

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

* Fix message loss problem when rebalance with LitePullConsumer, update 2
This commit is contained in:
huangli
2021-04-27 17:33:01 +08:00
committed by GitHub
parent c3d464108e
commit f77d7fee68
2 changed files with 13 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/impl/consumer/AssignedMessageQueue.java
+4 -1
View File
@@ -83,9 +83,12 @@ public class AssignedMessageQueue {
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
+9 -4
View File
@@ -612,9 +612,9 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
}
}
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 class DefaultLitePullConsumerImpl implements MQConsumerInner {
}
long offset = nextPullOffset(messageQueue);
if (this.isCancelled() || processQueue.isDropped()) {
return;
}
long pullDelayTimeMills = 0;
try {
SubscriptionData subscriptionData;
@@ -752,7 +755,9 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
}
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 class DefaultLitePullConsumerImpl implements MQConsumerInner {
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);