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

[ISSUE #7601] Fix slave acting master bug (#7603)

Browse Source 
* fix NullPointerException when message escape to remote

* fix NumberFormatException when message retry to escape to remote

* fix timerCheckPoint of the master is not updated, causing the timer message to be replayed after master is restarted

* Use properties copies instead of referencing the same map when converting message
This commit is contained in:
gaoyf
2023-12-07 11:25:22 +08:00
committed by GitHub
parent c2c29c2435
commit faae64715d
4 changed files with 20 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+1
View File
@@ -2108,6 +2108,7 @@ public class BrokerController {
isScheduleServiceStart = shouldStart;
if (timerMessageStore != null) {
timerMessageStore.syncLastReadTimeMs();
timerMessageStore.setShouldRunningDequeue(shouldStart);
}
}
broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java
+3 -1
View File
@@ -215,11 +215,13 @@ public class SlaveSynchronize {
String masterAddrBak = this.masterAddr;
if (masterAddrBak != null) {
try {
if (null != brokerController.getMessageStore().getTimerMessageStore()) {
if (null != brokerController.getMessageStore().getTimerMessageStore() &&
!brokerController.getTimerMessageStore().isShouldRunningDequeue()) {
TimerCheckpoint checkpoint = this.brokerController.getBrokerOuterAPI().getTimerCheckPoint(masterAddrBak);
if (null != this.brokerController.getTimerCheckpoint()) {
this.brokerController.getTimerCheckpoint().setLastReadTimeMs(checkpoint.getLastReadTimeMs());
this.brokerController.getTimerCheckpoint().setMasterTimerQueueOffset(checkpoint.getMasterTimerQueueOffset());
this.brokerController.getTimerCheckpoint().getDataVersion().assignNewOne(checkpoint.getDataVersion());
}
}
} catch (Exception e) {
common/src/main/java/org/apache/rocketmq/common/message/MessageAccessor.java
+7
View File
@@ -17,6 +17,7 @@
package org.apache.rocketmq.common.message;
import java.util.HashMap;
import java.util.Map;
public class MessageAccessor {
@@ -96,4 +97,10 @@ public class MessageAccessor {
return newMsg;
}
public static Map<String, String> deepCopyProperties(Map<String, String> properties) {
if (properties == null) {
return null;
}
return new HashMap<>(properties);
}
}
store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java
+9 -3
View File
@@ -602,6 +602,10 @@ public class TimerMessageStore {
this.shouldRunningDequeue = shouldRunningDequeue;
}
public boolean isShouldRunningDequeue() {
return shouldRunningDequeue;
}
public void addMetric(MessageExt msg, int value) {
try {
if (null == msg || null == msg.getProperty(MessageConst.PROPERTY_REAL_TOPIC)) {
@@ -1084,8 +1088,10 @@ public class TimerMessageStore {
case PUT_OK:
if (brokerStatsManager != null) {
this.brokerStatsManager.incTopicPutNums(message.getTopic(), 1, 1);
this.brokerStatsManager.incTopicPutSize(message.getTopic(),
putMessageResult.getAppendMessageResult().getWroteBytes());
if (putMessageResult.getAppendMessageResult() != null) {
this.brokerStatsManager.incTopicPutSize(message.getTopic(),
putMessageResult.getAppendMessageResult().getWroteBytes());
}
this.brokerStatsManager.incBrokerPutNums(message.getTopic(), 1);
}
return PUT_OK;
@@ -1119,7 +1125,7 @@ public class TimerMessageStore {
MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
msgInner.setBody(msgExt.getBody());
msgInner.setFlag(msgExt.getFlag());
MessageAccessor.setProperties(msgInner, msgExt.getProperties());
MessageAccessor.setProperties(msgInner, MessageAccessor.deepCopyProperties(msgExt.getProperties()));
TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag());
long tagsCodeValue =
MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());