rocketmq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yu...@apache.org
Subject [06/43] incubator-rocketmq git commit: Finish code dump. Reviewed by: @yukon @vongosling @stevenschew @vintagewang @lollipop @zander
Date Mon, 19 Dec 2016 09:40:23 GMT
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-store/src/main/java/com/alibaba/rocketmq/store/stats/BrokerStatsManager.java
----------------------------------------------------------------------
diff --git a/rocketmq-store/src/main/java/com/alibaba/rocketmq/store/stats/BrokerStatsManager.java b/rocketmq-store/src/main/java/com/alibaba/rocketmq/store/stats/BrokerStatsManager.java
new file mode 100644
index 0000000..60b3fff
--- /dev/null
+++ b/rocketmq-store/src/main/java/com/alibaba/rocketmq/store/stats/BrokerStatsManager.java
@@ -0,0 +1,220 @@
+/**
+ * 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 com.alibaba.rocketmq.store.stats;
+
+import com.alibaba.rocketmq.common.ThreadFactoryImpl;
+import com.alibaba.rocketmq.common.constant.LoggerName;
+import com.alibaba.rocketmq.common.stats.MomentStatsItemSet;
+import com.alibaba.rocketmq.common.stats.StatsItem;
+import com.alibaba.rocketmq.common.stats.StatsItemSet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+
+public class BrokerStatsManager {
+
+    public static final String TOPIC_PUT_NUMS = "TOPIC_PUT_NUMS";
+    public static final String TOPIC_PUT_SIZE = "TOPIC_PUT_SIZE";
+    public static final String GROUP_GET_NUMS = "GROUP_GET_NUMS";
+    public static final String GROUP_GET_SIZE = "GROUP_GET_SIZE";
+    public static final String SNDBCK_PUT_NUMS = "SNDBCK_PUT_NUMS";
+    public static final String BROKER_PUT_NUMS = "BROKER_PUT_NUMS";
+    public static final String BROKER_GET_NUMS = "BROKER_GET_NUMS";
+    public static final String GROUP_GET_FROM_DISK_NUMS = "GROUP_GET_FROM_DISK_NUMS";
+    public static final String GROUP_GET_FROM_DISK_SIZE = "GROUP_GET_FROM_DISK_SIZE";
+    public static final String BROKER_GET_FROM_DISK_NUMS = "BROKER_GET_FROM_DISK_NUMS";
+    public static final String BROKER_GET_FROM_DISK_SIZE = "BROKER_GET_FROM_DISK_SIZE";
+    // For commercial
+    public static final String COMMERCIAL_SEND_TIMES = "COMMERCIAL_SEND_TIMES";
+    public static final String COMMERCIAL_SNDBCK_TIMES = "COMMERCIAL_SNDBCK_TIMES";
+    public static final String COMMERCIAL_RCV_TIMES = "COMMERCIAL_RCV_TIMES";
+    public static final String COMMERCIAL_RCV_EPOLLS = "COMMERCIAL_RCV_EPOLLS";
+    public static final String COMMERCIAL_SEND_SIZE = "COMMERCIAL_SEND_SIZE";
+    public static final String COMMERCIAL_RCV_SIZE = "COMMERCIAL_RCV_SIZE";
+    public static final String COMMERCIAL_PERM_FAILURES = "COMMERCIAL_PERM_FAILURES";
+    public static final String COMMERCIAL_OWNER = "Owner";
+    // Message Size limit for one api-calling count.
+    public static final double SIZE_PER_COUNT = 64 * 1024;
+
+    public static final String GROUP_GET_FALL_SIZE = "GROUP_GET_FALL_SIZE";
+    public static final String GROUP_GET_FALL_TIME = "GROUP_GET_FALL_TIME";
+    // Pull Message Latency
+    public static final String GROUP_GET_LATENCY = "GROUP_GET_LATENCY";
+
+    /**
+     * read disk follow stats
+     */
+    private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_STATS_LOGGER_NAME);
+    private static final Logger COMMERCIAL_LOG = LoggerFactory.getLogger(LoggerName.COMMERCIAL_LOGGER_NAME);
+    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
+            "BrokerStatsThread"));
+    private final ScheduledExecutorService commercialExecutor = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
+            "CommercialStatsThread"));
+    private final HashMap<String, StatsItemSet> statsTable = new HashMap<String, StatsItemSet>();
+    private final String clusterName;
+    private final MomentStatsItemSet momentStatsItemSetFallSize = new MomentStatsItemSet(GROUP_GET_FALL_SIZE, scheduledExecutorService, log);
+    private final MomentStatsItemSet momentStatsItemSetFallTime = new MomentStatsItemSet(GROUP_GET_FALL_TIME, scheduledExecutorService, log);
+
+    public BrokerStatsManager(String clusterName) {
+        this.clusterName = clusterName;
+
+        this.statsTable.put(TOPIC_PUT_NUMS, new StatsItemSet(TOPIC_PUT_NUMS, this.scheduledExecutorService, log));
+        this.statsTable.put(TOPIC_PUT_SIZE, new StatsItemSet(TOPIC_PUT_SIZE, this.scheduledExecutorService, log));
+        this.statsTable.put(GROUP_GET_NUMS, new StatsItemSet(GROUP_GET_NUMS, this.scheduledExecutorService, log));
+        this.statsTable.put(GROUP_GET_SIZE, new StatsItemSet(GROUP_GET_SIZE, this.scheduledExecutorService, log));
+        this.statsTable.put(GROUP_GET_LATENCY, new StatsItemSet(GROUP_GET_LATENCY, this.scheduledExecutorService, log));
+        this.statsTable.put(SNDBCK_PUT_NUMS, new StatsItemSet(SNDBCK_PUT_NUMS, this.scheduledExecutorService, log));
+        this.statsTable.put(BROKER_PUT_NUMS, new StatsItemSet(BROKER_PUT_NUMS, this.scheduledExecutorService, log));
+        this.statsTable.put(BROKER_GET_NUMS, new StatsItemSet(BROKER_GET_NUMS, this.scheduledExecutorService, log));
+        this.statsTable.put(GROUP_GET_FROM_DISK_NUMS, new StatsItemSet(GROUP_GET_FROM_DISK_NUMS, this.scheduledExecutorService, log));
+        this.statsTable.put(GROUP_GET_FROM_DISK_SIZE, new StatsItemSet(GROUP_GET_FROM_DISK_SIZE, this.scheduledExecutorService, log));
+        this.statsTable.put(BROKER_GET_FROM_DISK_NUMS, new StatsItemSet(BROKER_GET_FROM_DISK_NUMS, this.scheduledExecutorService, log));
+        this.statsTable.put(BROKER_GET_FROM_DISK_SIZE, new StatsItemSet(BROKER_GET_FROM_DISK_SIZE, this.scheduledExecutorService, log));
+
+
+        this.statsTable.put(COMMERCIAL_SEND_TIMES, new StatsItemSet(COMMERCIAL_SEND_TIMES, this.commercialExecutor, COMMERCIAL_LOG));
+        this.statsTable.put(COMMERCIAL_RCV_TIMES, new StatsItemSet(COMMERCIAL_RCV_TIMES, this.commercialExecutor, COMMERCIAL_LOG));
+        this.statsTable.put(COMMERCIAL_SEND_SIZE, new StatsItemSet(COMMERCIAL_SEND_SIZE, this.commercialExecutor, COMMERCIAL_LOG));
+        this.statsTable.put(COMMERCIAL_RCV_SIZE, new StatsItemSet(COMMERCIAL_RCV_SIZE, this.commercialExecutor, COMMERCIAL_LOG));
+        this.statsTable.put(COMMERCIAL_RCV_EPOLLS, new StatsItemSet(COMMERCIAL_RCV_EPOLLS, this.commercialExecutor, COMMERCIAL_LOG));
+        this.statsTable.put(COMMERCIAL_SNDBCK_TIMES, new StatsItemSet(COMMERCIAL_SNDBCK_TIMES, this.commercialExecutor, COMMERCIAL_LOG));
+        this.statsTable.put(COMMERCIAL_PERM_FAILURES, new StatsItemSet(COMMERCIAL_PERM_FAILURES, this.commercialExecutor, COMMERCIAL_LOG));
+    }
+
+    public MomentStatsItemSet getMomentStatsItemSetFallSize() {
+        return momentStatsItemSetFallSize;
+    }
+
+    public MomentStatsItemSet getMomentStatsItemSetFallTime() {
+        return momentStatsItemSetFallTime;
+    }
+
+    public void start() {
+    }
+
+    public void shutdown() {
+        this.scheduledExecutorService.shutdown();
+    }
+
+    public StatsItem getStatsItem(final String statsName, final String statsKey) {
+        try {
+            return this.statsTable.get(statsName).getStatsItem(statsKey);
+        } catch (Exception e) {
+        }
+
+        return null;
+    }
+
+    public void incTopicPutNums(final String topic) {
+        this.statsTable.get(TOPIC_PUT_NUMS).addValue(topic, 1, 1);
+    }
+
+    public void incTopicPutSize(final String topic, final int size) {
+        this.statsTable.get(TOPIC_PUT_SIZE).addValue(topic, size, 1);
+    }
+
+    public void incGroupGetNums(final String group, final String topic, final int incValue) {
+        final String statsKey = buildStatsKey(topic, group);
+        this.statsTable.get(GROUP_GET_NUMS).addValue(statsKey, incValue, 1);
+    }
+
+    public String buildStatsKey(String topic, String group) {
+        StringBuffer strBuilder = new StringBuffer();
+        strBuilder.append(topic);
+        strBuilder.append("@");
+        strBuilder.append(group);
+        return strBuilder.toString();
+    }
+
+    public void incGroupGetSize(final String group, final String topic, final int incValue) {
+        final String statsKey = buildStatsKey(topic, group);
+        this.statsTable.get(GROUP_GET_SIZE).addValue(statsKey, incValue, 1);
+    }
+
+    public void incGroupGetLatency(final String group, final String topic, final int queueId, final int incValue) {
+        final String statsKey = String.format("%d@%s@%s", queueId, topic, group);
+        this.statsTable.get(GROUP_GET_LATENCY).addValue(statsKey, incValue, 1);
+    }
+
+
+    public void incBrokerPutNums() {
+        this.statsTable.get(BROKER_PUT_NUMS).getAndCreateStatsItem(this.clusterName).getValue().incrementAndGet();
+    }
+
+
+    public void incBrokerGetNums(final int incValue) {
+        this.statsTable.get(BROKER_GET_NUMS).getAndCreateStatsItem(this.clusterName).getValue().addAndGet(incValue);
+    }
+
+
+    public void incSendBackNums(final String group, final String topic) {
+        final String statsKey = buildStatsKey(topic, group);
+        this.statsTable.get(SNDBCK_PUT_NUMS).addValue(statsKey, 1, 1);
+    }
+
+
+    public double tpsGroupGetNums(final String group, final String topic) {
+        final String statsKey = buildStatsKey(topic, group);
+        return this.statsTable.get(GROUP_GET_NUMS).getStatsDataInMinute(statsKey).getTps();
+    }
+
+
+    public void recordDiskFallBehindTime(final String group, final String topic, final int queueId, final long fallBehind) {
+        final String statsKey = String.format("%d@%s@%s", queueId, topic, group);
+        this.momentStatsItemSetFallTime.getAndCreateStatsItem(statsKey).getValue().set(fallBehind);
+    }
+
+
+    public void recordDiskFallBehindSize(final String group, final String topic, final int queueId, final long fallBehind) {
+        final String statsKey = String.format("%d@%s@%s", queueId, topic, group);
+        this.momentStatsItemSetFallSize.getAndCreateStatsItem(statsKey).getValue().set(fallBehind);
+    }
+
+    public void incCommercialValue(final String key, final String owner, final String group,
+                                   final String topic, final String type, final int incValue) {
+        final String statsKey = buildCommercialStatsKey(owner, topic, group, type);
+        this.statsTable.get(key).addValue(statsKey, incValue, 1);
+    }
+
+    public String buildCommercialStatsKey(String owner, String topic, String group, String type) {
+        StringBuffer strBuilder = new StringBuffer();
+        strBuilder.append(owner);
+        strBuilder.append("@");
+        strBuilder.append(topic);
+        strBuilder.append("@");
+        strBuilder.append(group);
+        strBuilder.append("@");
+        strBuilder.append(type);
+        return strBuilder.toString();
+    }
+
+
+    public enum StatsType {
+        SEND_SUCCESS,
+        SEND_FAILURE,
+        SEND_BACK,
+        SEND_TIMER,
+        SEND_TRANSACTION,
+        RCV_SUCCESS,
+        RCV_EPOLLS,
+        PERM_FAILURE
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-store/src/main/java/com/alibaba/rocketmq/store/util/LibC.java
----------------------------------------------------------------------
diff --git a/rocketmq-store/src/main/java/com/alibaba/rocketmq/store/util/LibC.java b/rocketmq-store/src/main/java/com/alibaba/rocketmq/store/util/LibC.java
new file mode 100644
index 0000000..7a3b37c
--- /dev/null
+++ b/rocketmq-store/src/main/java/com/alibaba/rocketmq/store/util/LibC.java
@@ -0,0 +1,54 @@
+/**
+ * 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 com.alibaba.rocketmq.store.util;
+
+import com.sun.jna.Library;
+import com.sun.jna.Native;
+import com.sun.jna.NativeLong;
+import com.sun.jna.Platform;
+import com.sun.jna.Pointer;
+
+
+public interface LibC extends Library {
+    LibC INSTANCE = (LibC) Native.loadLibrary(Platform.isWindows() ? "msvcrt" : "c", LibC.class);
+
+    int MADV_WILLNEED = 3;
+    int MADV_DONTNEED = 4;
+
+    int MCL_CURRENT = 1;
+    int MCL_FUTURE = 2;
+    int MCL_ONFAULT = 4;
+
+    /* sync memory asynchronously */
+    int MS_ASYNC = 0x0001;
+    /* invalidate mappings & caches */
+    int MS_INVALIDATE = 0x0002;
+    /* synchronous memory sync */
+    int MS_SYNC = 0x0004;
+
+    int mlock(Pointer var1, NativeLong var2);
+
+    int munlock(Pointer var1, NativeLong var2);
+
+    int madvise(Pointer var1, NativeLong var2, int var3);
+
+    Pointer memset(Pointer p, int v, long len);
+
+    int mlockall(int flags);
+
+    int msync(Pointer p, NativeLong length, int flags);
+}

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/DefaultMessageStoreTest.java
----------------------------------------------------------------------
diff --git a/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/DefaultMessageStoreTest.java b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/DefaultMessageStoreTest.java
new file mode 100644
index 0000000..ac577c7
--- /dev/null
+++ b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/DefaultMessageStoreTest.java
@@ -0,0 +1,154 @@
+/**
+ * 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 com.alibaba.rocketmq.store;
+
+import com.alibaba.rocketmq.store.config.FlushDiskType;
+import com.alibaba.rocketmq.store.config.MessageStoreConfig;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertTrue;
+
+
+/**
+ * @author shijia.wxr
+ */
+public class DefaultMessageStoreTest {
+    private static final String StoreMessage = "Once, there was a chance for me!";
+
+    private static int QUEUE_TOTAL = 100;
+
+    private static AtomicInteger QueueId = new AtomicInteger(0);
+
+    private static SocketAddress BornHost;
+
+    private static SocketAddress StoreHost;
+
+    private static byte[] MessageBody;
+
+    @BeforeClass
+    public static void setUpBeforeClass() throws Exception {
+        StoreHost = new InetSocketAddress(InetAddress.getLocalHost(), 8123);
+        BornHost = new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0);
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() throws Exception {
+    }
+
+    @Test
+    public void test_write_read() throws Exception {
+        System.out.println("================================================================");
+        long totalMsgs = 100;
+        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, null, null);
+
+        boolean load = master.load();
+        assertTrue(load);
+
+        master.start();
+        for (long i = 0; i < totalMsgs; i++) {
+            PutMessageResult result = master.putMessage(buildMessage());
+            System.out.println(i + "\t" + result.getAppendMessageResult().getMsgId());
+        }
+
+        for (long i = 0; i < totalMsgs; i++) {
+            try {
+                GetMessageResult result = master.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
+                if (result == null) {
+                    System.out.println("result == null " + i);
+                }
+                assertTrue(result != null);
+                result.release();
+                System.out.println("read " + i + " OK");
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+
+        }
+        master.shutdown();
+        master.destroy();
+        System.out.println("================================================================");
+    }
+
+    public MessageExtBrokerInner buildMessage() {
+        MessageExtBrokerInner msg = new MessageExtBrokerInner();
+        msg.setTopic("AAA");
+        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(4);
+        msg.setBornTimestamp(System.currentTimeMillis());
+        msg.setStoreHost(StoreHost);
+        msg.setBornHost(BornHost);
+        return msg;
+    }
+
+    @Test
+    public void test_group_commit() throws Exception {
+        System.out.println("================================================================");
+        long totalMsgs = 100;
+        QUEUE_TOTAL = 1;
+        MessageBody = StoreMessage.getBytes();
+        MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+        messageStoreConfig.setMapedFileSizeCommitLog(1024 * 8);
+        messageStoreConfig.setFlushDiskType(FlushDiskType.SYNC_FLUSH);
+        MessageStore master = new DefaultMessageStore(messageStoreConfig, null, null, null);
+        boolean load = master.load();
+        assertTrue(load);
+
+        master.start();
+        for (long i = 0; i < totalMsgs; i++) {
+            PutMessageResult result = master.putMessage(buildMessage());
+            System.out.println(i + "\t" + result.getAppendMessageResult().getMsgId());
+        }
+
+        for (long i = 0; i < totalMsgs; i++) {
+            try {
+                GetMessageResult result = master.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
+                if (result == null) {
+                    System.out.println("result == null " + i);
+                }
+                assertTrue(result != null);
+                result.release();
+                System.out.println("read " + i + " OK");
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+
+        }
+        master.shutdown();
+        master.destroy();
+        System.out.println("================================================================");
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/MappedFileQueueTest.java
----------------------------------------------------------------------
diff --git a/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/MappedFileQueueTest.java b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/MappedFileQueueTest.java
new file mode 100644
index 0000000..ce7666f
--- /dev/null
+++ b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/MappedFileQueueTest.java
@@ -0,0 +1,197 @@
+/**
+ * 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.
+ */
+
+/**
+ * $Id: MappedFileQueueTest.java 1831 2013-05-16 01:39:51Z shijia.wxr $
+ */
+package com.alibaba.rocketmq.store;
+
+import org.junit.*;
+
+import static org.junit.Assert.*;
+
+
+public class MappedFileQueueTest {
+
+    // private static final String StoreMessage =
+    // "Once, there was a chance for me! but I did not treasure it. if";
+
+    @BeforeClass
+    public static void setUpBeforeClass() throws Exception {
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() throws Exception {
+    }
+
+    @Before
+    public void setUp() throws Exception {
+    }
+
+    @After
+    public void tearDown() throws Exception {
+    }
+
+    @Test
+    public void test_getLastMapedFile() {
+        final String fixedMsg = "0123456789abcdef";
+        System.out.println("================================================================");
+        MappedFileQueue mappedFileQueue =
+                new MappedFileQueue("target/unit_test_store/a/", 1024, null);
+
+        for (int i = 0; i < 1024; i++) {
+            MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
+            assertTrue(mappedFile != null);
+            boolean result = mappedFile.appendMessage(fixedMsg.getBytes());
+            if (!result) {
+                System.out.println("appendMessage " + i);
+            }
+            assertTrue(result);
+        }
+
+        mappedFileQueue.shutdown(1000);
+        mappedFileQueue.destroy();
+        System.out.println("MappedFileQueue.getLastMappedFile() OK");
+    }
+
+
+    @Test
+    public void test_findMapedFileByOffset() {
+        final String fixedMsg = "abcd";
+        System.out.println("================================================================");
+        MappedFileQueue mappedFileQueue =
+                new MappedFileQueue("target/unit_test_store/b/", 1024, null);
+
+        for (int i = 0; i < 1024; i++) {
+            MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
+            assertTrue(mappedFile != null);
+            boolean result = mappedFile.appendMessage(fixedMsg.getBytes());
+            // System.out.println("appendMessage " + bytes);
+            assertTrue(result);
+        }
+
+        MappedFile mappedFile = mappedFileQueue.findMappedFileByOffset(0);
+        assertTrue(mappedFile != null);
+        assertEquals(mappedFile.getFileFromOffset(), 0);
+        System.out.println(mappedFile.getFileFromOffset());
+
+        mappedFile = mappedFileQueue.findMappedFileByOffset(100);
+        assertTrue(mappedFile != null);
+        assertEquals(mappedFile.getFileFromOffset(), 0);
+        System.out.println(mappedFile.getFileFromOffset());
+
+        mappedFile = mappedFileQueue.findMappedFileByOffset(1024);
+        assertTrue(mappedFile != null);
+        assertEquals(mappedFile.getFileFromOffset(), 1024);
+        System.out.println(mappedFile.getFileFromOffset());
+
+        mappedFile = mappedFileQueue.findMappedFileByOffset(1024 + 100);
+        assertTrue(mappedFile != null);
+        assertEquals(mappedFile.getFileFromOffset(), 1024);
+        System.out.println(mappedFile.getFileFromOffset());
+
+        mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 2);
+        assertTrue(mappedFile != null);
+        assertEquals(mappedFile.getFileFromOffset(), 1024 * 2);
+        System.out.println(mappedFile.getFileFromOffset());
+
+        mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 2 + 100);
+        assertTrue(mappedFile != null);
+        assertEquals(mappedFile.getFileFromOffset(), 1024 * 2);
+        System.out.println(mappedFile.getFileFromOffset());
+
+        mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 4);
+        assertTrue(mappedFile == null);
+
+        mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 4 + 100);
+        assertTrue(mappedFile == null);
+
+        mappedFileQueue.shutdown(1000);
+        mappedFileQueue.destroy();
+        System.out.println("MappedFileQueue.findMappedFileByOffset() OK");
+    }
+
+    @Test
+    public void test_commit() {
+        final String fixedMsg = "0123456789abcdef";
+        System.out.println("================================================================");
+        MappedFileQueue mappedFileQueue =
+                new MappedFileQueue("target/unit_test_store/c/", 1024, null);
+
+        for (int i = 0; i < 1024; i++) {
+            MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
+            assertTrue(mappedFile != null);
+            boolean result = mappedFile.appendMessage(fixedMsg.getBytes());
+            assertTrue(result);
+        }
+
+        boolean result = mappedFileQueue.flush(0);
+        assertFalse(result);
+        assertEquals(1024 * 1, mappedFileQueue.getFlushedWhere());
+        System.out.println("1 " + result + " " + mappedFileQueue.getFlushedWhere());
+
+        result = mappedFileQueue.flush(0);
+        assertFalse(result);
+        assertEquals(1024 * 2, mappedFileQueue.getFlushedWhere());
+        System.out.println("2 " + result + " " + mappedFileQueue.getFlushedWhere());
+
+        result = mappedFileQueue.flush(0);
+        assertFalse(result);
+        assertEquals(1024 * 3, mappedFileQueue.getFlushedWhere());
+        System.out.println("3 " + result + " " + mappedFileQueue.getFlushedWhere());
+
+        result = mappedFileQueue.flush(0);
+        assertFalse(result);
+        assertEquals(1024 * 4, mappedFileQueue.getFlushedWhere());
+        System.out.println("4 " + result + " " + mappedFileQueue.getFlushedWhere());
+
+        result = mappedFileQueue.flush(0);
+        assertFalse(result);
+        assertEquals(1024 * 5, mappedFileQueue.getFlushedWhere());
+        System.out.println("5 " + result + " " + mappedFileQueue.getFlushedWhere());
+
+        result = mappedFileQueue.flush(0);
+        assertFalse(result);
+        assertEquals(1024 * 6, mappedFileQueue.getFlushedWhere());
+        System.out.println("6 " + result + " " + mappedFileQueue.getFlushedWhere());
+
+        mappedFileQueue.shutdown(1000);
+        mappedFileQueue.destroy();
+        System.out.println("MappedFileQueue.flush() OK");
+    }
+
+    @Test
+    public void test_getMapedMemorySize() {
+        final String fixedMsg = "abcd";
+        System.out.println("================================================================");
+        MappedFileQueue mappedFileQueue =
+                new MappedFileQueue("target/unit_test_store/d/", 1024, null);
+
+        for (int i = 0; i < 1024; i++) {
+            MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0);
+            assertTrue(mappedFile != null);
+            boolean result = mappedFile.appendMessage(fixedMsg.getBytes());
+            assertTrue(result);
+        }
+
+        assertEquals(fixedMsg.length() * 1024, mappedFileQueue.getMappedMemorySize());
+        mappedFileQueue.shutdown(1000);
+        mappedFileQueue.destroy();
+        System.out.println("MappedFileQueue.getMappedMemorySize() OK");
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/MappedFileTest.java
----------------------------------------------------------------------
diff --git a/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/MappedFileTest.java b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/MappedFileTest.java
new file mode 100644
index 0000000..94fd5ee
--- /dev/null
+++ b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/MappedFileTest.java
@@ -0,0 +1,93 @@
+/**
+ * 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.
+ */
+
+/**
+ * $Id: MappedFileTest.java 1831 2013-05-16 01:39:51Z shijia.wxr $
+ */
+package com.alibaba.rocketmq.store;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertTrue;
+
+
+public class MappedFileTest {
+
+    private static final String StoreMessage = "Once, there was a chance for me!";
+
+    @BeforeClass
+    public static void setUpBeforeClass() throws Exception {
+
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() throws Exception {
+    }
+
+    @Test
+    public void test_write_read() {
+        try {
+            MappedFile mappedFile = new MappedFile("target/unit_test_store/MappedFileTest/000", 1024 * 64);
+            boolean result = mappedFile.appendMessage(StoreMessage.getBytes());
+            assertTrue(result);
+            System.out.println("write OK");
+
+            SelectMappedBufferResult selectMappedBufferResult = mappedFile.selectMappedBuffer(0);
+            byte[] data = new byte[StoreMessage.length()];
+            selectMappedBufferResult.getByteBuffer().get(data);
+            String readString = new String(data);
+
+            System.out.println("Read: " + readString);
+            assertTrue(readString.equals(StoreMessage));
+
+            mappedFile.shutdown(1000);
+            assertTrue(!mappedFile.isAvailable());
+            selectMappedBufferResult.release();
+            assertTrue(mappedFile.isCleanupOver());
+            assertTrue(mappedFile.destroy(1000));
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Ignore
+    public void test_jvm_crashed() {
+        try {
+            MappedFile mappedFile = new MappedFile("target/unit_test_store/MappedFileTest/10086", 1024 * 64);
+            boolean result = mappedFile.appendMessage(StoreMessage.getBytes());
+            assertTrue(result);
+            System.out.println("write OK");
+
+            SelectMappedBufferResult selectMappedBufferResult = mappedFile.selectMappedBuffer(0);
+            selectMappedBufferResult.release();
+            mappedFile.shutdown(1000);
+
+            byte[] data = new byte[StoreMessage.length()];
+            selectMappedBufferResult.getByteBuffer().get(data);
+            String readString = new String(data);
+            System.out.println(readString);
+        } catch (IOException e) {
+            // TODO Auto-generated catch block
+            e.printStackTrace();
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/RecoverTest.java
----------------------------------------------------------------------
diff --git a/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/RecoverTest.java b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/RecoverTest.java
new file mode 100644
index 0000000..ea83375
--- /dev/null
+++ b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/RecoverTest.java
@@ -0,0 +1,212 @@
+/**
+ * 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.
+ */
+
+/**
+ * $Id: RecoverTest.java 1831 2013-05-16 01:39:51Z shijia.wxr $
+ */
+package com.alibaba.rocketmq.store;
+
+import com.alibaba.rocketmq.common.message.MessageDecoder;
+import com.alibaba.rocketmq.common.message.MessageExt;
+import com.alibaba.rocketmq.store.config.MessageStoreConfig;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertTrue;
+
+
+public class RecoverTest {
+    private static final String StoreMessage = "Once, there was a chance for me!aaaaaaaaaaaaaaaaaaaaaaaa";
+
+    private static int QUEUE_TOTAL = 10;
+
+    private static AtomicInteger QueueId = new AtomicInteger(0);
+
+    private static SocketAddress BornHost;
+
+    private static SocketAddress StoreHost;
+
+    private static byte[] MessageBody;
+    private MessageStore storeWrite1;
+    private MessageStore storeWrite2;
+    private MessageStore storeRead;
+
+    @BeforeClass
+    public static void setUpBeforeClass() throws Exception {
+        StoreHost = new InetSocketAddress(InetAddress.getLocalHost(), 8123);
+        BornHost = new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0);
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() throws Exception {
+    }
+
+    @Test
+    public void test_recover_normally() throws Exception {
+        this.writeMessage(true, true);
+        Thread.sleep(1000 * 3);
+        this.readMessage(1000);
+        this.destroy();
+    }
+
+    public void writeMessage(boolean normal, boolean first) throws Exception {
+        System.out.println("================================================================");
+        long totalMsgs = 100;
+        QUEUE_TOTAL = 3;
+
+        MessageBody = StoreMessage.getBytes();
+        MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+        messageStoreConfig.setMapedFileSizeCommitLog(1024 * 32);
+        messageStoreConfig.setMapedFileSizeConsumeQueue(100 * 20);
+        messageStoreConfig.setMessageIndexEnable(false);
+
+        MessageStore messageStore = new DefaultMessageStore(messageStoreConfig, null, null, null);
+        if (first) {
+            this.storeWrite1 = messageStore;
+        } else {
+            this.storeWrite2 = messageStore;
+        }
+
+        boolean loadResult = messageStore.load();
+        assertTrue(loadResult);
+        messageStore.start();
+        for (long i = 0; i < totalMsgs; i++) {
+            PutMessageResult result = messageStore.putMessage(buildMessage());
+            System.out.println(i + "\t" + result.getAppendMessageResult().getMsgId());
+        }
+
+        if (normal) {
+            messageStore.shutdown();
+        }
+        System.out.println("========================writeMessage OK========================================");
+    }
+
+    public void readMessage(final long msgCnt) throws Exception {
+        System.out.println("================================================================");
+        QUEUE_TOTAL = 3;
+        MessageBody = StoreMessage.getBytes();
+        MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+        messageStoreConfig.setMapedFileSizeCommitLog(1024 * 32);
+        messageStoreConfig.setMapedFileSizeConsumeQueue(100 * 20);
+        messageStoreConfig.setMessageIndexEnable(false);
+        storeRead = new DefaultMessageStore(messageStoreConfig, null, null, null);
+        boolean loadResult = storeRead.load();
+        assertTrue(loadResult);
+        storeRead.start();
+
+        long readCnt = 0;
+        for (int queueId = 0; queueId < QUEUE_TOTAL; queueId++) {
+            for (long offset = 0; ; ) {
+                GetMessageResult result = storeRead.getMessage("GROUP_A", "TOPIC_A", queueId, offset, 1024 * 1024, null);
+                if (result.getStatus() == GetMessageStatus.FOUND) {
+                    System.out.println(queueId + "\t" + result.getMessageCount());
+                    this.veryReadMessage(queueId, offset, result.getMessageBufferList());
+                    offset += result.getMessageCount();
+                    readCnt += result.getMessageCount();
+                    result.release();
+                } else {
+                    break;
+                }
+            }
+        }
+
+        System.out.println("readCnt = " + readCnt);
+        assertTrue(readCnt == msgCnt);
+        System.out.println("========================readMessage OK========================================");
+    }
+
+    private void destroy() {
+        if (storeWrite1 != null) {
+            storeWrite1.shutdown();
+            storeWrite1.destroy();
+        }
+
+        if (storeWrite2 != null) {
+            storeWrite2.shutdown();
+            storeWrite2.destroy();
+        }
+
+        if (storeRead != null) {
+            storeRead.shutdown();
+            storeRead.destroy();
+        }
+    }
+
+    public MessageExtBrokerInner buildMessage() {
+        MessageExtBrokerInner msg = new MessageExtBrokerInner();
+        msg.setTopic("TOPIC_A");
+        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(4);
+        msg.setBornTimestamp(System.currentTimeMillis());
+        msg.setStoreHost(StoreHost);
+        msg.setBornHost(BornHost);
+
+        return msg;
+    }
+
+    private void veryReadMessage(int queueId, long queueOffset, List<ByteBuffer> byteBuffers) {
+        for (ByteBuffer byteBuffer : byteBuffers) {
+            MessageExt msg = MessageDecoder.decode(byteBuffer);
+            System.out.println("request queueId " + queueId + ", request queueOffset " + queueOffset + " msg queue offset "
+                    + msg.getQueueOffset());
+
+            assertTrue(msg.getQueueOffset() == queueOffset);
+
+            queueOffset++;
+        }
+    }
+
+    @Test
+    public void test_recover_normally_write() throws Exception {
+        this.writeMessage(true, true);
+        Thread.sleep(1000 * 3);
+        this.writeMessage(true, false);
+        Thread.sleep(1000 * 3);
+        this.readMessage(2000);
+        this.destroy();
+    }
+
+    @Test
+    public void test_recover_abnormally() throws Exception {
+        this.writeMessage(false, true);
+        Thread.sleep(1000 * 3);
+        this.readMessage(1000);
+        this.destroy();
+    }
+
+    @Test
+    public void test_recover_abnormally_write() throws Exception {
+        this.writeMessage(false, true);
+        Thread.sleep(1000 * 3);
+        this.writeMessage(false, false);
+        Thread.sleep(1000 * 3);
+        this.readMessage(2000);
+        this.destroy();
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/StoreCheckpointTest.java
----------------------------------------------------------------------
diff --git a/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/StoreCheckpointTest.java b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/StoreCheckpointTest.java
new file mode 100644
index 0000000..e0a550d
--- /dev/null
+++ b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/StoreCheckpointTest.java
@@ -0,0 +1,61 @@
+/**
+ * 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.
+ */
+
+/**
+ * $Id: StoreCheckpointTest.java 1831 2013-05-16 01:39:51Z shijia.wxr $
+ */
+package com.alibaba.rocketmq.store;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+
+public class StoreCheckpointTest {
+    @BeforeClass
+    public static void setUpBeforeClass() throws Exception {
+
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() throws Exception {
+    }
+
+    @Test
+    public void test_write_read() {
+        try {
+            StoreCheckpoint storeCheckpoint = new StoreCheckpoint("target/checkpoint_test/0000");
+            long physicMsgTimestamp = 0xAABB;
+            long logicsMsgTimestamp = 0xCCDD;
+            storeCheckpoint.setPhysicMsgTimestamp(physicMsgTimestamp);
+            storeCheckpoint.setLogicsMsgTimestamp(logicsMsgTimestamp);
+            storeCheckpoint.flush();
+
+            long diff = physicMsgTimestamp - storeCheckpoint.getMinTimestamp();
+            assertTrue(diff == 3000);
+            storeCheckpoint.shutdown();
+            storeCheckpoint = new StoreCheckpoint("target/checkpoint_test/0000");
+            assertTrue(physicMsgTimestamp == storeCheckpoint.getPhysicMsgTimestamp());
+            assertTrue(logicsMsgTimestamp == storeCheckpoint.getLogicsMsgTimestamp());
+        } catch (Throwable e) {
+            e.printStackTrace();
+            assertTrue(false);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/index/IndexFileTest.java
----------------------------------------------------------------------
diff --git a/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/index/IndexFileTest.java b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/index/IndexFileTest.java
new file mode 100644
index 0000000..288b87e
--- /dev/null
+++ b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/index/IndexFileTest.java
@@ -0,0 +1,80 @@
+/**
+ * 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.
+ */
+
+/**
+ * $Id: IndexFileTest.java 1831 2013-05-16 01:39:51Z shijia.wxr $
+ */
+package com.alibaba.rocketmq.store.index;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+
+public class IndexFileTest {
+    private static final int hashSlotNum = 100;
+    private static final int indexNum = 400;
+
+    @Test
+    public void test_put_index() {
+        try {
+            IndexFile indexFile = new IndexFile("100", hashSlotNum, indexNum, 0, 0);
+            for (long i = 0; i < (indexNum - 1); i++) {
+                boolean putResult = indexFile.putKey(Long.toString(i), i, System.currentTimeMillis());
+                assertTrue(putResult);
+            }
+
+            boolean putResult = indexFile.putKey(Long.toString(400), 400, System.currentTimeMillis());
+            assertFalse(putResult);
+
+            indexFile.destroy(0);
+        } catch (Exception e) {
+            e.printStackTrace();
+            assertTrue(false);
+        }
+    }
+
+
+    @Test
+    public void test_put_get_index() {
+        try {
+            IndexFile indexFile = new IndexFile("200", hashSlotNum, indexNum, 0, 0);
+
+            for (long i = 0; i < (indexNum - 1); i++) {
+                boolean putResult = indexFile.putKey(Long.toString(i), i, System.currentTimeMillis());
+                assertTrue(putResult);
+            }
+            boolean putResult = indexFile.putKey(Long.toString(400), 400, System.currentTimeMillis());
+            assertFalse(putResult);
+
+            final List<Long> phyOffsets = new ArrayList<Long>();
+            indexFile.selectPhyOffset(phyOffsets, "60", 10, 0, Long.MAX_VALUE, true);
+            for (Long offset : phyOffsets) {
+                System.out.println(offset);
+            }
+            assertFalse(phyOffsets.isEmpty());
+            indexFile.destroy(0);
+        } catch (Exception e) {
+            e.printStackTrace();
+            assertTrue(false);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/schedule/ScheduleMessageTest.java
----------------------------------------------------------------------
diff --git a/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/schedule/ScheduleMessageTest.java b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/schedule/ScheduleMessageTest.java
new file mode 100644
index 0000000..d7de738
--- /dev/null
+++ b/rocketmq-store/src/test/java/com/alibaba/rocketmq/store/schedule/ScheduleMessageTest.java
@@ -0,0 +1,135 @@
+/**
+ * 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.
+ */
+
+/**
+ * $Id: ScheduleMessageTest.java 1831 2013-05-16 01:39:51Z shijia.wxr $
+ */
+package com.alibaba.rocketmq.store.schedule;
+
+import com.alibaba.rocketmq.store.*;
+import com.alibaba.rocketmq.store.config.MessageStoreConfig;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertTrue;
+
+@Ignore
+public class ScheduleMessageTest {
+    private static final String StoreMessage = "Once, there was a chance for me!";
+
+    private static int QUEUE_TOTAL = 100;
+
+    private static AtomicInteger QueueId = new AtomicInteger(0);
+
+    private static SocketAddress BornHost;
+
+    private static SocketAddress StoreHost;
+
+    private static byte[] MessageBody;
+
+    @BeforeClass
+    public static void setUpBeforeClass() throws Exception {
+        StoreHost = new InetSocketAddress(InetAddress.getLocalHost(), 8123);
+        BornHost = new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 0);
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() throws Exception {
+    }
+
+    @Test
+    public void test_delay_message() throws Exception {
+        System.out.println("================================================================");
+        long totalMsgs = 10000;
+        QUEUE_TOTAL = 32;
+
+
+        MessageBody = StoreMessage.getBytes();
+
+        MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
+        messageStoreConfig.setMapedFileSizeCommitLog(1024 * 32);
+        messageStoreConfig.setMapedFileSizeConsumeQueue(1024 * 16);
+        messageStoreConfig.setMaxHashSlotNum(100);
+        messageStoreConfig.setMaxIndexNum(1000 * 10);
+
+        MessageStore master = new DefaultMessageStore(messageStoreConfig, null, null, null);
+
+        boolean load = master.load();
+        assertTrue(load);
+
+
+        master.start();
+        for (int i = 0; i < totalMsgs; i++) {
+            MessageExtBrokerInner msg = buildMessage();
+            msg.setDelayTimeLevel(i % 4);
+
+            PutMessageResult result = master.putMessage(msg);
+            System.out.println(i + "\t" + result.getAppendMessageResult().getMsgId());
+        }
+
+        System.out.println("write message over, wait time up");
+        Thread.sleep(1000 * 20);
+
+
+        for (long i = 0; i < totalMsgs; i++) {
+            try {
+                GetMessageResult result = master.getMessage("GROUP_A", "TOPIC_A", 0, i, 1024 * 1024, null);
+                if (result == null) {
+                    System.out.println("result == null " + i);
+                }
+                assertTrue(result != null);
+                result.release();
+                System.out.println("read " + i + " OK");
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+
+        }
+
+        Thread.sleep(1000 * 15);
+
+
+        master.shutdown();
+
+
+        master.destroy();
+        System.out.println("================================================================");
+    }
+
+    public MessageExtBrokerInner buildMessage() {
+        MessageExtBrokerInner msg = new MessageExtBrokerInner();
+        msg.setTopic("AAA");
+        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(4);
+        msg.setBornTimestamp(System.currentTimeMillis());
+        msg.setStoreHost(StoreHost);
+        msg.setBornHost(BornHost);
+
+        return msg;
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-tools/pom.xml
----------------------------------------------------------------------
diff --git a/rocketmq-tools/pom.xml b/rocketmq-tools/pom.xml
new file mode 100644
index 0000000..5070a68
--- /dev/null
+++ b/rocketmq-tools/pom.xml
@@ -0,0 +1,66 @@
+<!--
+  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 producerGroup 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>com.alibaba.rocketmq</groupId>
+        <artifactId>rocketmq-all</artifactId>
+        <version>4.0.0-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <packaging>jar</packaging>
+    <artifactId>rocketmq-tools</artifactId>
+    <name>rocketmq-tools ${project.version}</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-client</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-store</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-srvutil</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+    </dependencies>
+</project>

http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/057d0e9b/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/DefaultMQAdminExt.java
----------------------------------------------------------------------
diff --git a/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/DefaultMQAdminExt.java b/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/DefaultMQAdminExt.java
new file mode 100644
index 0000000..4576886
--- /dev/null
+++ b/rocketmq-tools/src/main/java/com/alibaba/rocketmq/tools/admin/DefaultMQAdminExt.java
@@ -0,0 +1,458 @@
+/**
+ * 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 com.alibaba.rocketmq.tools.admin;
+
+import com.alibaba.rocketmq.client.ClientConfig;
+import com.alibaba.rocketmq.client.QueryResult;
+import com.alibaba.rocketmq.client.exception.MQBrokerException;
+import com.alibaba.rocketmq.client.exception.MQClientException;
+import com.alibaba.rocketmq.common.MixAll;
+import com.alibaba.rocketmq.common.TopicConfig;
+import com.alibaba.rocketmq.common.admin.ConsumeStats;
+import com.alibaba.rocketmq.common.admin.RollbackStats;
+import com.alibaba.rocketmq.common.admin.TopicStatsTable;
+import com.alibaba.rocketmq.common.message.MessageExt;
+import com.alibaba.rocketmq.common.message.MessageQueue;
+import com.alibaba.rocketmq.common.protocol.body.*;
+import com.alibaba.rocketmq.common.protocol.route.TopicRouteData;
+import com.alibaba.rocketmq.common.subscription.SubscriptionGroupConfig;
+import com.alibaba.rocketmq.remoting.RPCHook;
+import com.alibaba.rocketmq.remoting.exception.*;
+import com.alibaba.rocketmq.tools.admin.api.MessageTrack;
+
+import java.io.UnsupportedEncodingException;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+
+/**
+ * @author shijia.wxr
+ */
+public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
+    private final DefaultMQAdminExtImpl defaultMQAdminExtImpl;
+    private String adminExtGroup = "admin_ext_group";
+    private String createTopicKey = MixAll.DEFAULT_TOPIC;
+    private long timeoutMillis = 5000;
+
+    public DefaultMQAdminExt() {
+        this.defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(this, null, timeoutMillis);
+    }
+
+    public DefaultMQAdminExt(long timeoutMillis) {
+        this.defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(this, null, timeoutMillis);
+    }
+
+    public DefaultMQAdminExt(RPCHook rpcHook) {
+        this.defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(this, rpcHook, timeoutMillis);
+    }
+
+    public DefaultMQAdminExt(RPCHook rpcHook, long timeoutMillis) {
+        this.defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(this, rpcHook, timeoutMillis);
+    }
+
+    public DefaultMQAdminExt(final String adminExtGroup) {
+        this.adminExtGroup = adminExtGroup;
+        this.defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(this, timeoutMillis);
+    }
+
+    public DefaultMQAdminExt(final String adminExtGroup, long timeoutMillis) {
+        this.adminExtGroup = adminExtGroup;
+        this.defaultMQAdminExtImpl = new DefaultMQAdminExtImpl(this, timeoutMillis);
+    }
+
+    @Override
+    public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
+        createTopic(key, newTopic, queueNum, 0);
+    }
+
+    @Override
+    public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag) throws MQClientException {
+        defaultMQAdminExtImpl.createTopic(key, newTopic, queueNum, topicSysFlag);
+    }
+
+    @Override
+    public long searchOffset(MessageQueue mq, long timestamp) throws MQClientException {
+        return defaultMQAdminExtImpl.searchOffset(mq, timestamp);
+    }
+
+    @Override
+    public long maxOffset(MessageQueue mq) throws MQClientException {
+        return defaultMQAdminExtImpl.maxOffset(mq);
+    }
+
+
+    @Override
+    public long minOffset(MessageQueue mq) throws MQClientException {
+        return defaultMQAdminExtImpl.minOffset(mq);
+    }
+
+    @Override
+    public long earliestMsgStoreTime(MessageQueue mq) throws MQClientException {
+        return defaultMQAdminExtImpl.earliestMsgStoreTime(mq);
+    }
+
+    @Override
+    public MessageExt viewMessage(String offsetMsgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        return defaultMQAdminExtImpl.viewMessage(offsetMsgId);
+    }
+
+    @Override
+    public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end) throws MQClientException,
+            InterruptedException {
+        return defaultMQAdminExtImpl.queryMessage(topic, key, maxNum, begin, end);
+    }
+
+    @Override
+    public void start() throws MQClientException {
+        defaultMQAdminExtImpl.start();
+    }
+
+    @Override
+    public void shutdown() {
+        defaultMQAdminExtImpl.shutdown();
+    }
+
+    @Override
+    public void updateBrokerConfig(String brokerAddr, Properties properties) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, UnsupportedEncodingException, InterruptedException, MQBrokerException {
+        defaultMQAdminExtImpl.updateBrokerConfig(brokerAddr, properties);
+    }
+
+    @Override
+    public Properties getBrokerConfig(final String brokerAddr) throws RemotingConnectException,
+            RemotingSendRequestException, RemotingTimeoutException, UnsupportedEncodingException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExtImpl.getBrokerConfig(brokerAddr);
+    }
+
+    @Override
+    public void createAndUpdateTopicConfig(String addr, TopicConfig config) throws RemotingException, MQBrokerException,
+            InterruptedException, MQClientException {
+        defaultMQAdminExtImpl.createAndUpdateTopicConfig(addr, config);
+    }
+
+    @Override
+    public void createAndUpdateSubscriptionGroupConfig(String addr, SubscriptionGroupConfig config) throws RemotingException,
+            MQBrokerException, InterruptedException, MQClientException {
+        defaultMQAdminExtImpl.createAndUpdateSubscriptionGroupConfig(addr, config);
+    }
+
+    @Override
+    public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) {
+        return defaultMQAdminExtImpl.examineSubscriptionGroupConfig(addr, group);
+    }
+
+    @Override
+    public TopicConfig examineTopicConfig(String addr, String topic) {
+        return defaultMQAdminExtImpl.examineTopicConfig(addr, topic);
+    }
+
+    @Override
+    public TopicStatsTable examineTopicStats(String topic) throws RemotingException, MQClientException, InterruptedException,
+            MQBrokerException {
+        return defaultMQAdminExtImpl.examineTopicStats(topic);
+    }
+
+    @Override
+    public TopicList fetchAllTopicList() throws RemotingException, MQClientException, InterruptedException {
+        return this.defaultMQAdminExtImpl.fetchAllTopicList();
+    }
+
+    @Override
+    public TopicList fetchTopicsByCLuster(String clusterName) throws RemotingException, MQClientException, InterruptedException {
+        return this.defaultMQAdminExtImpl.fetchTopicsByCLuster(clusterName);
+    }
+
+    @Override
+    public KVTable fetchBrokerRuntimeStats(final String brokerAddr) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, InterruptedException, MQBrokerException {
+        return this.defaultMQAdminExtImpl.fetchBrokerRuntimeStats(brokerAddr);
+    }
+
+    @Override
+    public ConsumeStats examineConsumeStats(String consumerGroup) throws RemotingException, MQClientException, InterruptedException,
+            MQBrokerException {
+        return examineConsumeStats(consumerGroup, null);
+    }
+
+    @Override
+    public ConsumeStats examineConsumeStats(String consumerGroup, String topic) throws RemotingException, MQClientException,
+            InterruptedException, MQBrokerException {
+        return defaultMQAdminExtImpl.examineConsumeStats(consumerGroup, topic);
+    }
+
+    @Override
+    public ClusterInfo examineBrokerClusterInfo() throws InterruptedException, RemotingConnectException, RemotingTimeoutException,
+            RemotingSendRequestException, MQBrokerException {
+        return defaultMQAdminExtImpl.examineBrokerClusterInfo();
+    }
+
+    @Override
+    public TopicRouteData examineTopicRouteInfo(String topic) throws RemotingException, MQClientException, InterruptedException {
+        return defaultMQAdminExtImpl.examineTopicRouteInfo(topic);
+    }
+
+    @Override
+    public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup) throws InterruptedException, MQBrokerException,
+            RemotingException, MQClientException {
+        return defaultMQAdminExtImpl.examineConsumerConnectionInfo(consumerGroup);
+    }
+
+    @Override
+    public ProducerConnection examineProducerConnectionInfo(String producerGroup, final String topic) throws RemotingException,
+            MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExtImpl.examineProducerConnectionInfo(producerGroup, topic);
+    }
+
+    @Override
+    public List<String> getNameServerAddressList() {
+        return this.defaultMQAdminExtImpl.getNameServerAddressList();
+    }
+
+    @Override
+    public int wipeWritePermOfBroker(final String namesrvAddr, String brokerName) throws RemotingCommandException,
+            RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQClientException {
+        return defaultMQAdminExtImpl.wipeWritePermOfBroker(namesrvAddr, brokerName);
+    }
+
+    @Override
+    public void putKVConfig(String namespace, String key, String value) {
+        defaultMQAdminExtImpl.putKVConfig(namespace, key, value);
+    }
+
+    @Override
+    public String getKVConfig(String namespace, String key) throws RemotingException, MQClientException, InterruptedException {
+        return defaultMQAdminExtImpl.getKVConfig(namespace, key);
+    }
+
+    @Override
+    public KVTable getKVListByNamespace(String namespace) throws RemotingException, MQClientException, InterruptedException {
+        return defaultMQAdminExtImpl.getKVListByNamespace(namespace);
+    }
+
+    @Override
+    public void deleteTopicInBroker(Set<String> addrs, String topic) throws RemotingException, MQBrokerException, InterruptedException,
+            MQClientException {
+        defaultMQAdminExtImpl.deleteTopicInBroker(addrs, topic);
+    }
+
+    @Override
+    public void deleteTopicInNameServer(Set<String> addrs, String topic) throws RemotingException, MQBrokerException, InterruptedException,
+            MQClientException {
+        defaultMQAdminExtImpl.deleteTopicInNameServer(addrs, topic);
+    }
+
+    @Override
+    public void deleteSubscriptionGroup(String addr, String groupName) throws RemotingException, MQBrokerException, InterruptedException,
+            MQClientException {
+        defaultMQAdminExtImpl.deleteSubscriptionGroup(addr, groupName);
+    }
+
+    @Override
+    public void createAndUpdateKvConfig(String namespace, String key, String value) throws RemotingException, MQBrokerException,
+            InterruptedException, MQClientException {
+        defaultMQAdminExtImpl.createAndUpdateKvConfig(namespace, key, value);
+    }
+
+    @Override
+    public void deleteKvConfig(String namespace, String key) throws RemotingException, MQBrokerException, InterruptedException,
+            MQClientException {
+        defaultMQAdminExtImpl.deleteKvConfig(namespace, key);
+    }
+
+    public List<RollbackStats> resetOffsetByTimestampOld(String consumerGroup, String topic, long timestamp, boolean force)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        return defaultMQAdminExtImpl.resetOffsetByTimestampOld(consumerGroup, topic, timestamp, force);
+    }
+
+    @Override
+    public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        return resetOffsetByTimestamp(topic, group, timestamp, isForce, false);
+    }
+
+    public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce, boolean isC)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        return defaultMQAdminExtImpl.resetOffsetByTimestamp(topic, group, timestamp, isForce, isC);
+    }
+
+    @Override
+    public void resetOffsetNew(String consumerGroup, String topic, long timestamp) throws RemotingException, MQBrokerException,
+            InterruptedException, MQClientException {
+        this.defaultMQAdminExtImpl.resetOffsetNew(consumerGroup, topic, timestamp);
+    }
+
+    @Override
+    public Map<String, Map<MessageQueue, Long>> getConsumeStatus(String topic, String group, String clientAddr) throws RemotingException,
+            MQBrokerException, InterruptedException, MQClientException {
+        return defaultMQAdminExtImpl.getConsumeStatus(topic, group, clientAddr);
+    }
+
+    @Override
+    public void createOrUpdateOrderConf(String key, String value, boolean isCluster) throws RemotingException, MQBrokerException,
+            InterruptedException, MQClientException {
+        defaultMQAdminExtImpl.createOrUpdateOrderConf(key, value, isCluster);
+    }
+
+    @Override
+    public GroupList queryTopicConsumeByWho(String topic) throws InterruptedException, MQBrokerException, RemotingException,
+            MQClientException {
+        return this.defaultMQAdminExtImpl.queryTopicConsumeByWho(topic);
+    }
+
+    @Override
+    public List<QueueTimeSpan> queryConsumeTimeSpan(final String topic, final String group) throws InterruptedException, MQBrokerException,
+            RemotingException, MQClientException {
+        return this.defaultMQAdminExtImpl.queryConsumeTimeSpan(topic, group);
+    }
+
+    @Override
+    public boolean cleanExpiredConsumerQueue(String cluster) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, MQClientException, InterruptedException {
+        return defaultMQAdminExtImpl.cleanExpiredConsumerQueue(cluster);
+    }
+
+    @Override
+    public boolean cleanExpiredConsumerQueueByAddr(String addr) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, MQClientException, InterruptedException {
+        return defaultMQAdminExtImpl.cleanExpiredConsumerQueueByAddr(addr);
+    }
+
+    @Override
+    public boolean cleanUnusedTopic(String cluster) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, MQClientException, InterruptedException {
+        return defaultMQAdminExtImpl.cleanUnusedTopicByAddr(cluster);
+    }
+
+    @Override
+    public boolean cleanUnusedTopicByAddr(String addr) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, MQClientException, InterruptedException {
+        return defaultMQAdminExtImpl.cleanUnusedTopicByAddr(addr);
+    }
+
+    @Override
+    public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack) throws RemotingException,
+            MQClientException, InterruptedException {
+        return defaultMQAdminExtImpl.getConsumerRunningInfo(consumerGroup, clientId, jstack);
+    }
+
+    @Override
+    public ConsumeMessageDirectlyResult consumeMessageDirectly(String consumerGroup, String clientId, String msgId)
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExtImpl.consumeMessageDirectly(consumerGroup, clientId, msgId);
+    }
+
+    @Override
+    public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId, final String topic, final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        return defaultMQAdminExtImpl.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
+    }
+
+    @Override
+    public List<MessageTrack> messageTrackDetail(MessageExt msg) throws RemotingException, MQClientException, InterruptedException,
+            MQBrokerException {
+        return this.defaultMQAdminExtImpl.messageTrackDetail(msg);
+    }
+
+    @Override
+    public void cloneGroupOffset(String srcGroup, String destGroup, String topic, boolean isOffline) throws RemotingException,
+            MQClientException, InterruptedException, MQBrokerException {
+        this.defaultMQAdminExtImpl.cloneGroupOffset(srcGroup, destGroup, topic, isOffline);
+    }
+
+    @Override
+    public BrokerStatsData viewBrokerStatsData(String brokerAddr, String statsName, String statsKey) throws RemotingConnectException,
+            RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
+        return this.defaultMQAdminExtImpl.viewBrokerStatsData(brokerAddr, statsName, statsKey);
+    }
+
+    @Override
+    public Set<String> getClusterList(String topic) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, MQClientException, InterruptedException {
+        return this.defaultMQAdminExtImpl.getClusterList(topic);
+    }
+
+    @Override
+    public ConsumeStatsList fetchConsumeStatsInBroker(final String brokerAddr, boolean isOrder, long timeoutMillis) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, MQClientException, InterruptedException {
+        return this.defaultMQAdminExtImpl.fetchConsumeStatsInBroker(brokerAddr, isOrder, timeoutMillis);
+    }
+
+    @Override
+    public Set<String> getTopicClusterList(final String topic) throws InterruptedException, MQBrokerException, MQClientException, RemotingException {
+        return this.defaultMQAdminExtImpl.getTopicClusterList(topic);
+    }
+
+    @Override
+    public SubscriptionGroupWrapper getAllSubscriptionGroup(final String brokerAddr, long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
+            RemotingConnectException, MQBrokerException {
+        return this.defaultMQAdminExtImpl.getAllSubscriptionGroup(brokerAddr, timeoutMillis);
+    }
+
+    @Override
+    public TopicConfigSerializeWrapper getAllTopicGroup(final String brokerAddr, long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
+            RemotingConnectException, MQBrokerException {
+        return this.defaultMQAdminExtImpl.getAllTopicGroup(brokerAddr, timeoutMillis);
+    }
+
+
+    /* (non-Javadoc)
+     * @see com.alibaba.rocketmq.client.MQAdmin#queryMessageByUniqKey(java.lang.String, java.lang.String)
+     */
+    @Override
+    public MessageExt viewMessage(String topic, String msgId)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        return this.defaultMQAdminExtImpl.viewMessage(topic, msgId);
+    }
+
+    public String getAdminExtGroup() {
+        return adminExtGroup;
+    }
+
+    public void setAdminExtGroup(String adminExtGroup) {
+        this.adminExtGroup = adminExtGroup;
+    }
+
+    public String getCreateTopicKey() {
+        return createTopicKey;
+    }
+
+    public void setCreateTopicKey(String createTopicKey) {
+        this.createTopicKey = createTopicKey;
+    }
+
+    @Override
+    public void updateConsumeOffset(String brokerAddr, String consumeGroup, MessageQueue mq, long offset) throws RemotingException, InterruptedException, MQBrokerException {
+        this.defaultMQAdminExtImpl.updateConsumeOffset(brokerAddr, consumeGroup, mq, offset);
+    }
+
+    @Override
+    public void updateNameServerConfig(final Properties properties, final List<String> nameServers)
+            throws InterruptedException, RemotingConnectException,
+            UnsupportedEncodingException, MQBrokerException, RemotingTimeoutException,
+            MQClientException, RemotingSendRequestException {
+        this.defaultMQAdminExtImpl.updateNameServerConfig(properties, nameServers);
+    }
+
+    @Override
+    public Map<String, Properties> getNameServerConfig(final List<String> nameServers)
+            throws InterruptedException, RemotingTimeoutException,
+            RemotingSendRequestException, RemotingConnectException, MQClientException,
+            UnsupportedEncodingException {
+        return this.defaultMQAdminExtImpl.getNameServerConfig(nameServers);
+    }
+}


Mime
View raw message