From 68aa58da82059d7c894f923767256217c0df7097 Mon Sep 17 00:00:00 2001 From: chenhoudao Date: Mon, 21 Jan 2019 19:45:30 +0800 Subject: [PATCH] recover unit test file --- .../store/DefaultMessageStoreTest.java | 298 ++++++++++++++++++ 1 file changed, 298 insertions(+) create mode 100644 store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java diff --git a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java new file mode 100644 index 00000000..57b6999c --- /dev/null +++ b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java @@ -0,0 +1,298 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.store; + +import java.io.File; +import java.io.RandomAccessFile; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.nio.MappedByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.channels.OverlappingFileLockException; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.rocketmq.common.BrokerConfig; +import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.store.config.FlushDiskType; +import org.apache.rocketmq.store.config.MessageStoreConfig; +import org.apache.rocketmq.store.config.StorePathConfigHelper; +import org.junit.After; +import org.apache.rocketmq.store.stats.BrokerStatsManager; +import org.junit.Before; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertTrue; + +public class DefaultMessageStoreTest { + private final String StoreMessage = "Once, there was a chance for me!"; + private int QUEUE_TOTAL = 100; + private AtomicInteger QueueId = new AtomicInteger(0); + private SocketAddress BornHost; + private SocketAddress StoreHost; + private byte[] MessageBody; + private MessageStore messageStore; + + @Before + public void init() throws Exception { + StoreHost = new InetSocketAddress(InetAddress.getLocalHost(), 8123); + BornHost = new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0); + + messageStore = buildMessageStore(); + boolean load = messageStore.load(); + assertTrue(load); + messageStore.start(); + } + + @Test(expected = OverlappingFileLockException.class) + public void test_repate_restart() throws Exception { + QUEUE_TOTAL = 1; + MessageBody = StoreMessage.getBytes(); + + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + messageStoreConfig.setMapedFileSizeCommitLog(1024 * 8); + messageStoreConfig.setMapedFileSizeConsumeQueue(1024 * 4); + messageStoreConfig.setMaxHashSlotNum(100); + messageStoreConfig.setMaxIndexNum(100 * 10); + MessageStore master = new DefaultMessageStore(messageStoreConfig, null, new MyMessageArrivingListener(), new BrokerConfig()); + + boolean load = master.load(); + assertTrue(load); + + try { + master.start(); + master.start(); + } finally { + master.shutdown(); + master.destroy(); + } + } + + @After + public void destory() { + messageStore.shutdown(); + messageStore.destroy(); + + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + File file = new File(messageStoreConfig.getStorePathRootDir()); + UtilAll.deleteFile(file); + } + + private MessageStore buildMessageStore() throws Exception { + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + messageStoreConfig.setMapedFileSizeCommitLog(1024 * 1024 * 10); + messageStoreConfig.setMapedFileSizeConsumeQueue(1024 * 1024 * 10); + messageStoreConfig.setMaxHashSlotNum(10000); + messageStoreConfig.setMaxIndexNum(100 * 100); + messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH); + messageStoreConfig.setFlushIntervalConsumeQueue(1); + return new DefaultMessageStore(messageStoreConfig, new BrokerStatsManager("simpleTest"), new MyMessageArrivingListener(), new BrokerConfig()); + } + + @Test + public void testWriteAndRead() { + long totalMsgs = 10; + QUEUE_TOTAL = 1; + MessageBody = StoreMessage.getBytes(); + for (long i = 0; i < totalMsgs; i++) { + messageStore.putMessage(buildMessage()); + } + + for (long i = 0; i < totalMsgs; i++) { + GetMessageResult result = messageStore.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null); + assertThat(result).isNotNull(); + result.release(); + } + verifyThatMasterIsFunctional(totalMsgs, messageStore); + } + + private MessageExtBrokerInner buildMessage() { + MessageExtBrokerInner msg = new MessageExtBrokerInner(); + msg.setTopic("FooBar"); + msg.setTags("TAG1"); + msg.setKeys("Hello"); + msg.setBody(MessageBody); + msg.setKeys(String.valueOf(System.currentTimeMillis())); + msg.setQueueId(Math.abs(QueueId.getAndIncrement()) % QUEUE_TOTAL); + msg.setSysFlag(0); + msg.setBornTimestamp(System.currentTimeMillis()); + msg.setStoreHost(StoreHost); + msg.setBornHost(BornHost); + return msg; + } + + private void verifyThatMasterIsFunctional(long totalMsgs, MessageStore master) { + for (long i = 0; i < totalMsgs; i++) { + master.putMessage(buildMessage()); + } + + for (long i = 0; i < totalMsgs; i++) { + GetMessageResult result = master.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null); + assertThat(result).isNotNull(); + result.release(); + + } + } + + @Test + public void testPullSize() throws Exception { + String topic = "pullSizeTopic"; + + for (int i = 0; i < 32; i++) { + MessageExtBrokerInner messageExtBrokerInner = buildMessage(); + messageExtBrokerInner.setTopic(topic); + messageExtBrokerInner.setQueueId(0); + messageStore.putMessage(messageExtBrokerInner); + } + // wait for consume queue build + // the sleep time should be great than consume queue flush interval + Thread.sleep(100); + String group = "simple"; + GetMessageResult getMessageResult32 = messageStore.getMessage(group, topic, 0, 0, 32, null); + assertThat(getMessageResult32.getMessageBufferList().size()).isEqualTo(32); + + GetMessageResult getMessageResult20 = messageStore.getMessage(group, topic, 0, 0, 20, null); + assertThat(getMessageResult20.getMessageBufferList().size()).isEqualTo(20); + + GetMessageResult getMessageResult45 = messageStore.getMessage(group, topic, 0, 0, 10, null); + assertThat(getMessageResult45.getMessageBufferList().size()).isEqualTo(10); + } + + @Test + public void testRecover() throws Exception { + String topic = "recoverTopic"; + MessageBody = StoreMessage.getBytes(); + for (int i = 0; i < 100; i++) { + MessageExtBrokerInner messageExtBrokerInner = buildMessage(); + messageExtBrokerInner.setTopic(topic); + messageExtBrokerInner.setQueueId(0); + messageStore.putMessage(messageExtBrokerInner); + } + + Thread.sleep(100);//wait for build consumer queue + long maxPhyOffset = messageStore.getMaxPhyOffset(); + long maxCqOffset = messageStore.getMaxOffsetInQueue(topic, 0); + + //1.just reboot + messageStore.shutdown(); + messageStore = buildMessageStore(); + boolean load = messageStore.load(); + assertTrue(load); + messageStore.start(); + assertTrue(maxPhyOffset == messageStore.getMaxPhyOffset()); + assertTrue(maxCqOffset == messageStore.getMaxOffsetInQueue(topic, 0)); + + //2.damage commitlog and reboot normal + for (int i = 0; i < 100; i++) { + MessageExtBrokerInner messageExtBrokerInner = buildMessage(); + messageExtBrokerInner.setTopic(topic); + messageExtBrokerInner.setQueueId(0); + messageStore.putMessage(messageExtBrokerInner); + } + Thread.sleep(100); + long secondLastPhyOffset = messageStore.getMaxPhyOffset(); + long secondLastCqOffset = messageStore.getMaxOffsetInQueue(topic, 0); + + MessageExtBrokerInner messageExtBrokerInner = buildMessage(); + messageExtBrokerInner.setTopic(topic); + messageExtBrokerInner.setQueueId(0); + messageStore.putMessage(messageExtBrokerInner); + + messageStore.shutdown(); + + //damage last message + damageCommitlog(secondLastPhyOffset); + + //reboot + messageStore = buildMessageStore(); + load = messageStore.load(); + assertTrue(load); + messageStore.start(); + assertTrue(secondLastPhyOffset == messageStore.getMaxPhyOffset()); + assertTrue(secondLastCqOffset == messageStore.getMaxOffsetInQueue(topic, 0)); + + //3.damage commitlog and reboot abnormal + for (int i = 0; i < 100; i++) { + messageExtBrokerInner = buildMessage(); + messageExtBrokerInner.setTopic(topic); + messageExtBrokerInner.setQueueId(0); + messageStore.putMessage(messageExtBrokerInner); + } + Thread.sleep(100); + secondLastPhyOffset = messageStore.getMaxPhyOffset(); + secondLastCqOffset = messageStore.getMaxOffsetInQueue(topic, 0); + + messageExtBrokerInner = buildMessage(); + messageExtBrokerInner.setTopic(topic); + messageExtBrokerInner.setQueueId(0); + messageStore.putMessage(messageExtBrokerInner); + messageStore.shutdown(); + + //damage last message + damageCommitlog(secondLastPhyOffset); + //add abort file + String fileName = StorePathConfigHelper.getAbortFile(((DefaultMessageStore) messageStore).getMessageStoreConfig().getStorePathRootDir()); + File file = new File(fileName); + MappedFile.ensureDirOK(file.getParent()); + file.createNewFile(); + + messageStore = buildMessageStore(); + load = messageStore.load(); + assertTrue(load); + messageStore.start(); + assertTrue(secondLastPhyOffset == messageStore.getMaxPhyOffset()); + assertTrue(secondLastCqOffset == messageStore.getMaxOffsetInQueue(topic, 0)); + + //message write again + for (int i = 0; i < 100; i++) { + messageExtBrokerInner = buildMessage(); + messageExtBrokerInner.setTopic(topic); + messageExtBrokerInner.setQueueId(0); + messageStore.putMessage(messageExtBrokerInner); + } + } + + private void damageCommitlog(long offset) throws Exception { + MessageStoreConfig messageStoreConfig = new MessageStoreConfig(); + File file = new File(messageStoreConfig.getStorePathCommitLog() + File.separator + "00000000000000000000"); + + FileChannel fileChannel = new RandomAccessFile(file, "rw").getChannel(); + MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, 1024 * 1024 * 10); + + int bodyLen = mappedByteBuffer.getInt((int) offset + 84); + int topicLenIndex = (int) offset + 84 + bodyLen + 4; + mappedByteBuffer.position(topicLenIndex); + mappedByteBuffer.putInt(0); + mappedByteBuffer.putInt(0); + mappedByteBuffer.putInt(0); + mappedByteBuffer.putInt(0); + + mappedByteBuffer.force(); + fileChannel.force(true); + fileChannel.close(); + } + + private class MyMessageArrivingListener implements MessageArrivingListener { + @Override + public void arriving(String topic, int queueId, long logicOffset, long tagsCode, long msgStoreTime, + byte[] filterBitMap, Map properties) { + } + } +} -- GitLab