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 #2708] Client may submit wrong offset when network instability

Browse Source
This commit is contained in:
斜阳
2021-05-25 22:13:12 +08:00
parent 6ab6afd006
commit da589b1975
4 changed files with 15 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
+4 -1
View File
@@ -115,6 +115,8 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
*/
private static final long PULL_TIME_DELAY_MILLS_WHEN_PAUSE = 1000;
private static final long PULL_TIME_DELAY_MILLS_ON_EXCEPTION = 3 * 1000;
private DefaultLitePullConsumer defaultLitePullConsumer;
private final ConcurrentMap<MessageQueue, PullTaskImpl> taskTable =
@@ -743,7 +745,8 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
try {
offset = nextPullOffset(messageQueue);
} catch (MQClientException e) {
log.error("get next pull offset failed, maybe timeout exception");
log.error("Failed to get next pull offset", e);
scheduledThreadPoolExecutor.schedule(this, PULL_TIME_DELAY_MILLS_ON_EXCEPTION, TimeUnit.MILLISECONDS);
return;
}
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
+5 -3
View File
@@ -269,12 +269,14 @@ public class DefaultMQPushConsumerImpl implements MQConsumerInner {
}
} else {
if (processQueue.isLocked()) {
if (!pullRequest.isLockedFirst()) {
if (!pullRequest.isPreviouslyLocked()) {
long offset = -1L;
try {
offset = this.rebalanceImpl.computePullFromWhereWithException(pullRequest.getMessageQueue());
} catch (MQClientException e) {
log.error("compute consume offset failed, maybe timeout exception");
this.executePullRequestLater(pullRequest, pullTimeDelayMillsWhenException);
log.error("Failed to compute pull offset", e);
return;
}
boolean brokerBusy = offset < pullRequest.getNextOffset();
log.info("the first time to pull message, so fix offset from broker. pullRequest: {} NewOffset: {} brokerBusy: {}",
@@ -284,7 +286,7 @@ public class DefaultMQPushConsumerImpl implements MQConsumerInner {
pullRequest, offset);
}
pullRequest.setLockedFirst(true);
pullRequest.setPreviouslyLocked(true);
pullRequest.setNextOffset(offset);
}
} else {
client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullRequest.java
+5 -5
View File
@@ -23,14 +23,14 @@ public class PullRequest {
private MessageQueue messageQueue;
private ProcessQueue processQueue;
private long nextOffset;
private boolean lockedFirst = false;
private boolean previouslyLocked = false;
public boolean isLockedFirst() {
return lockedFirst;
public boolean isPreviouslyLocked() {
return previouslyLocked;
}
public void setLockedFirst(boolean lockedFirst) {
this.lockedFirst = lockedFirst;
public void setPreviouslyLocked(boolean previouslyLocked) {
this.previouslyLocked = previouslyLocked;
}
public String getConsumerGroup() {
client/src/main/java/org/apache/rocketmq/client/impl/consumer/RebalanceImpl.java
+1 -1
View File
@@ -375,7 +375,7 @@ public abstract class RebalanceImpl {
this.removeDirtyOffset(mq);
ProcessQueue pq = new ProcessQueue();
long nextOffset = 0L;
long nextOffset = -1L;
try {
nextOffset = this.computePullFromWhereWithException(mq);
} catch (MQClientException e) {