chi3316 commented on code in PR #8980:
URL: https://github.com/apache/rocketmq/pull/8980#discussion_r1861712755
##########
store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java:
##########
@@ -1084,46 +1084,42 @@ public int doPut(MessageExtBrokerInner message, boolean
roll) throws Exception {
putMessageResult = messageStore.putMessage(message);
}
- int retryNum = 0;
- while (retryNum < 3) {
- if (null == putMessageResult || null ==
putMessageResult.getPutMessageStatus()) {
- retryNum++;
- } else {
- switch (putMessageResult.getPutMessageStatus()) {
- case PUT_OK:
- if (brokerStatsManager != null) {
-
this.brokerStatsManager.incTopicPutNums(message.getTopic(), 1, 1);
- if (putMessageResult.getAppendMessageResult() !=
null) {
-
this.brokerStatsManager.incTopicPutSize(message.getTopic(),
-
putMessageResult.getAppendMessageResult().getWroteBytes());
- }
-
this.brokerStatsManager.incBrokerPutNums(message.getTopic(), 1);
+ if (putMessageResult != null && putMessageResult.getPutMessageStatus()
!= null) {
+ switch (putMessageResult.getPutMessageStatus()) {
+ case PUT_OK:
+ if (brokerStatsManager != null) {
+ brokerStatsManager.incTopicPutNums(message.getTopic(),
1, 1);
+ if (putMessageResult.getAppendMessageResult() != null)
{
+
brokerStatsManager.incTopicPutSize(message.getTopic(),
putMessageResult.getAppendMessageResult().getWroteBytes());
}
- return PUT_OK;
- case SERVICE_NOT_AVAILABLE:
- return PUT_NEED_RETRY;
- case MESSAGE_ILLEGAL:
- case PROPERTIES_SIZE_EXCEEDED:
+
brokerStatsManager.incBrokerPutNums(message.getTopic(), 1);
+ }
+ return PUT_OK;
+
+ case MESSAGE_ILLEGAL:
+ case PROPERTIES_SIZE_EXCEEDED:
+ return PUT_NO_RETRY;
+
+ case SERVICE_NOT_AVAILABLE:
+ case FLUSH_DISK_TIMEOUT:
+ case FLUSH_SLAVE_TIMEOUT:
+ case OS_PAGE_CACHE_BUSY:
+ case CREATE_MAPPED_FILE_FAILED:
+ case SLAVE_NOT_AVAILABLE:
+ return PUT_NEED_RETRY;
+
+ case UNKNOWN_ERROR:
+ default:
+ if (storeConfig.isTimerSkipUnknownError()) {
+ LOGGER.warn("Skipping message due to unknown error,
msg: {}", message);
return PUT_NO_RETRY;
- case CREATE_MAPPED_FILE_FAILED:
- case FLUSH_DISK_TIMEOUT:
- case FLUSH_SLAVE_TIMEOUT:
- case OS_PAGE_CACHE_BUSY:
- case SLAVE_NOT_AVAILABLE:
- case UNKNOWN_ERROR:
- default:
- retryNum++;
- }
- }
- Thread.sleep(50);
- if (escapeBridgeHook != null) {
- putMessageResult = escapeBridgeHook.apply(message);
- } else {
- putMessageResult = messageStore.putMessage(message);
+ } else {
+ holdMomentForUnknownError();
+ return PUT_NEED_RETRY;
+ }
}
- LOGGER.warn("Retrying to do put timer msg retryNum:{} putRes:{}
msg:{}", retryNum, putMessageResult, message);
}
- return PUT_NO_RETRY;
+ return PUT_NEED_RETRY;
}
public MessageExtBrokerInner convertMessage(MessageExt msgExt, boolean
needRoll) {
Review Comment:
I’ve added a unit test to cover this function. Please take a look👀
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
