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

[ISSUE #8979] Add configurable switch for timer message retry logic (#8980)

Browse Source
This commit is contained in:
小陈
2024-12-05 17:49:03 +08:00
committed by GitHub
parent 01a51231b3
commit d1fd7af3f1
3 changed files with 149 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+9
View File
@@ -99,6 +99,7 @@ public class MessageStoreConfig {
private boolean timerSkipUnknownError = false;
private boolean timerWarmEnable = false;
private boolean timerStopDequeue = false;
private boolean timerEnableRetryUntilSuccess = false;
private int timerCongestNumEachSlot = Integer.MAX_VALUE;
private int timerMetricSmallThreshold = 1000000;
@@ -1689,6 +1690,14 @@ public class MessageStoreConfig {
this.timerSkipUnknownError = timerSkipUnknownError;
}
public boolean isTimerEnableRetryUntilSuccess() {
return timerEnableRetryUntilSuccess;
}
public void setTimerEnableRetryUntilSuccess(boolean timerEnableRetryUntilSuccess) {
this.timerEnableRetryUntilSuccess = timerEnableRetryUntilSuccess;
}
public boolean isTimerWarmEnable() {
return timerWarmEnable;
}
store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java
+70 -51
View File
@@ -1097,46 +1097,44 @@ public class TimerMessageStore {
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:
case WHEEL_TIMER_NOT_ENABLE:
case WHEEL_TIMER_MSG_ILLEGAL:
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++;
}
} else {
holdMomentForUnknownError();
return PUT_NEED_RETRY;
}
}
Thread.sleep(50);
if (escapeBridgeHook != null) {
putMessageResult = escapeBridgeHook.apply(message);
} else {
putMessageResult = messageStore.putMessage(message);
}
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) {
@@ -1471,7 +1469,6 @@ public class TimerMessageStore {
}
public class TimerDequeuePutMessageService extends AbstractStateService {
@Override
public String getServiceName() {
return getServiceThreadName() + this.getClass().getSimpleName();
@@ -1481,6 +1478,7 @@ public class TimerMessageStore {
public void run() {
setState(AbstractStateService.START);
TimerMessageStore.LOGGER.info(this.getServiceName() + " service start");
while (!this.isStopped() || dequeuePutQueue.size() != 0) {
try {
setState(AbstractStateService.WAITING);
@@ -1488,41 +1486,63 @@ public class TimerMessageStore {
if (null == tr) {
continue;
}
setState(AbstractStateService.RUNNING);
boolean doRes = false;
boolean tmpDequeueChangeFlag = false;
try {
while (!isStopped() && !doRes) {
while (!isStopped()) {
if (!isRunningDequeue()) {
dequeueStatusChangeFlag = true;
tmpDequeueChangeFlag = true;
break;
}
try {
perfCounterTicks.startTick(DEQUEUE_PUT);
MessageExt msgExt = tr.getMsg();
DefaultStoreMetricsManager.incTimerDequeueCount(getRealTopic(msgExt));
if (tr.getEnqueueTime() == Long.MAX_VALUE) {
// never enqueue, mark it.
// Never enqueue, mark it.
MessageAccessor.putProperty(msgExt, TIMER_ENQUEUE_MS, String.valueOf(Long.MAX_VALUE));
}
addMetric(msgExt, -1);
MessageExtBrokerInner msg = convert(msgExt, tr.getEnqueueTime(), needRoll(tr.getMagic()));
doRes = PUT_NEED_RETRY != doPut(msg, needRoll(tr.getMagic()));
while (!doRes && !isStopped()) {
if (!isRunningDequeue()) {
dequeueStatusChangeFlag = true;
tmpDequeueChangeFlag = true;
break;
boolean processed = false;
int retryCount = 0;
while (!processed && !isStopped()) {
int result = doPut(msg, needRoll(tr.getMagic()));
if (result == PUT_OK) {
processed = true;
} else if (result == PUT_NO_RETRY) {
TimerMessageStore.LOGGER.warn("Skipping message due to unrecoverable error. Msg: {}", msg);
processed = true;
} else {
retryCount++;
// Without enabling TimerEnableRetryUntilSuccess, messages will retry up to 3 times before being discarded
if (!storeConfig.isTimerEnableRetryUntilSuccess() && retryCount >= 3) {
TimerMessageStore.LOGGER.error("Message processing failed after {} retries. Msg: {}", retryCount, msg);
processed = true;
} else {
Thread.sleep(500L * precisionMs / 1000);
TimerMessageStore.LOGGER.warn("Retrying to process message. Retry count: {}, Msg: {}", retryCount, msg);
}
}
doRes = PUT_NEED_RETRY != doPut(msg, needRoll(tr.getMagic()));
Thread.sleep(500L * precisionMs / 1000);
}
perfCounterTicks.endTick(DEQUEUE_PUT);
break;
} catch (Throwable t) {
LOGGER.info("Unknown error", t);
TimerMessageStore.LOGGER.info("Unknown error", t);
if (storeConfig.isTimerSkipUnknownError()) {
doRes = true;
break;
} else {
holdMomentForUnknownError();
}
@@ -1531,7 +1551,6 @@ public class TimerMessageStore {
} finally {
tr.idempotentRelease(!tmpDequeueChangeFlag);
}
} catch (Throwable e) {
TimerMessageStore.LOGGER.error("Error occurred in " + getServiceName(), e);
}
store/src/test/java/org/apache/rocketmq/store/timer/TimerMessageStoreTest.java
+70 -17
View File
@@ -19,6 +19,7 @@ package org.apache.rocketmq.store.timer;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
@@ -30,6 +31,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.rocketmq.common.BrokerConfig;
@@ -40,23 +42,26 @@ import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.GetMessageResult;
import org.apache.rocketmq.store.GetMessageStatus;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.config.FlushDiskType;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.AppendMessageStatus;
import org.apache.rocketmq.store.AppendMessageResult;
import org.apache.rocketmq.store.GetMessageResult;
import org.apache.rocketmq.store.GetMessageStatus;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
@@ -65,10 +70,16 @@ import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertFalse;
import static org.mockito.Mockito.when;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.times;
public class TimerMessageStoreTest {
private final byte[] msgBody = new byte[1024];
private static MessageStore messageStore;
private MessageStore mockMessageStore;
private SocketAddress bornHost;
private SocketAddress storeHost;
@@ -100,21 +111,23 @@ public class TimerMessageStoreTest {
storeConfig.setTimerInterceptDelayLevel(true);
storeConfig.setTimerPrecisionMs(precisionMs);
mockMessageStore = Mockito.mock(MessageStore.class);
messageStore = new DefaultMessageStore(storeConfig, new BrokerStatsManager("TimerTest",false), new MyMessageArrivingListener(), new BrokerConfig(), new ConcurrentHashMap<>());
boolean load = messageStore.load();
assertTrue(load);
messageStore.start();
}
public TimerMessageStore createTimerMessageStore(String rootDir) throws IOException {
public TimerMessageStore createTimerMessageStore(String rootDir , boolean needMock) throws IOException {
if (null == rootDir) {
rootDir = StoreTestUtils.createBaseDir();
}
TimerCheckpoint timerCheckpoint = new TimerCheckpoint(rootDir + File.separator + "config" + File.separator + "timercheck");
TimerMetrics timerMetrics = new TimerMetrics(rootDir + File.separator + "config" + File.separator + "timermetrics");
TimerMessageStore timerMessageStore = new TimerMessageStore(messageStore, storeConfig, timerCheckpoint, timerMetrics, null);
messageStore.setTimerMessageStore(timerMessageStore);
MessageStore ms = needMock ? mockMessageStore : messageStore;
TimerMessageStore timerMessageStore = new TimerMessageStore(ms, storeConfig, timerCheckpoint, timerMetrics, null);
ms.setTimerMessageStore(timerMessageStore);
baseDirs.add(rootDir);
timerStores.add(timerMessageStore);
@@ -170,7 +183,7 @@ public class TimerMessageStoreTest {
Assume.assumeFalse(MixAll.isWindows());
String topic = "TimerTest_testPutTimerMessage";
final TimerMessageStore timerMessageStore = createTimerMessageStore(null);
final TimerMessageStore timerMessageStore = createTimerMessageStore(null , false);
timerMessageStore.load();
timerMessageStore.start(true);
@@ -212,12 +225,52 @@ public class TimerMessageStoreTest {
}
}
@Test
public void testRetryUntilSuccess() throws Exception {
storeConfig.setTimerEnableRetryUntilSuccess(true);
TimerMessageStore timerMessageStore = createTimerMessageStore(null , true);
timerMessageStore.load();
timerMessageStore.setShouldRunningDequeue(true);
Field stateField = TimerMessageStore.class.getDeclaredField("state");
stateField.setAccessible(true);
stateField.set(timerMessageStore, TimerMessageStore.RUNNING);
MessageExtBrokerInner msg = buildMessage(3000L, "TestRetry", true);
transformTimerMessage(timerMessageStore, msg);
TimerRequest timerRequest = new TimerRequest(100, 200, 3000, System.currentTimeMillis(), 0, msg);
boolean offered = timerMessageStore.dequeuePutQueue.offer(timerRequest);
assertTrue(offered);
assertFalse(timerMessageStore.dequeuePutQueue.isEmpty());
// If enableRetryUntilSuccess is set and putMessage return NEED_RETRY type, the message should be retried until success.
when(mockMessageStore.putMessage(any(MessageExtBrokerInner.class)))
.thenReturn(new PutMessageResult(PutMessageStatus.FLUSH_DISK_TIMEOUT, null))
.thenReturn(new PutMessageResult(PutMessageStatus.FLUSH_SLAVE_TIMEOUT, null))
.thenReturn(new PutMessageResult(PutMessageStatus.OS_PAGE_CACHE_BUSY, null))
.thenReturn(new PutMessageResult(PutMessageStatus.OS_PAGE_CACHE_BUSY, null))
.thenReturn(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, null))
.thenReturn(new PutMessageResult(PutMessageStatus.PUT_OK, new AppendMessageResult(AppendMessageStatus.PUT_OK)));
final CountDownLatch latch = new CountDownLatch(1);
new Thread(() -> {
try {
timerMessageStore.getDequeuePutMessageServices()[0].run();
} finally {
latch.countDown();
}
}).start();
latch.await(10, TimeUnit.SECONDS);
assertTrue(timerMessageStore.dequeuePutQueue.isEmpty());
verify(mockMessageStore, times(6)).putMessage(any(MessageExtBrokerInner.class));
}
@Test
public void testTimerFlowControl() throws Exception {
String topic = "TimerTest_testTimerFlowControl";
storeConfig.setTimerCongestNumEachSlot(100);
TimerMessageStore timerMessageStore = createTimerMessageStore(null);
TimerMessageStore timerMessageStore = createTimerMessageStore(null , false);
timerMessageStore.load();
timerMessageStore.start(true);
@@ -264,7 +317,7 @@ public class TimerMessageStoreTest {
String topic = "TimerTest_testPutExpiredTimerMessage";
TimerMessageStore timerMessageStore = createTimerMessageStore(null);
TimerMessageStore timerMessageStore = createTimerMessageStore(null ,false);
timerMessageStore.load();
timerMessageStore.start(true);
@@ -288,7 +341,7 @@ public class TimerMessageStoreTest {
public void testDeleteTimerMessage() throws Exception {
String topic = "TimerTest_testDeleteTimerMessage";
TimerMessageStore timerMessageStore = createTimerMessageStore(null);
TimerMessageStore timerMessageStore = createTimerMessageStore(null ,false);
timerMessageStore.load();
timerMessageStore.start(true);
@@ -325,7 +378,7 @@ public class TimerMessageStoreTest {
public void testPutDeleteTimerMessage() throws Exception {
String topic = "TimerTest_testPutDeleteTimerMessage";
final TimerMessageStore timerMessageStore = createTimerMessageStore(null);
final TimerMessageStore timerMessageStore = createTimerMessageStore(null , false);
timerMessageStore.load();
timerMessageStore.start(true);
@@ -372,7 +425,7 @@ public class TimerMessageStoreTest {
final String topic = "TimerTest_testStateAndRecover";
String base = StoreTestUtils.createBaseDir();
final TimerMessageStore first = createTimerMessageStore(base);
final TimerMessageStore first = createTimerMessageStore(base , false);
first.load();
first.start(true);
@@ -417,7 +470,7 @@ public class TimerMessageStoreTest {
first.getTimerWheel().flush();
first.shutdown();
final TimerMessageStore second = createTimerMessageStore(base);
final TimerMessageStore second = createTimerMessageStore(base , false);
second.debug = true;
assertTrue(second.load());
assertEquals(msgNum, second.getQueueOffset());
@@ -446,7 +499,7 @@ public class TimerMessageStoreTest {
public void testMaxDelaySec() throws Exception {
String topic = "TimerTest_testMaxDelaySec";
TimerMessageStore first = createTimerMessageStore(null);
TimerMessageStore first = createTimerMessageStore(null , false);
first.load();
first.start(true);
@@ -468,7 +521,7 @@ public class TimerMessageStoreTest {
storeConfig.setTimerRollWindowSlot(2);
String topic = "TimerTest_testRollMessage";
TimerMessageStore timerMessageStore = createTimerMessageStore(null);
TimerMessageStore timerMessageStore = createTimerMessageStore(null , false);
timerMessageStore.load();
timerMessageStore.start(true);