Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[ISSUE 7117] check message is in memory or not when init consumer offset for pop #7118

Merged
merged 1 commit into from
Aug 9, 2023
Merged

[ISSUE 7117] check message is in memory or not when init consumer offset for pop #7118

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
1 change: 0 additions & 1 deletion broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -308,7 +308,6 @@ private void appendAck(final AckMessageRequestHeader requestHeader, final BatchA
&& putMessageResult.getPutMessageStatus() != PutMessageStatus.SLAVE_NOT_AVAILABLE) {
POP_LOGGER.error("put ack msg error:" + putMessageResult);
}
System.out.printf("put ack to store %s", ackMsg);
PopMetricsManager.incPopReviveAckPutCount(ackMsg, putMessageResult.getPutMessageStatus());
brokerController.getPopInflightMessageCounter().decrementInFlightMessageNum(topic, consumeGroup, popTime, qId, ackCount);
}
Expand Down
40 changes: 26 additions & 14 deletions broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -639,20 +639,7 @@ private long getPopOffset(String topic, String group, int queueId, int initMode,

long offset = this.brokerController.getConsumerOffsetManager().queryOffset(group, topic, queueId);
if (offset < 0) {
if (ConsumeInitMode.MIN == initMode) {
offset = this.brokerController.getMessageStore().getMinOffsetInQueue(topic, queueId);
} else {
// pop last one,then commit offset.
offset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId) - 1;
// max & no consumer offset
if (offset < 0) {
offset = 0;
}
if (init) {
this.brokerController.getConsumerOffsetManager().commitOffset(
"getPopOffset", group, topic, queueId, offset);
}
}
offset = this.getInitOffset(topic, group, queueId, initMode, init);
}

if (checkResetOffset) {
Expand All @@ -670,6 +657,31 @@ private long getPopOffset(String topic, String group, int queueId, int initMode,
}
}

private long getInitOffset(String topic, String group, int queueId, int initMode, boolean init) {
long offset;
if (ConsumeInitMode.MIN == initMode) {
return this.brokerController.getMessageStore().getMinOffsetInQueue(topic, queueId);
} else {
if (this.brokerController.getBrokerConfig().isInitPopOffsetByCheckMsgInMem() &&
this.brokerController.getMessageStore().getMinOffsetInQueue(topic, queueId) <= 0 &&
this.brokerController.getMessageStore().checkInMemByConsumeOffset(topic, queueId, 0, 1)) {
offset = 0;
} else {
// pop last one,then commit offset.
offset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId) - 1;
// max & no consumer offset
if (offset < 0) {
offset = 0;
}
}
if (init) {
this.brokerController.getConsumerOffsetManager().commitOffset(
"getPopOffset", group, topic, queueId, offset);
}
}
return offset;
}

public final MessageExtBrokerInner buildCkMsg(final PopCheckPoint ck, final int reviveQid) {
MessageExtBrokerInner msgInner = new MessageExtBrokerInner();

Expand Down
9 changes: 9 additions & 0 deletions common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -222,6 +222,7 @@ public class BrokerConfig extends BrokerIdentity {
private int popCkOffsetMaxQueueSize = 20000;
private boolean enablePopBatchAck = false;
private boolean enableNotifyAfterPopOrderLockRelease = true;
private boolean initPopOffsetByCheckMsgInMem = true;

private boolean realTimeNotifyConsumerChange = true;

Expand Down Expand Up @@ -1264,6 +1265,14 @@ public void setEnableNotifyAfterPopOrderLockRelease(boolean enableNotifyAfterPop
this.enableNotifyAfterPopOrderLockRelease = enableNotifyAfterPopOrderLockRelease;
}

public boolean isInitPopOffsetByCheckMsgInMem() {
return initPopOffsetByCheckMsgInMem;
}

public void setInitPopOffsetByCheckMsgInMem(boolean initPopOffsetByCheckMsgInMem) {
this.initPopOffsetByCheckMsgInMem = initPopOffsetByCheckMsgInMem;
}

public boolean isRealTimeNotifyConsumerChange() {
return realTimeNotifyConsumerChange;
}
Expand Down
2 changes: 1 addition & 1 deletion proxy/src/main/java/org/apache/rocketmq/proxy/service/route/TopicRouteService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,7 @@ protected static boolean isTopicRouteValid(TopicRouteData routeData) {
protected MessageQueueView buildMessageQueueView(String topic, TopicRouteData topicRouteData) {
if (isTopicRouteValid(topicRouteData)) {
MessageQueueView tmp = new MessageQueueView(topic, topicRouteData);
log.info("load topic route from namesrv. topic: {}, queue: {}", topic, tmp);
log.debug("load topic route from namesrv. topic: {}, queue: {}", topic, tmp);
return tmp;
}
return MessageQueueView.WRAPPED_EMPTY_QUEUE;
Expand Down