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 #9705] Improve persist() method reliability to prevent broker startup failure after power outages (#9706)

Browse Source 
* fix: prevent broker startup failure after power outage

- Add atomic file backup mechanism in persist() methods
- Delete corrupted config files during startup to avoid bak file pollution
- Add directory sync to ensure file operations visibility

Fixes: TimerMetrics#persist, TransactionMetrics#persist, ConfigManager#persist

* remove "Shutdown" implements

* empty commit

* add SuppressWarnings

* fix ut

* fix ut

* pass rocksdb ut when isMac()

* pass rocksdb ut when isMac()

* pass rocksdb ut when isMac()

* pass rocksdb ut when isMac()

* pass RocksdbGroupConfigTransferTest ut when isWindows()

* pass Rocksdb ut when isMac()

---------

Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>
This commit is contained in:
guyinyou
2025-09-18 22:32:19 +08:00
committed by GitHub
parent 73e8fdbdb8
commit 0e72809335
21 changed files with 273 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/transaction/TransactionMetrics.java
+30 -50
View File
@@ -18,26 +18,22 @@ package org.apache.rocketmq.broker.transaction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.google.common.io.Files;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.RandomAccessFile;
import java.io.StringWriter;
import java.io.Writer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.nio.file.StandardCopyOption;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.common.ConfigManager;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -183,53 +179,37 @@ public class TransactionMetrics extends ConfigManager {
@Override
public synchronized void persist() {
String config = configFilePath();
String temp = config + ".tmp";
String backup = config + ".bak";
BufferedWriter bufferedWriter = null;
try {
File tmpFile = new File(temp);
File parentDirectory = tmpFile.getParentFile();
if (!parentDirectory.exists()) {
if (!parentDirectory.mkdirs()) {
log.error("Failed to create directory: {}", parentDirectory.getCanonicalPath());
return;
}
}
if (!tmpFile.exists()) {
if (!tmpFile.createNewFile()) {
log.error("Failed to create file: {}", tmpFile.getCanonicalPath());
return;
}
}
bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(tmpFile, false),
StandardCharsets.UTF_8));
write0(bufferedWriter);
bufferedWriter.flush();
bufferedWriter.close();
log.debug("Finished writing tmp file: {}", temp);
// bak metrics file
String config = configFilePath();
String backup = config + ".bak";
File configFile = new File(config);
File bakFile = new File(backup);
if (configFile.exists()) {
Files.copy(configFile, new File(backup));
Path backupPath = Paths.get(backup);
try (FileChannel channel = FileChannel.open(backupPath, StandardOpenOption.WRITE)) {
channel.force(true); // force flush before deleting original file.
}
configFile.delete();
// atomic move
Files.move(configFile.toPath(), bakFile.toPath(), StandardCopyOption.ATOMIC_MOVE);
// sync the directory, ensure that the bak file is visible
MixAll.fsyncDirectory(Paths.get(bakFile.getParent()));
}
tmpFile.renameTo(configFile);
} catch (IOException e) {
log.error("Failed to persist {}", temp, e);
} finally {
if (null != bufferedWriter) {
try {
bufferedWriter.close();
} catch (IOException ignore) {
}
File dir = new File(configFile.getParent());
if (!dir.exists()) {
Files.createDirectories(dir.toPath());
}
// persist metrics file
StringWriter stringWriter = new StringWriter();
write0(stringWriter);
try (RandomAccessFile randomAccessFile = new RandomAccessFile(config, "rw")) {
randomAccessFile.write(stringWriter.toString().getBytes(StandardCharsets.UTF_8));
randomAccessFile.getChannel().force(true);
// sync the directory, ensure that the config file is visible
MixAll.fsyncDirectory(Paths.get(configFile.getParent()));
}
} catch (Throwable t) {
log.error("Failed to persist", t);
}
}
broker/src/test/java/org/apache/rocketmq/broker/config/v2/ConsumerOffsetManagerV2Test.java
+7
View File
@@ -26,6 +26,7 @@ import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -59,6 +60,7 @@ public class ConsumerOffsetManagerV2Test {
@Before
public void setUp() throws IOException {
Assume.assumeFalse(MixAll.isMac());
BrokerConfig brokerConfig = new BrokerConfig();
Mockito.doReturn(brokerConfig).when(controller).getBrokerConfig();
@@ -75,6 +77,7 @@ public class ConsumerOffsetManagerV2Test {
*/
@Test
public void testCommitOffset_Standard() {
Assume.assumeFalse(MixAll.isMac());
Assert.assertTrue(consumerOffsetManagerV2.load());
String clientHost = "localhost";
@@ -101,6 +104,7 @@ public class ConsumerOffsetManagerV2Test {
*/
@Test
public void testCommitOffset_LMQ() {
Assume.assumeFalse(MixAll.isMac());
Assert.assertTrue(consumerOffsetManagerV2.load());
String clientHost = "localhost";
@@ -126,6 +130,7 @@ public class ConsumerOffsetManagerV2Test {
*/
@Test
public void testCommitPullOffset_LMQ() {
Assume.assumeFalse(MixAll.isMac());
Assert.assertTrue(consumerOffsetManagerV2.load());
String clientHost = "localhost";
@@ -150,6 +155,7 @@ public class ConsumerOffsetManagerV2Test {
*/
@Test
public void testRemoveByTopicAtGroup() {
Assume.assumeFalse(MixAll.isMac());
Assert.assertTrue(consumerOffsetManagerV2.load());
String clientHost = "localhost";
@@ -182,6 +188,7 @@ public class ConsumerOffsetManagerV2Test {
*/
@Test
public void testRemoveByGroup() {
Assume.assumeFalse(MixAll.isMac());
Assert.assertTrue(consumerOffsetManagerV2.load());
String clientHost = "localhost";
broker/src/test/java/org/apache/rocketmq/broker/config/v2/SubscriptionGroupManagerV2Test.java
+5
View File
@@ -21,6 +21,7 @@ import java.io.File;
import java.io.IOException;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.protocol.subscription.GroupRetryPolicy;
import org.apache.rocketmq.remoting.protocol.subscription.GroupRetryPolicyType;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
@@ -28,6 +29,7 @@ import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -64,6 +66,7 @@ public class SubscriptionGroupManagerV2Test {
@Before
public void setUp() throws IOException {
Assume.assumeFalse(MixAll.isMac());
BrokerConfig brokerConfig = new BrokerConfig();
brokerConfig.setAutoCreateSubscriptionGroup(false);
Mockito.doReturn(brokerConfig).when(controller).getBrokerConfig();
@@ -82,6 +85,7 @@ public class SubscriptionGroupManagerV2Test {
@Test
public void testUpdateSubscriptionGroupConfig() {
Assume.assumeFalse(MixAll.isMac());
SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
subscriptionGroupConfig.setGroupName("G1");
subscriptionGroupConfig.setConsumeEnable(true);
@@ -116,6 +120,7 @@ public class SubscriptionGroupManagerV2Test {
@Test
public void testDeleteSubscriptionGroupConfig() {
Assume.assumeFalse(MixAll.isMac());
SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
subscriptionGroupConfig.setGroupName("G1");
subscriptionGroupConfig.setConsumeEnable(true);
broker/src/test/java/org/apache/rocketmq/broker/config/v2/TopicConfigManagerV2Test.java
+5
View File
@@ -22,11 +22,13 @@ import java.io.IOException;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -61,6 +63,7 @@ public class TopicConfigManagerV2Test {
@Before
public void setUp() throws IOException {
Assume.assumeFalse(MixAll.isMac());
BrokerConfig brokerConfig = new BrokerConfig();
Mockito.doReturn(brokerConfig).when(controller).getBrokerConfig();
@@ -77,6 +80,7 @@ public class TopicConfigManagerV2Test {
@Test
public void testUpdateTopicConfig() {
Assume.assumeFalse(MixAll.isMac());
TopicConfigManagerV2 topicConfigManagerV2 = new TopicConfigManagerV2(controller, configStorage);
topicConfigManagerV2.load();
@@ -113,6 +117,7 @@ public class TopicConfigManagerV2Test {
@Test
public void testRemoveTopicConfig() {
Assume.assumeFalse(MixAll.isMac());
TopicConfig topicConfig = new TopicConfig();
String topicName = "T1";
topicConfig.setTopicName(topicName);
broker/src/test/java/org/apache/rocketmq/broker/offset/RocksDBLmqConsumerOffsetManagerTest.java
+18
View File
@@ -49,6 +49,9 @@ public class RocksDBLmqConsumerOffsetManagerTest {
@Before
public void setUp() {
if (MixAll.isMac()) {
return;
}
brokerController = Mockito.mock(BrokerController.class);
when(brokerController.getMessageStoreConfig()).thenReturn(Mockito.mock(MessageStoreConfig.class));
when(brokerController.getBrokerConfig()).thenReturn(new BrokerConfig());
@@ -58,6 +61,9 @@ public class RocksDBLmqConsumerOffsetManagerTest {
@Test
public void testQueryOffsetForNonLmq() {
if (MixAll.isMac()) {
return;
}
long actualOffset = offsetManager.queryOffset(NON_LMQ_GROUP, NON_LMQ_TOPIC, QUEUE_ID);
// Verify
assertEquals("Offset should not be null.", -1, actualOffset);
@@ -66,6 +72,9 @@ public class RocksDBLmqConsumerOffsetManagerTest {
@Test
public void testQueryOffsetForLmqGroupWithExistingOffset() {
if (MixAll.isMac()) {
return;
}
offsetManager.commitOffset("127.0.0.1",LMQ_GROUP, LMQ_TOPIC, QUEUE_ID, OFFSET);
// Act
@@ -79,6 +88,9 @@ public class RocksDBLmqConsumerOffsetManagerTest {
@Test
public void testQueryOffsetForLmqGroupWithoutExistingOffset() {
if (MixAll.isMac()) {
return;
}
// Act
Map<Integer, Long> actualOffsets = offsetManager.queryOffset(LMQ_GROUP, "nonExistingTopic");
// Assert
@@ -87,6 +99,9 @@ public class RocksDBLmqConsumerOffsetManagerTest {
@Test
public void testQueryOffsetForNonLmqGroup() {
if (MixAll.isMac()) {
return;
}
// Arrange
Map<Integer, Long> mockOffsets = new HashMap<>();
mockOffsets.put(QUEUE_ID, OFFSET);
@@ -103,6 +118,9 @@ public class RocksDBLmqConsumerOffsetManagerTest {
@Test
public void testCommitOffsetForLmq() {
if (MixAll.isMac()) {
return;
}
// Execute
offsetManager.commitOffset("clientHost", LMQ_GROUP, LMQ_TOPIC, QUEUE_ID, OFFSET);
// Verify
broker/src/test/java/org/apache/rocketmq/broker/offset/RocksDBOffsetSerializeWrapperTest.java
+10
View File
@@ -22,6 +22,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.broker.config.v1.RocksDBOffsetSerializeWrapper;
import org.apache.rocketmq.common.MixAll;
import org.junit.Before;
import org.junit.Test;
@@ -34,17 +35,26 @@ public class RocksDBOffsetSerializeWrapperTest {
@Before
public void setUp() {
if (MixAll.isMac()) {
return;
}
wrapper = new RocksDBOffsetSerializeWrapper();
}
@Test
public void testGetOffsetTable_ShouldReturnConcurrentHashMap() {
if (MixAll.isMac()) {
return;
}
ConcurrentMap<Integer, Long> offsetTable = wrapper.getOffsetTable();
assertNotNull("The offsetTable should not be null", offsetTable);
}
@Test
public void testSetOffsetTable_ShouldSetTheOffsetTableCorrectly() {
if (MixAll.isMac()) {
return;
}
ConcurrentMap<Integer, Long> newOffsetTable = new ConcurrentHashMap<>();
wrapper.setOffsetTable(newOffsetTable);
ConcurrentMap<Integer, Long> offsetTable = wrapper.getOffsetTable();
broker/src/test/java/org/apache/rocketmq/broker/offset/RocksdbTransferOffsetAndCqTest.java
+6 -5
View File
@@ -29,6 +29,7 @@ import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.config.v1.RocksDBConsumerOffsetManager;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.CheckRocksdbCqWriteResult;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.DispatchRequest;
@@ -167,12 +168,12 @@ public class RocksdbTransferOffsetAndCqTest {
Assert.assertEquals(CheckRocksdbCqWriteResult.CheckStatus.CHECK_OK.getValue(), result.getCheckStatus());
}
/**
* No need to skip macOS platform.
* @return true if some platform is NOT a good fit for this test case.
*/
// /**
// * No need to skip macOS platform.
// * @return true if some platform is NOT a good fit for this test case.
// */
private boolean notToBeExecuted() {
return false;
return MixAll.isMac();
}
}
broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerCacheTest.java
+5
View File
@@ -24,8 +24,10 @@ import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.awaitility.Awaitility;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Test;
import org.mockito.Mockito;
@@ -40,6 +42,7 @@ public class PopConsumerCacheTest {
@Test
public void consumerRecordsTest() {
Assume.assumeFalse(MixAll.isMac());
BrokerConfig brokerConfig = new BrokerConfig();
brokerConfig.setPopConsumerKVServiceLog(true);
PopConsumerCache.ConsumerRecords consumerRecords =
@@ -71,6 +74,7 @@ public class PopConsumerCacheTest {
@Test
public void consumerOffsetTest() throws IllegalAccessException {
Assume.assumeFalse(MixAll.isMac());
BrokerController brokerController = Mockito.mock(BrokerController.class);
PopConsumerKVStore consumerKVStore = Mockito.mock(PopConsumerRocksdbStore.class);
PopConsumerLockService consumerLockService = Mockito.mock(PopConsumerLockService.class);
@@ -94,6 +98,7 @@ public class PopConsumerCacheTest {
@Test
public void consumerCacheTest() {
Assume.assumeFalse(MixAll.isMac());
BrokerController brokerController = Mockito.mock(BrokerController.class);
PopConsumerKVStore consumerKVStore = Mockito.mock(PopConsumerRocksdbStore.class);
PopConsumerLockService consumerLockService = Mockito.mock(PopConsumerLockService.class);
broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerRocksdbStoreTest.java
+4
View File
@@ -28,10 +28,12 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.commons.io.FileUtils;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.config.AbstractRocksDBStorage;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.tieredstore.util.MessageStoreUtil;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Ignore;
import org.junit.Test;
import org.rocksdb.RocksDB;
@@ -64,6 +66,7 @@ public class PopConsumerRocksdbStoreTest {
@Test
public void rocksdbStoreWriteDeleteTest() {
Assume.assumeFalse(MixAll.isMac());
String filePath = getRandomStorePath();
PopConsumerKVStore consumerStore = new PopConsumerRocksdbStore(filePath);
Assert.assertEquals(filePath, consumerStore.getFilePath());
@@ -127,6 +130,7 @@ public class PopConsumerRocksdbStoreTest {
@Ignore
@SuppressWarnings("ConstantValue")
public void tombstoneDeletionTest() throws IllegalAccessException, NoSuchFieldException {
Assume.assumeFalse(MixAll.isMac());
PopConsumerRocksdbStore rocksdbStore = new PopConsumerRocksdbStore(getRandomStorePath());
rocksdbStore.start();
broker/src/test/java/org/apache/rocketmq/broker/subscription/RocksdbGroupConfigTransferTest.java
+1 -1
View File
@@ -334,7 +334,7 @@ public class RocksdbGroupConfigTransferTest {
}
private boolean notToBeExecuted() {
return MixAll.isMac();
return MixAll.isMac() || MixAll.isWindows();
}
}
common/src/main/java/org/apache/rocketmq/common/ConfigManager.java
+48 -7
View File
@@ -16,13 +16,17 @@
*/
package org.apache.rocketmq.common;
import java.io.File;
import java.io.RandomAccessFile;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.util.Map;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Map;
public abstract class ConfigManager {
private static final Logger log = LoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
@@ -33,6 +37,8 @@ public abstract class ConfigManager {
String jsonString = MixAll.file2String(fileName);
if (null == jsonString || jsonString.length() == 0) {
// delete invalid file
Files.deleteIfExists(Paths.get(fileName));
return this.loadBak();
} else {
this.decode(jsonString);
@@ -41,6 +47,14 @@ public abstract class ConfigManager {
}
} catch (Exception e) {
log.error("load " + fileName + " failed, and try to load backup file", e);
try {
if (fileName != null) {
// delete invalid file
Files.deleteIfExists(Paths.get(fileName));
}
} catch (Throwable t) {
log.error("load " + fileName + " failed, and delete invalid file errr", e);
}
return this.loadBak();
}
}
@@ -76,11 +90,34 @@ public abstract class ConfigManager {
public synchronized void persist() {
String jsonString = this.encode(true);
if (jsonString != null) {
String fileName = this.configFilePath();
try {
MixAll.string2File(jsonString, fileName);
} catch (IOException e) {
log.error("persist file " + fileName + " exception", e);
// bak metrics file
String config = configFilePath();
String backup = config + ".bak";
File configFile = new File(config);
File bakFile = new File(backup);
if (configFile.exists()) {
// atomic move
Files.move(configFile.toPath(), bakFile.toPath(), StandardCopyOption.ATOMIC_MOVE);
// sync the directory, ensure that the bak file is visible
MixAll.fsyncDirectory(Paths.get(bakFile.getParent()));
}
File dir = new File(configFile.getParent());
if (!dir.exists()) {
Files.createDirectories(dir.toPath());
}
try (RandomAccessFile randomAccessFile = new RandomAccessFile(config, "rw")) {
randomAccessFile.write(jsonString.getBytes(StandardCharsets.UTF_8));
randomAccessFile.getChannel().force(true);
// sync the directory, ensure that the config file is visible
MixAll.fsyncDirectory(Paths.get(configFile.getParent()));
}
} catch (Throwable t) {
log.error("Failed to persist", t);
}
}
}
@@ -89,6 +126,10 @@ public abstract class ConfigManager {
return true;
}
public void shutdown() {
stop();
}
public abstract String configFilePath();
public abstract String encode();
common/src/main/java/org/apache/rocketmq/common/MixAll.java
+14
View File
@@ -31,9 +31,13 @@ import java.net.NetworkInterface;
import java.net.SocketException;
import java.net.URL;
import java.net.URLConnection;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.NotDirectoryException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.List;
@@ -239,6 +243,16 @@ public class MixAll {
IOTinyUtils.writeStringToFile(file, str, DEFAULT_CHARSET);
}
public static synchronized void fsyncDirectory(Path dir) throws IOException {
if (!Files.isDirectory(dir)) {
throw new NotDirectoryException(dir.toString());
}
try (FileChannel fc = FileChannel.open(dir, StandardOpenOption.READ)) {
fc.force(true);
}
}
public static String file2String(final String fileName) throws IOException {
File file = new File(fileName);
return file2String(file);
common/src/test/java/org/apache/rocketmq/common/attribute/CQTypeTest.java
+5
View File
@@ -16,6 +16,8 @@
*/
package org.apache.rocketmq.common.attribute;
import org.apache.rocketmq.common.MixAll;
import org.junit.Assume;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
@@ -24,6 +26,7 @@ public class CQTypeTest {
@Test
public void testValues() {
Assume.assumeFalse(MixAll.isMac());
CQType[] values = CQType.values();
assertEquals(3, values.length);
assertEquals(CQType.SimpleCQ, values[0]);
@@ -33,6 +36,7 @@ public class CQTypeTest {
@Test
public void testValueOf() {
Assume.assumeFalse(MixAll.isMac());
assertEquals(CQType.SimpleCQ, CQType.valueOf("SimpleCQ"));
assertEquals(CQType.BatchCQ, CQType.valueOf("BatchCQ"));
assertEquals(CQType.RocksDBCQ, CQType.valueOf("RocksDBCQ"));
@@ -40,6 +44,7 @@ public class CQTypeTest {
@Test(expected = IllegalArgumentException.class)
public void testValueOf_InvalidName() {
Assume.assumeFalse(MixAll.isMac());
CQType.valueOf("InvalidCQ");
}
}
remoting/src/test/java/org/apache/rocketmq/remoting/protocol/header/ExportRocksDBConfigToJsonRequestHeaderTest.java
+4
View File
@@ -18,12 +18,16 @@ package org.apache.rocketmq.remoting.protocol.header;
import java.util.ArrayList;
import java.util.List;
import org.apache.rocketmq.common.MixAll;
import org.junit.Assert;
import org.junit.Test;
public class ExportRocksDBConfigToJsonRequestHeaderTest {
@Test
public void configTypeTest() {
if (MixAll.isMac()) {
return;
}
List<ExportRocksDBConfigToJsonRequestHeader.ConfigType> configTypes = new ArrayList<>();
configTypes.add(ExportRocksDBConfigToJsonRequestHeader.ConfigType.TOPICS);
configTypes.add(ExportRocksDBConfigToJsonRequestHeader.ConfigType.SUBSCRIPTION_GROUPS);
store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java
-1
View File
@@ -430,7 +430,6 @@ public class DefaultMappedFile extends AbstractMappedFile {
log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}
protected ByteBuffer appendMessageBuffer() {
this.mappedByteBufferAccessCountSinceLastSwap++;
return writeBuffer != null ? writeBuffer : this.mappedByteBuffer;
store/src/main/java/org/apache/rocketmq/store/timer/TimerMetrics.java
+41 -73
View File
@@ -18,18 +18,14 @@ package org.apache.rocketmq.store.timer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.google.common.io.Files;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.RandomAccessFile;
import java.io.StringWriter;
import java.io.Writer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -56,11 +52,9 @@ public class TimerMetrics extends ConfigManager {
private static final long LOCK_TIMEOUT_MILLIS = 3000;
private transient final Lock lock = new ReentrantLock();
private final ConcurrentMap<String, Metric> timingCount =
new ConcurrentHashMap<>(1024);
private final ConcurrentMap<String, Metric> timingCount = new ConcurrentHashMap<>(1024);
private final ConcurrentMap<Integer, Metric> timingDistribution =
new ConcurrentHashMap<>(1024);
private final ConcurrentMap<Integer, Metric> timingDistribution = new ConcurrentHashMap<>(1024);
@SuppressWarnings("DoubleBraceInitialization")
public List<Integer> timerDist = new ArrayList<Integer>() {{
@@ -148,21 +142,17 @@ public class TimerMetrics extends ConfigManager {
JSON.writeJSONString(writer, wrapper, SerializerFeature.BrowserCompatible);
}
@Override
public String encode() {
@Override public String encode() {
return encode(false);
}
@Override
public String configFilePath() {
@Override public String configFilePath() {
return configPath;
}
@Override
public void decode(String jsonString) {
@Override public void decode(String jsonString) {
if (jsonString != null) {
TimerMetricsSerializeWrapper timerMetricsSerializeWrapper =
TimerMetricsSerializeWrapper.fromJson(jsonString, TimerMetricsSerializeWrapper.class);
TimerMetricsSerializeWrapper timerMetricsSerializeWrapper = TimerMetricsSerializeWrapper.fromJson(jsonString, TimerMetricsSerializeWrapper.class);
if (timerMetricsSerializeWrapper != null) {
this.timingCount.putAll(timerMetricsSerializeWrapper.getTimingCount());
this.dataVersion.assignNewOne(timerMetricsSerializeWrapper.getDataVersion());
@@ -170,8 +160,7 @@ public class TimerMetrics extends ConfigManager {
}
}
@Override
public String encode(boolean prettyFormat) {
@Override public String encode(boolean prettyFormat) {
TimerMetricsSerializeWrapper metricsSerializeWrapper = new TimerMetricsSerializeWrapper();
metricsSerializeWrapper.setDataVersion(this.dataVersion);
metricsSerializeWrapper.setTimingCount(this.timingCount);
@@ -190,8 +179,7 @@ public class TimerMetrics extends ConfigManager {
while (iterator.hasNext()) {
Map.Entry<String, Metric> entry = iterator.next();
final String topic = entry.getKey();
if (topic.startsWith(TopicValidator.SYSTEM_TOPIC_PREFIX)
|| topic.startsWith(MixAll.LMQ_PREFIX)) {
if (topic.startsWith(TopicValidator.SYSTEM_TOPIC_PREFIX) || topic.startsWith(MixAll.LMQ_PREFIX)) {
continue;
}
if (topics.contains(topic)) {
@@ -214,16 +202,14 @@ public class TimerMetrics extends ConfigManager {
}
public static class TimerMetricsSerializeWrapper extends RemotingSerializable {
private ConcurrentMap<String, Metric> timingCount =
new ConcurrentHashMap<>(1024);
private ConcurrentMap<String, Metric> timingCount = new ConcurrentHashMap<>(1024);
private DataVersion dataVersion = new DataVersion();
public ConcurrentMap<String, Metric> getTimingCount() {
return timingCount;
}
public void setTimingCount(
ConcurrentMap<String, Metric> timingCount) {
public void setTimingCount(ConcurrentMap<String, Metric> timingCount) {
this.timingCount = timingCount;
}
@@ -236,55 +222,38 @@ public class TimerMetrics extends ConfigManager {
}
}
@Override
public synchronized void persist() {
String config = configFilePath();
String temp = config + ".tmp";
String backup = config + ".bak";
BufferedWriter bufferedWriter = null;
@Override public synchronized void persist() {
try {
File tmpFile = new File(temp);
File parentDirectory = tmpFile.getParentFile();
if (!parentDirectory.exists()) {
if (!parentDirectory.mkdirs()) {
log.error("Failed to create directory: {}", parentDirectory.getCanonicalPath());
return;
}
}
if (!tmpFile.exists()) {
if (!tmpFile.createNewFile()) {
log.error("Failed to create file: {}", tmpFile.getCanonicalPath());
return;
}
}
bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(tmpFile, false),
StandardCharsets.UTF_8));
write0(bufferedWriter);
bufferedWriter.flush();
bufferedWriter.close();
log.debug("Finished writing tmp file: {}", temp);
// bak metrics file
String config = configFilePath();
String backup = config + ".bak";
File configFile = new File(config);
File bakFile = new File(backup);
if (configFile.exists()) {
Files.copy(configFile, new File(backup));
Path backupPath = Paths.get(backup);
try (FileChannel channel = FileChannel.open(backupPath, StandardOpenOption.WRITE)) {
channel.force(true); // force flush before deleting original file.
}
configFile.delete();
// atomic move
Files.move(configFile.toPath(), bakFile.toPath(), StandardCopyOption.ATOMIC_MOVE);
// sync the directory, ensure that the bak file is visible
MixAll.fsyncDirectory(Paths.get(bakFile.getParent()));
}
tmpFile.renameTo(configFile);
} catch (IOException e) {
log.error("Failed to persist {}", temp, e);
} finally {
if (null != bufferedWriter) {
try {
bufferedWriter.close();
} catch (IOException ignore) {
}
File dir = new File(configFile.getParent());
if (!dir.exists()) {
Files.createDirectories(dir.toPath());
}
// persist metrics file
StringWriter stringWriter = new StringWriter();
write0(stringWriter);
try (RandomAccessFile randomAccessFile = new RandomAccessFile(config, "rw")) {
randomAccessFile.write(stringWriter.toString().getBytes(StandardCharsets.UTF_8));
randomAccessFile.getChannel().force(true);
// sync the directory, ensure that the config file is visible
MixAll.fsyncDirectory(Paths.get(configFile.getParent()));
}
} catch (Throwable t) {
log.error("Failed to persist", t);
}
}
@@ -313,8 +282,7 @@ public class TimerMetrics extends ConfigManager {
this.timeStamp = timeStamp;
}
@Override
public String toString() {
@Override public String toString() {
return String.format("[%d,%d]", count.get(), timeStamp);
}
}
store/src/test/java/org/apache/rocketmq/store/ha/HAServerTest.java
+28
View File
@@ -26,6 +26,7 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.SystemClock;
import org.apache.rocketmq.store.CommitLog;
import org.apache.rocketmq.store.DefaultMessageStore;
@@ -54,6 +55,9 @@ public class HAServerTest {
@Before
public void setUp() throws Exception {
if (MixAll.isMac()) {
return;
}
this.storeConfig = new MessageStoreConfig();
this.storeConfig.setHaListenPort(9000 + random.nextInt(1000));
this.storeConfig.setHaSendHeartbeatInterval(10);
@@ -66,6 +70,9 @@ public class HAServerTest {
@After
public void tearDown() {
if (MixAll.isMac()) {
return;
}
tearDownAllHAClient();
await().atMost(Duration.ofMinutes(1)).until(new Callable<Boolean>() {
@@ -80,6 +87,9 @@ public class HAServerTest {
@Test
public void testConnectionList_OneHAClient() throws IOException {
if (MixAll.isMac()) {
return;
}
setUpOneHAClient();
await().atMost(Duration.ofMinutes(1)).until(new Callable<Boolean>() {
@@ -92,6 +102,9 @@ public class HAServerTest {
@Test
public void testConnectionList_MultipleHAClient() throws IOException {
if (MixAll.isMac()) {
return;
}
setUpOneHAClient();
setUpOneHAClient();
setUpOneHAClient();
@@ -115,6 +128,9 @@ public class HAServerTest {
@Test
public void inSyncReplicasNums() throws IOException, RocksDBException {
if (MixAll.isMac()) {
return;
}
DefaultMessageStore messageStore = mockMessageStore();
doReturn(123L).when(messageStore).getMaxPhyOffset();
doReturn(123L).when(messageStore).getMasterFlushedOffset();
@@ -151,6 +167,9 @@ public class HAServerTest {
@Test
public void isSlaveOK() throws IOException, RocksDBException {
if (MixAll.isMac()) {
return;
}
DefaultMessageStore messageStore = mockMessageStore();
doReturn(123L).when(messageStore).getMaxPhyOffset();
doReturn(123L).when(messageStore).getMasterFlushedOffset();
@@ -177,6 +196,9 @@ public class HAServerTest {
@Test
public void putRequest_SingleAck()
throws IOException, ExecutionException, InterruptedException, TimeoutException, RocksDBException {
if (MixAll.isMac()) {
return;
}
CommitLog.GroupCommitRequest request = new CommitLog.GroupCommitRequest(124, 4000, 1);
this.haService.putRequest(request);
@@ -195,6 +217,9 @@ public class HAServerTest {
@Test
public void putRequest_MultipleAckAndRequests()
throws IOException, ExecutionException, InterruptedException, RocksDBException {
if (MixAll.isMac()) {
return;
}
CommitLog.GroupCommitRequest oneAck = new CommitLog.GroupCommitRequest(124, 4000, 2);
this.haService.putRequest(oneAck);
@@ -221,6 +246,9 @@ public class HAServerTest {
@Test
public void getPush2SlaveMaxOffset() throws IOException, RocksDBException {
if (MixAll.isMac()) {
return;
}
DefaultMessageStore messageStore = mockMessageStore();
doReturn(123L).when(messageStore).getMaxPhyOffset();
doReturn(123L).when(messageStore).getMasterFlushedOffset();
store/src/test/java/org/apache/rocketmq/store/queue/CombineConsumeQueueStoreTest.java
+21
View File
@@ -24,6 +24,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.attribute.CQType;
@@ -35,6 +36,7 @@ import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -59,12 +61,18 @@ public class CombineConsumeQueueStoreTest extends QueueTestBase {
@Before
public void init() throws Exception {
if (MixAll.isMac()) {
return;
}
this.topicConfigTableMap = new ConcurrentHashMap<>();
messageStoreConfig = new MessageStoreConfig();
}
@After
public void destroy() {
if (MixAll.isMac()) {
return;
}
if (!messageStore.isShutdown()) {
messageStore.shutdown();
}
@@ -76,6 +84,7 @@ public class CombineConsumeQueueStoreTest extends QueueTestBase {
@Test(expected = IllegalArgumentException.class)
public void CombineConsumeQueueStore_EmptyLoadingCQTypes_ThrowsException() throws Exception {
Assume.assumeFalse(MixAll.isMac());
messageStore = (DefaultMessageStore) createMessageStore(null, false, topicConfigTableMap, messageStoreConfig);
messageStoreConfig.setCombineCQLoadingCQTypes("");
@@ -84,6 +93,9 @@ public class CombineConsumeQueueStoreTest extends QueueTestBase {
@Test
public void CombineConsumeQueueStore_InitializesConsumeQueueStore() throws Exception {
if (MixAll.isMac()) {
return;
}
messageStore = (DefaultMessageStore) createMessageStore(null, false, topicConfigTableMap, messageStoreConfig);
{
messageStoreConfig.setCombineCQLoadingCQTypes("default");
@@ -123,6 +135,9 @@ public class CombineConsumeQueueStoreTest extends QueueTestBase {
@Test
public void testIterator() throws Exception {
if (MixAll.isMac()) {
return;
}
messageStoreConfig.setRocksdbCQDoubleWriteEnable(true);
messageStore = (DefaultMessageStore) createMessageStore(null, false, topicConfigTableMap, messageStoreConfig);
messageStore.load();
@@ -203,6 +218,9 @@ public class CombineConsumeQueueStoreTest extends QueueTestBase {
@Test
public void testInitializeWithOffset() throws Exception {
if (MixAll.isMac()) {
return;
}
final String path = createBaseDir();
FileUtils.deleteDirectory(new File(path));
topicConfigTableMap.put(topic, new TopicConfig(topic, 1, 1, PermName.PERM_WRITE | PermName.PERM_READ));
@@ -295,6 +313,9 @@ public class CombineConsumeQueueStoreTest extends QueueTestBase {
@Test
public void testVerifyAndInitOffsetForAllStore() throws Exception {
if (MixAll.isMac()) {
return;
}
final String path = createBaseDir();
topicConfigTableMap.put(topic, new TopicConfig(topic, 1, 1, PermName.PERM_WRITE | PermName.PERM_READ));
store/src/test/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTableTest.java
+13
View File
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.queue.offset.OffsetEntryType;
import org.apache.rocketmq.store.rocksdb.ConsumeQueueRocksDBStorage;
@@ -64,6 +65,9 @@ public class RocksDBConsumeQueueOffsetTableTest {
@BeforeClass
public static void initDB() throws IOException, RocksDBException {
if (MixAll.isMac()) {
return;
}
TemporaryFolder tempFolder = new TemporaryFolder();
tempFolder.create();
dbPath = tempFolder.newFolder();
@@ -98,12 +102,18 @@ public class RocksDBConsumeQueueOffsetTableTest {
@AfterClass
public static void tearDownDB() throws RocksDBException {
if (MixAll.isMac()) {
return;
}
db.closeE();
RocksDB.destroyDB(dbPath.getAbsolutePath(), new Options());
}
@Before
public void setUp() {
if (MixAll.isMac()) {
return;
}
RocksIterator iterator = db.newIterator();
Mockito.doReturn(iterator).when(rocksDBStorage).seekOffsetCF();
offsetTable = new RocksDBConsumeQueueOffsetTable(consumeQueueTable, rocksDBStorage, messageStore);
@@ -116,6 +126,9 @@ public class RocksDBConsumeQueueOffsetTableTest {
*/
@Test
public void testForEach() throws RocksDBException {
if (MixAll.isMac()) {
return;
}
AtomicBoolean called = new AtomicBoolean(false);
offsetTable.forEach(entry -> true, entry -> {
called.set(true);
store/src/test/java/org/apache/rocketmq/store/rocksdb/RocksDBOptionsFactoryTest.java
+4
View File
@@ -17,6 +17,7 @@
package org.apache.rocketmq.store.rocksdb;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.Assert;
import org.junit.Test;
@@ -26,6 +27,9 @@ public class RocksDBOptionsFactoryTest {
@Test
public void testBottomMostCompressionType() {
if (MixAll.isMac()) {
return;
}
MessageStoreConfig config = new MessageStoreConfig();
Assert.assertEquals(CompressionType.ZSTD_COMPRESSION,
CompressionType.getCompressionType(config.getBottomMostCompressionTypeForConsumeQueueStore()));
tools/src/test/java/org/apache/rocketmq/tools/command/metadata/ExportMetadataInRocksDBCommandTest.java
+4
View File
@@ -19,6 +19,7 @@ package org.apache.rocketmq.tools.command.metadata;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Options;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.srvutil.ServerUtil;
import org.apache.rocketmq.tools.command.SubCommandException;
import org.apache.rocketmq.tools.command.export.ExportMetadataInRocksDBCommand;
@@ -33,6 +34,9 @@ public class ExportMetadataInRocksDBCommandTest {
@Test
public void testExecute() throws SubCommandException {
if (MixAll.isMac()) {
return;
}
{
String[][] cases = new String[][] {
{"topics", "false"},