diff --git a/.travis.yml b/.travis.yml index 45b7d5d79a6..a9927e27687 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,9 +4,9 @@ jdk: - oraclejdk7 - openjdk7 -#script: -# - travis_retry mvn --batch-mode clean apache-rat:check -# - travis_wait 60 mvn --batch-mode clean package findbugs:check +script: + - travis_retry mvn --batch-mode clean apache-rat:check + - travis_wait 60 mvn --batch-mode clean package findbugs:check after_success: - - mvn clean apache-rat:check cobertura:cobertura coveralls:report + - mvn clean cobertura:cobertura coveralls:report diff --git a/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java index 4a19c0cf0e1..e059f992d4a 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/BrokerControllerTest.java @@ -1,4 +1,4 @@ -/* +/** * 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. @@ -8,28 +8,39 @@ * * 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. + * 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.broker; +package com.alibaba.rocketmq.broker; -import org.apache.rocketmq.common.BrokerConfig; -import org.apache.rocketmq.remoting.netty.NettyClientConfig; -import org.apache.rocketmq.remoting.netty.NettyServerConfig; -import org.apache.rocketmq.store.config.MessageStoreConfig; +import com.alibaba.rocketmq.common.BrokerConfig; +import com.alibaba.rocketmq.remoting.netty.NettyClientConfig; +import com.alibaba.rocketmq.remoting.netty.NettyServerConfig; +import com.alibaba.rocketmq.store.config.MessageStoreConfig; +<<<<<<< HEAD +import org.junit.Test; +======= import org.junit.Assert; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +>>>>>>> b85645996a573b19159ad184007484a99742cc5f +/** + * @author shtykh_roman + */ public class BrokerControllerTest { - private static final int RESTART_NUM = 3; +<<<<<<< HEAD +======= protected Logger logger = LoggerFactory.getLogger(BrokerControllerTest.class); +>>>>>>> b85645996a573b19159ad184007484a99742cc5f + private static final int RESTART_NUM = 3; + /** * Tests if the controller can be properly stopped and started. * @@ -45,6 +56,12 @@ public void testRestart() throws Exception { new NettyClientConfig(), // new MessageStoreConfig()); boolean initResult = brokerController.initialize(); +<<<<<<< HEAD + System.out.println("initialize " + initResult); + brokerController.start(); + + brokerController.shutdown(); +======= Assert.assertTrue(initResult); logger.info("Broker is initialized " + initResult); brokerController.start(); @@ -52,6 +69,7 @@ public void testRestart() throws Exception { brokerController.shutdown(); logger.info("Broker is stopped"); +>>>>>>> b85645996a573b19159ad184007484a99742cc5f } } } diff --git a/build.sh b/build.sh new file mode 100755 index 00000000000..1b2cf076cde --- /dev/null +++ b/build.sh @@ -0,0 +1 @@ +mvn install -DskipTests=true -Dmaven.javadoc.skip=true -B -V -X diff --git a/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java b/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java index 69ebfefd87a..f5be396048f 100644 --- a/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java +++ b/client/src/main/java/org/apache/rocketmq/client/ClientConfig.java @@ -20,6 +20,9 @@ import org.apache.rocketmq.common.UtilAll; import org.apache.rocketmq.remoting.common.RemotingUtil; +import java.util.ArrayList; +import java.util.List; + /** * Client Common configuration */ @@ -27,6 +30,8 @@ public class ClientConfig { public static final String SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY = "com.rocketmq.sendMessageWithVIPChannel"; private String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV)); private String clientIP = RemotingUtil.getLocalAddress(); + private List clientIPArray = RemotingUtil.getLocalAddressArray(); + private boolean isClientIPReplaced = false; private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT"); private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors(); /** @@ -63,7 +68,42 @@ public String getClientIP() { return clientIP; } + /** + * to match the right ip, + * side effort:if the candidate is only one,the clientIP will be replace by the first candidate + * @param ipPattern:ip parttern,if ipPattern is null or empty,will return all the candidate, e.g:10.1.2 + * + * + * @return ip candidate list + */ + public List computeClientIP(String ipPattern) { + List rstIPArray = new ArrayList(this.clientIPArray.size()); + if (ipPattern != null && !ipPattern.isEmpty()) + { + for (String ip : this.clientIPArray) + { + if (ip.contains(ipPattern)) + rstIPArray.add(ip); + } + if (!isClientIPReplaced && rstIPArray.size() <= 0) + throw new RuntimeException("you do not have ip address pls use ifconfig to check you ip"); + + if (!isClientIPReplaced && rstIPArray.size() == 1) + this.setClientIP(rstIPArray.get(0)); + } else { + rstIPArray.addAll(this.clientIPArray); + } + if (rstIPArray.size() > 1 && !isClientIPReplaced) + try { + throw new RuntimeException("you have more than one ip address, pls use to avoid it ,or use "); + } + catch (Exception e) { + e.printStackTrace(); + } + return rstIPArray; + } public void setClientIP(String clientIP) { + isClientIPReplaced = true; this.clientIP = clientIP; } diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java index 3480c920e89..1a82f374642 100644 --- a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java +++ b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java @@ -161,6 +161,7 @@ public DefaultMQProducer(RPCHook rpcHook) { */ @Override public void start() throws MQClientException { + this.computeClientIP(null); this.defaultMQProducerImpl.start(); } diff --git a/client/src/test/java/org/apache/rocketmq/client/SimpleConsumerProducerTest.java b/client/src/test/java/org/apache/rocketmq/client/SimpleConsumerProducerTest.java new file mode 100644 index 00000000000..d7e3ff21b18 --- /dev/null +++ b/client/src/test/java/org/apache/rocketmq/client/SimpleConsumerProducerTest.java @@ -0,0 +1,88 @@ +/** + * 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.client; + +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.client.producer.DefaultMQProducer; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.common.consumer.ConsumeFromWhere; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.common.message.MessageExt; +import org.junit.Test; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + + +public class SimpleConsumerProducerTest { + private static final String TOPIC_TEST = "TopicTest-fundmng"; + + @Test + public void producerConsumerTest() throws MQClientException, InterruptedException, IOException { + System.setProperty("rocketmq.namesrv.domain", "jmenv.tbsite.alipay.net"); + + DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("S_fundmng_demo_producer"); + DefaultMQProducer producer = new DefaultMQProducer("P_fundmng_demo_producer"); + //producer.computeClientIP("10.1.33"); + consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); + consumer.subscribe(TOPIC_TEST, null); + + final AtomicLong lastReceivedMills = new AtomicLong(System.currentTimeMillis()); + + final AtomicLong consumeTimes = new AtomicLong(0); + + consumer.registerMessageListener(new MessageListenerConcurrently() { + public ConsumeConcurrentlyStatus consumeMessage(final List msgs, + final ConsumeConcurrentlyContext context) { + System.out.println("Received" + consumeTimes.incrementAndGet() + "messages !"); + + lastReceivedMills.set(System.currentTimeMillis()); + + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + } + }); + + consumer.start(); + producer.start(); + + for (int i = 0; i < 100; i++) { + try { + Message msg = new Message(TOPIC_TEST, ("Hello RocketMQ " + i).getBytes()); + SendResult sendResult = producer.send(msg); + System.out.println(sendResult); + } catch (Exception e) { + System.err.println(e.toString()); + TimeUnit.SECONDS.sleep(1); + } + } + +// // wait no messages + while ((System.currentTimeMillis() - lastReceivedMills.get()) < 5000) { + TimeUnit.MILLISECONDS.sleep(200); + } + + consumer.shutdown(); + producer.shutdown(); + } +} diff --git a/merge.sh b/merge.sh new file mode 100755 index 00000000000..4dc256c5a46 --- /dev/null +++ b/merge.sh @@ -0,0 +1,3 @@ +git fetch upstream +git checkout master +git merge upstream/master diff --git a/pom.xml b/pom.xml index f189bb422c5..afd3d847b0c 100644 --- a/pom.xml +++ b/pom.xml @@ -32,7 +32,7 @@ rocketmq-all 4.0.0-SNAPSHOT pom - Apache RocketMQ git s${project.version} + rocketmq-all ${project.version} http://rocketmq.incubator.apache.org/ @@ -42,8 +42,7 @@ https://git-wip-us.apache.org/repos/asf/incubator-rocketmq.git scm:git:https://git-wip-us.apache.org/repos/asf/incubator-rocketmq.git - scm:git:https://git-wip-us.apache.org/repos/asf/incubator-rocketmq.git - + scm:git:https://git-wip-us.apache.org/repos/asf/incubator-rocketmq.git @@ -158,11 +157,6 @@ 1.7 1.7 - - - https://builds.apache.org/analysis - - file:**/generated-sources/** @@ -343,11 +337,6 @@ findbugs-maven-plugin 3.0.4 - - org.sonarsource.scanner.maven - sonar-maven-plugin - 3.0.2 - diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java index f64f9e153ef..868cd7d373b 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingUtil.java @@ -32,7 +32,10 @@ import java.nio.channels.SocketChannel; import java.nio.channels.spi.SelectorProvider; import java.util.ArrayList; +import java.util.Collections; import java.util.Enumeration; +import java.util.List; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -91,8 +94,9 @@ public static Selector openSelector() throws IOException { public static boolean isLinuxPlatform() { return isLinuxPlatform; } - - public static String getLocalAddress() { + public static List getLocalAddressArray() + { + ArrayList rstIP = new ArrayList(); try { // Traversal Network interface to get the first non-loopback and non-private address Enumeration enumeration = NetworkInterface.getNetworkInterfaces(); @@ -119,17 +123,17 @@ public static String getLocalAddress() { if (ip.startsWith("127.0") || ip.startsWith("192.168")) { continue; } + rstIP.add(ip); - return ip; } - - return ipv4Result.get(ipv4Result.size() - 1); + //to adapt old rocketmq,make the last ent address be the first + Collections.reverse(rstIP); } else if (!ipv6Result.isEmpty()) { - return ipv6Result.get(0); + rstIP.addAll(ipv6Result); + } else { + rstIP.add(normalizeHostAddress(InetAddress.getLocalHost())); } - //If failed to find,fall back to localhost - final InetAddress localHost = InetAddress.getLocalHost(); - return normalizeHostAddress(localHost); + return rstIP; } catch (SocketException e) { e.printStackTrace(); } catch (UnknownHostException e) { @@ -138,6 +142,14 @@ public static String getLocalAddress() { return null; } + public static String getLocalAddress() { + final List address = getLocalAddressArray(); + if (address != null && !address.isEmpty()) { + return address.get(0); + } + + return null; + } public static String normalizeHostAddress(final InetAddress localHost) { if (localHost instanceof Inet6Address) { diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/LocalAddressTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/LocalAddressTest.java new file mode 100644 index 00000000000..84438100801 --- /dev/null +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/LocalAddressTest.java @@ -0,0 +1,19 @@ +package org.apache.rocketmq.remoting; + + +import org.apache.rocketmq.remoting.common.RemotingUtil; +import org.junit.Test; + +import java.util.List; + +/** + * Created by tain on 2017/1/11. + */ +public class LocalAddressTest { + @Test + public void getLoacalAddress() + { + List address = RemotingUtil.getLocalAddressArray(); + System.out.println(address); + } +} diff --git a/rocketmq-store/src/test/resources/logback-test.xml b/rocketmq-store/src/test/resources/logback-test.xml new file mode 100644 index 00000000000..02844756b5b --- /dev/null +++ b/rocketmq-store/src/test/resources/logback-test.xml @@ -0,0 +1,42 @@ + + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{5} - %msg%n +<<<<<<< HEAD +======= + UTF-8 +>>>>>>> b85645996a573b19159ad184007484a99742cc5f + + + + + + + +<<<<<<< HEAD + +======= + +>>>>>>> b85645996a573b19159ad184007484a99742cc5f + + + + diff --git a/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java b/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java index f98b26a194b..0d15ece1d24 100644 --- a/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java +++ b/store/src/main/java/org/apache/rocketmq/store/MappedFileQueue.java @@ -1,4 +1,4 @@ -/* +/** * 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. @@ -6,7 +6,7 @@ * (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 + * 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, @@ -14,21 +14,22 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.store; +package com.alibaba.rocketmq.store; + +import com.alibaba.rocketmq.common.UtilAll; +import com.alibaba.rocketmq.common.constant.LoggerName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Iterator; -import java.util.List; -import java.util.ListIterator; +import java.util.*; import java.util.concurrent.CopyOnWriteArrayList; -import org.apache.rocketmq.common.UtilAll; -import org.apache.rocketmq.common.constant.LoggerName; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; + +/** + * @author shijia.wxr + */ public class MappedFileQueue { private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); private static final Logger LOG_ERROR = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME); @@ -48,13 +49,15 @@ public class MappedFileQueue { private volatile long storeTimestamp = 0; + public MappedFileQueue(final String storePath, int mappedFileSize, - AllocateMappedFileService allocateMappedFileService) { + AllocateMappedFileService allocateMappedFileService) { this.storePath = storePath; this.mappedFileSize = mappedFileSize; this.allocateMappedFileService = allocateMappedFileService; } + public void checkSelf() { if (!this.mappedFiles.isEmpty()) { @@ -66,7 +69,7 @@ public void checkSelf() { if (pre != null) { if (cur.getFileFromOffset() - pre.getFileFromOffset() != this.mappedFileSize) { LOG_ERROR.error("[BUG]The mappedFile queue's data is damaged, the adjacent mappedFile's offset don't match. pre file {}, cur file {}", - pre.getFileName(), cur.getFileName()); + pre.getFileName(), cur.getFileName()); } } pre = cur; @@ -74,6 +77,7 @@ public void checkSelf() { } } + public MappedFile getMappedFileByTime(final long timestamp) { Object[] mfs = this.copyMappedFiles(0); @@ -90,6 +94,7 @@ public MappedFile getMappedFileByTime(final long timestamp) { return (MappedFile) mfs[mfs.length - 1]; } + private Object[] copyMappedFiles(final int reservedMappedFiles) { Object[] mfs; @@ -101,6 +106,7 @@ private Object[] copyMappedFiles(final int reservedMappedFiles) { return mfs; } + public void truncateDirtyFiles(long offset) { List willRemoveFiles = new ArrayList(); @@ -121,6 +127,7 @@ public void truncateDirtyFiles(long offset) { this.deleteExpiredFile(willRemoveFiles); } + private void deleteExpiredFile(List files) { if (!files.isEmpty()) { @@ -144,6 +151,7 @@ private void deleteExpiredFile(List files) { } } + public boolean load() { File dir = new File(this.storePath); File[] files = dir.listFiles(); @@ -154,10 +162,11 @@ public boolean load() { if (file.length() != this.mappedFileSize) { log.warn(file + "\t" + file.length() - + " length not matched message store config value, ignore it"); + + " length not matched message store config value, ignore it"); return true; } + try { MappedFile mappedFile = new MappedFile(file.getPath(), mappedFileSize); @@ -176,6 +185,7 @@ public boolean load() { return true; } + public long howMuchFallBehind() { if (this.mappedFiles.isEmpty()) return 0; @@ -191,6 +201,7 @@ public long howMuchFallBehind() { return 0; } + public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) { long createOffset = -1; MappedFile mappedFileLast = getLastMappedFile(); @@ -206,12 +217,12 @@ public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) if (createOffset != -1 && needCreate) { String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset); String nextNextFilePath = this.storePath + File.separator - + UtilAll.offset2FileName(createOffset + this.mappedFileSize); + + UtilAll.offset2FileName(createOffset + this.mappedFileSize); MappedFile mappedFile = null; if (this.allocateMappedFileService != null) { mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath, - nextNextFilePath, this.mappedFileSize); + nextNextFilePath, this.mappedFileSize); } else { try { mappedFile = new MappedFile(nextFilePath, this.mappedFileSize); @@ -260,12 +271,11 @@ public boolean resetOffset(long offset) { if (mappedFileLast != null) { long lastOffset = mappedFileLast.getFileFromOffset() + - mappedFileLast.getWrotePosition(); + mappedFileLast.getWrotePosition(); long diff = lastOffset - offset; final int maxDiff = this.mappedFileSize * 2; - if (diff > maxDiff) - return false; + if (diff > maxDiff) return false; } ListIterator iterator = this.mappedFiles.listIterator(); @@ -299,6 +309,7 @@ public long getMinOffset() { return -1; } + public long getMaxOffset() { MappedFile mappedFile = getLastMappedFile(); if (mappedFile != null) { @@ -334,9 +345,9 @@ public void deleteLastMappedFile() { } public int deleteExpiredFileByTime(final long expiredTime, - final int deleteFilesInterval, - final long intervalForcibly, - final boolean cleanImmediately) { + final int deleteFilesInterval, + final long intervalForcibly, + final boolean cleanImmediately) { Object[] mfs = this.copyMappedFiles(0); if (null == mfs) @@ -376,6 +387,7 @@ public int deleteExpiredFileByTime(final long expiredTime, return deleteCount; } + public int deleteExpiredFileByOffset(long offset, int unitSize) { Object[] mfs = this.copyMappedFiles(0); @@ -395,7 +407,7 @@ public int deleteExpiredFileByOffset(long offset, int unitSize) { destroy = maxOffsetInLogicQueue < offset; if (destroy) { log.info("physic min offset " + offset + ", logics in current mappedFile max offset " - + maxOffsetInLogicQueue + ", delete it"); + + maxOffsetInLogicQueue + ", delete it"); } } else { log.warn("this being not executed forever."); @@ -416,6 +428,7 @@ public int deleteExpiredFileByOffset(long offset, int unitSize) { return deleteCount; } + public boolean flush(final int flushLeastPages) { boolean result = true; MappedFile mappedFile = this.findMappedFileByOffset(this.flushedWhere, false); @@ -460,7 +473,7 @@ public MappedFile findMappedFileByOffset(final long offset, final boolean return int index = (int) ((offset / this.mappedFileSize) - (mappedFile.getFileFromOffset() / this.mappedFileSize)); if (index < 0 || index >= this.mappedFiles.size()) { LOG_ERROR.warn("Offset for {} not matched. Request offset: {}, index: {}, " + - "mappedFileSize: {}, mappedFiles count: {}", + "mappedFileSize: {}, mappedFiles count: {}", mappedFile, offset, index, @@ -484,6 +497,7 @@ public MappedFile findMappedFileByOffset(final long offset, final boolean return return null; } + public MappedFile getFirstMappedFile() { MappedFile mappedFileFirst = null; @@ -504,6 +518,7 @@ public MappedFile findMappedFileByOffset(final long offset) { return findMappedFileByOffset(offset, false); } + public long getMappedMemorySize() { long size = 0; @@ -519,6 +534,7 @@ public long getMappedMemorySize() { return size; } + public boolean retryDeleteFirstFile(final long intervalForcibly) { MappedFile mappedFile = this.getFirstMappedFile(); if (mappedFile != null) { @@ -541,12 +557,14 @@ public boolean retryDeleteFirstFile(final long intervalForcibly) { return false; } + public void shutdown(final long intervalForcibly) { for (MappedFile mf : this.mappedFiles) { mf.shutdown(intervalForcibly); } } + public void destroy() { for (MappedFile mf : this.mappedFiles) { mf.destroy(1000 * 3); @@ -561,22 +579,27 @@ public void destroy() { } } + public long getFlushedWhere() { return flushedWhere; } + public void setFlushedWhere(long flushedWhere) { this.flushedWhere = flushedWhere; } + public long getStoreTimestamp() { return storeTimestamp; } + public List getMappedFiles() { return mappedFiles; } + public int getMappedFileSize() { return mappedFileSize; } diff --git a/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java index 762bdb6adca..5f188a611dc 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java +++ b/store/src/main/java/org/apache/rocketmq/store/ha/HAService.java @@ -1,4 +1,4 @@ -/* +/** * 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. @@ -6,7 +6,7 @@ * (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 + * 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, @@ -14,17 +14,21 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.store.ha; +package com.alibaba.rocketmq.store.ha; + +import com.alibaba.rocketmq.common.ServiceThread; +import com.alibaba.rocketmq.common.constant.LoggerName; +import com.alibaba.rocketmq.remoting.common.RemotingUtil; +import com.alibaba.rocketmq.store.CommitLog.GroupCommitRequest; +import com.alibaba.rocketmq.store.DefaultMessageStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.nio.ByteBuffer; -import java.nio.channels.ClosedChannelException; -import java.nio.channels.SelectionKey; -import java.nio.channels.Selector; -import java.nio.channels.ServerSocketChannel; -import java.nio.channels.SocketChannel; +import java.nio.channels.*; import java.util.ArrayList; import java.util.LinkedList; import java.util.List; @@ -32,14 +36,11 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; -import org.apache.rocketmq.common.ServiceThread; -import org.apache.rocketmq.common.constant.LoggerName; -import org.apache.rocketmq.remoting.common.RemotingUtil; -import org.apache.rocketmq.store.CommitLog; -import org.apache.rocketmq.store.DefaultMessageStore; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; + +/** + * @author shijia.wxr + */ public class HAService { private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); @@ -58,33 +59,38 @@ public class HAService { private final HAClient haClient; + public HAService(final DefaultMessageStore defaultMessageStore) throws IOException { this.defaultMessageStore = defaultMessageStore; this.acceptSocketService = - new AcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort()); + new AcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort()); this.groupTransferService = new GroupTransferService(); this.haClient = new HAClient(); } + public void updateMasterAddress(final String newAddr) { if (this.haClient != null) { this.haClient.updateMasterAddress(newAddr); } } - public void putRequest(final CommitLog.GroupCommitRequest request) { + + public void putRequest(final GroupCommitRequest request) { this.groupTransferService.putRequest(request); } + public boolean isSlaveOK(final long masterPutWhere) { boolean result = this.connectionCount.get() > 0; result = - result - && ((masterPutWhere - this.push2SlaveMaxOffset.get()) < this.defaultMessageStore - .getMessageStoreConfig().getHaSlaveFallbehindMax()); + result + && ((masterPutWhere - this.push2SlaveMaxOffset.get()) < this.defaultMessageStore + .getMessageStoreConfig().getHaSlaveFallbehindMax()); return result; } + /** */ @@ -100,10 +106,12 @@ public void notifyTransferSome(final long offset) { } } + public AtomicInteger getConnectionCount() { return connectionCount; } + // public void notifyTransferSome() { // this.groupTransferService.notifyTransferSome(); // } @@ -115,18 +123,21 @@ public void start() throws Exception { this.haClient.start(); } + public void addConnection(final HAConnection conn) { synchronized (this.connectionList) { this.connectionList.add(conn); } } + public void removeConnection(final HAConnection conn) { synchronized (this.connectionList) { this.connectionList.remove(conn); } } + public void shutdown() { this.haClient.shutdown(); this.acceptSocketService.shutdown(true); @@ -134,6 +145,7 @@ public void shutdown() { this.groupTransferService.shutdown(); } + public void destroyConnections() { synchronized (this.connectionList) { for (HAConnection c : this.connectionList) { @@ -144,10 +156,12 @@ public void destroyConnections() { } } + public DefaultMessageStore getDefaultMessageStore() { return defaultMessageStore; } + public WaitNotifyObject getWaitNotifyObject() { return waitNotifyObject; } @@ -160,9 +174,9 @@ public AtomicLong getPush2SlaveMaxOffset() { * Listens to slave connections to create {@link HAConnection}. */ class AcceptSocketService extends ServiceThread { - private final SocketAddress socketAddressListen; private ServerSocketChannel serverSocketChannel; private Selector selector; + private final SocketAddress socketAddressListen; public AcceptSocketService(final int port) { this.socketAddressListen = new InetSocketAddress(port); @@ -170,7 +184,6 @@ public AcceptSocketService(final int port) { /** * Starts listening to slave connections. - * * @throws Exception If fails. */ public void beginAccept() throws Exception { @@ -182,19 +195,39 @@ public void beginAccept() throws Exception { this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT); } +<<<<<<< HEAD + public void beginAccept() throws Exception { + this.serverSocketChannel = ServerSocketChannel.open(); + this.selector = RemotingUtil.openSelector(); + this.serverSocketChannel.socket().setReuseAddress(true); + this.serverSocketChannel.socket().bind(this.socketAddressListen); + this.serverSocketChannel.configureBlocking(false); + this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT); + } + +======= /** {@inheritDoc} */ +>>>>>>> b85645996a573b19159ad184007484a99742cc5f @Override public void shutdown(final boolean interrupt) { super.shutdown(interrupt); try { +<<<<<<< HEAD + serverSocketChannel.close(); +======= this.serverSocketChannel.close(); this.selector.close(); - } catch (IOException e) { +>>>>>>> b85645996a573b19159ad184007484a99742cc5f + } + catch (IOException e) { log.error("AcceptSocketService shutdown exception", e); } } +<<<<<<< HEAD +======= /** {@inheritDoc} */ +>>>>>>> b85645996a573b19159ad184007484a99742cc5f @Override public void run() { log.info(this.getServiceName() + " service started"); @@ -211,7 +244,7 @@ public void run() { if (sc != null) { HAService.log.info("HAService receive new connection, " - + sc.socket().getRemoteSocketAddress()); + + sc.socket().getRemoteSocketAddress()); try { HAConnection conn = new HAConnection(HAService.this, sc); @@ -250,10 +283,11 @@ public String getServiceName() { class GroupTransferService extends ServiceThread { private final WaitNotifyObject notifyTransferObject = new WaitNotifyObject(); - private volatile List requestsWrite = new ArrayList<>(); - private volatile List requestsRead = new ArrayList<>(); + private volatile List requestsWrite = new ArrayList<>(); + private volatile List requestsRead = new ArrayList<>(); + - public void putRequest(final CommitLog.GroupCommitRequest request) { + public void putRequest(final GroupCommitRequest request) { synchronized (this) { this.requestsWrite.add(request); if (hasNotified.compareAndSet(false, true)) { @@ -262,19 +296,22 @@ public void putRequest(final CommitLog.GroupCommitRequest request) { } } + public void notifyTransferSome() { this.notifyTransferObject.wakeup(); } + private void swapRequests() { - List tmp = this.requestsWrite; + List tmp = this.requestsWrite; this.requestsWrite = this.requestsRead; this.requestsRead = tmp; } + private void doWaitTransfer() { if (!this.requestsRead.isEmpty()) { - for (CommitLog.GroupCommitRequest req : this.requestsRead) { + for (GroupCommitRequest req : this.requestsRead) { boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset(); for (int i = 0; !transferOK && i < 5; i++) { this.notifyTransferObject.waitForRunning(1000); @@ -292,12 +329,13 @@ private void doWaitTransfer() { } } + public void run() { log.info(this.getServiceName() + " service started"); while (!this.isStopped()) { try { - this.waitForRunning(10); + this.waitForRunning(0); this.doWaitTransfer(); } catch (Exception e) { log.warn(this.getServiceName() + " service has exception. ", e); @@ -307,11 +345,13 @@ public void run() { log.info(this.getServiceName() + " service end"); } + @Override protected void onWaitEnd() { this.swapRequests(); } + @Override public String getServiceName() { return GroupTransferService.class.getSimpleName(); @@ -331,10 +371,12 @@ class HAClient extends ServiceThread { private ByteBuffer byteBufferRead = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE); private ByteBuffer byteBufferBackup = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE); + public HAClient() throws IOException { this.selector = RemotingUtil.openSelector(); } + public void updateMasterAddress(final String newAddr) { String currentAddr = this.masterAddress.get(); if (currentAddr == null || !currentAddr.equals(newAddr)) { @@ -343,15 +385,17 @@ public void updateMasterAddress(final String newAddr) { } } + private boolean isTimeToReportOffset() { long interval = - HAService.this.defaultMessageStore.getSystemClock().now() - this.lastWriteTimestamp; + HAService.this.defaultMessageStore.getSystemClock().now() - this.lastWriteTimestamp; boolean needHeart = interval > HAService.this.defaultMessageStore.getMessageStoreConfig() - .getHaSendHeartbeatInterval(); + .getHaSendHeartbeatInterval(); return needHeart; } + private boolean reportSlaveMaxOffset(final long maxOffset) { this.reportOffset.position(0); this.reportOffset.limit(8); @@ -364,7 +408,7 @@ private boolean reportSlaveMaxOffset(final long maxOffset) { this.socketChannel.write(this.reportOffset); } catch (IOException e) { log.error(this.getServiceName() - + "reportSlaveMaxOffset this.socketChannel.write exception", e); + + "reportSlaveMaxOffset this.socketChannel.write exception", e); return false; } } @@ -372,6 +416,7 @@ private boolean reportSlaveMaxOffset(final long maxOffset) { return !this.reportOffset.hasRemaining(); } + // private void reallocateByteBuffer() { // ByteBuffer bb = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE); // int remain = this.byteBufferRead.limit() - this.dispatchPostion; @@ -400,12 +445,14 @@ private void reallocateByteBuffer() { this.dispatchPostion = 0; } + private void swapByteBuffer() { ByteBuffer tmp = this.byteBufferRead; this.byteBufferRead = this.byteBufferBackup; this.byteBufferBackup = tmp; } + private boolean processReadEvent() { int readSizeZeroTimes = 0; while (this.byteBufferRead.hasRemaining()) { @@ -437,6 +484,7 @@ private boolean processReadEvent() { return true; } + private boolean dispatchReadRequest() { final int msgHeaderSize = 8 + 4; // phyoffset + size int readSocketPos = this.byteBufferRead.position(); @@ -449,19 +497,22 @@ private boolean dispatchReadRequest() { long slavePhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset(); + if (slavePhyOffset != 0) { if (slavePhyOffset != masterPhyOffset) { log.error("master pushed offset not equal the max phy offset in slave, SLAVE: " - + slavePhyOffset + " MASTER: " + masterPhyOffset); + + slavePhyOffset + " MASTER: " + masterPhyOffset); return false; } } + if (diff >= (msgHeaderSize + bodySize)) { byte[] bodyData = new byte[bodySize]; this.byteBufferRead.position(this.dispatchPostion + msgHeaderSize); this.byteBufferRead.get(bodyData); + HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData); this.byteBufferRead.position(readSocketPos); @@ -485,6 +536,7 @@ private boolean dispatchReadRequest() { return true; } + private boolean reportSlaveMaxOffsetPlus() { boolean result = true; long currentPhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset(); @@ -500,6 +552,7 @@ private boolean reportSlaveMaxOffsetPlus() { return result; } + private boolean connectMaster() throws ClosedChannelException { if (null == socketChannel) { String addr = this.masterAddress.get(); @@ -522,6 +575,7 @@ private boolean connectMaster() throws ClosedChannelException { return this.socketChannel != null; } + private void closeMaster() { if (null != this.socketChannel) { try { @@ -549,6 +603,7 @@ private void closeMaster() { } } + @Override public void run() { log.info(this.getServiceName() + " service started"); @@ -564,8 +619,10 @@ public void run() { } } + this.selector.select(1000); + boolean ok = this.processReadEvent(); if (!ok) { this.closeMaster(); @@ -575,13 +632,14 @@ public void run() { continue; } + long interval = - HAService.this.getDefaultMessageStore().getSystemClock().now() - - this.lastWriteTimestamp; + HAService.this.getDefaultMessageStore().getSystemClock().now() + - this.lastWriteTimestamp; if (interval > HAService.this.getDefaultMessageStore().getMessageStoreConfig() - .getHaHousekeepingInterval()) { + .getHaHousekeepingInterval()) { log.warn("HAClient, housekeeping, found this connection[" + this.masterAddress - + "] expired, " + interval); + + "] expired, " + interval); this.closeMaster(); log.warn("HAClient, master not response some time, so close connection"); } @@ -597,6 +655,7 @@ public void run() { log.info(this.getServiceName() + " service end"); } + // // private void disableWriteFlag() { // if (this.socketChannel != null) { diff --git a/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java b/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java index 1ebf52a6dc4..f4f27bc3b8d 100644 --- a/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java +++ b/store/src/main/java/org/apache/rocketmq/store/index/IndexService.java @@ -1,4 +1,4 @@ -/* +/** * 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. @@ -6,7 +6,7 @@ * (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 + * 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, @@ -14,7 +14,17 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.store.index; +package com.alibaba.rocketmq.store.index; + +import com.alibaba.rocketmq.common.UtilAll; +import com.alibaba.rocketmq.common.constant.LoggerName; +import com.alibaba.rocketmq.common.message.MessageConst; +import com.alibaba.rocketmq.common.sysflag.MessageSysFlag; +import com.alibaba.rocketmq.store.DefaultMessageStore; +import com.alibaba.rocketmq.store.DispatchRequest; +import com.alibaba.rocketmq.store.config.StorePathConfigHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; @@ -23,35 +33,35 @@ import java.util.List; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; -import org.apache.rocketmq.common.UtilAll; -import org.apache.rocketmq.common.constant.LoggerName; -import org.apache.rocketmq.common.message.MessageConst; -import org.apache.rocketmq.common.sysflag.MessageSysFlag; -import org.apache.rocketmq.store.DefaultMessageStore; -import org.apache.rocketmq.store.DispatchRequest; -import org.apache.rocketmq.store.config.StorePathConfigHelper; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; + +/** + * @author shijia.wxr + */ public class IndexService { private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); - /** Maximum times to attempt index file creation. */ - private static final int MAX_TRY_IDX_CREATE = 3; private final DefaultMessageStore defaultMessageStore; + private final int hashSlotNum; private final int indexNum; private final String storePath; + private final ArrayList indexFileList = new ArrayList(); private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); + /** Maximum times to attempt index file creation. */ + private static final int MAX_TRY_IDX_CREATE = 3; + + public IndexService(final DefaultMessageStore store) { this.defaultMessageStore = store; this.hashSlotNum = store.getMessageStoreConfig().getMaxHashSlotNum(); this.indexNum = store.getMessageStoreConfig().getMaxIndexNum(); this.storePath = - StorePathConfigHelper.getStorePathIndex(store.getMessageStoreConfig().getStorePathRootDir()); + StorePathConfigHelper.getStorePathIndex(store.getMessageStoreConfig().getStorePathRootDir()); } + public boolean load(final boolean lastExitOK) { File dir = new File(this.storePath); File[] files = dir.listFiles(); @@ -65,7 +75,7 @@ public boolean load(final boolean lastExitOK) { if (!lastExitOK) { if (f.getEndTimestamp() > this.defaultMessageStore.getStoreCheckpoint() - .getIndexMsgTimestamp()) { + .getIndexMsgTimestamp()) { f.destroy(0); continue; } @@ -74,10 +84,10 @@ public boolean load(final boolean lastExitOK) { log.info("load index file OK, " + f.getFileName()); this.indexFileList.add(f); } catch (IOException e) { - log.error("load file {} error", file, e); + log.error("load file " + file + " error", e); return false; } catch (NumberFormatException e) { - log.error("load file {} error", file, e); + continue; } } } @@ -138,6 +148,7 @@ private void deleteExpiredFile(List files) { } } + public void destroy() { try { this.readWriteLock.readLock().lock(); @@ -152,6 +163,7 @@ public void destroy() { } } + public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long begin, long end) { List phyOffsets = new ArrayList(maxNum); @@ -174,6 +186,7 @@ public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long f.selectPhyOffset(phyOffsets, buildKey(topic, key), maxNum, begin, end, lastFile); } + if (f.getBeginTimestamp() < begin) { break; } @@ -192,10 +205,12 @@ public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long return new QueryOffsetResult(phyOffsets, indexLastUpdateTimestamp, indexLastUpdatePhyoffset); } + private String buildKey(final String topic, final String key) { return topic + "#" + key; } + public void buildIndex(DispatchRequest req) { IndexFile indexFile = retryGetAndCreateIndexFile(); if (indexFile != null) { @@ -243,6 +258,7 @@ public void buildIndex(DispatchRequest req) { } } + private IndexFile putKey(IndexFile indexFile, DispatchRequest msg, String idxKey) { for (boolean ok = indexFile.putKey(idxKey, msg.getCommitLogOffset(), msg.getStoreTimestamp()); !ok; ) { log.warn("Index file [" + indexFile.getFileName() + "] is full, trying to create another one"); @@ -287,6 +303,7 @@ public IndexFile retryGetAndCreateIndexFile() { return indexFile; } + public IndexFile getAndCreateLastIndexFile() { IndexFile indexFile = null; IndexFile prevIndexFile = null; @@ -309,14 +326,15 @@ public IndexFile getAndCreateLastIndexFile() { this.readWriteLock.readLock().unlock(); } + if (indexFile == null) { try { String fileName = - this.storePath + File.separator - + UtilAll.timeMillisToHumanString(System.currentTimeMillis()); + this.storePath + File.separator + + UtilAll.timeMillisToHumanString(System.currentTimeMillis()); indexFile = - new IndexFile(fileName, this.hashSlotNum, this.indexNum, lastUpdateEndPhyOffset, - lastUpdateIndexTimestamp); + new IndexFile(fileName, this.hashSlotNum, this.indexNum, lastUpdateEndPhyOffset, + lastUpdateIndexTimestamp); this.readWriteLock.writeLock().lock(); this.indexFileList.add(indexFile); } catch (Exception e) { @@ -325,6 +343,7 @@ public IndexFile getAndCreateLastIndexFile() { this.readWriteLock.writeLock().unlock(); } + if (indexFile != null) { final IndexFile flushThisFile = prevIndexFile; Thread flushThread = new Thread(new Runnable() { @@ -342,6 +361,7 @@ public void run() { return indexFile; } + public void flush(final IndexFile f) { if (null == f) return; @@ -360,10 +380,12 @@ public void flush(final IndexFile f) { } } + public void start() { } + public void shutdown() { } diff --git a/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java b/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java index 90fbbd35015..e3bd8a73c09 100644 --- a/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/MappedFileQueueTest.java @@ -1,4 +1,4 @@ -/* +/** * 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. @@ -8,29 +8,24 @@ * * 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. + * 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 vintagewang@apache.org $ + * $Id: MappedFileQueueTest.java 1831 2013-05-16 01:39:51Z shijia.wxr $ */ -package org.apache.rocketmq.store; +package com.alibaba.rocketmq.store; -import org.junit.After; -import org.junit.AfterClass; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.Test; +import org.junit.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; +import static org.junit.Assert.*; + public class MappedFileQueueTest { private static final Logger logger = LoggerFactory.getLogger(MappedFileQueueTest.class); @@ -57,10 +52,13 @@ public void tearDown() throws Exception { @Test public void test_getLastMapedFile() { final String fixedMsg = "0123456789abcdef"; +<<<<<<< HEAD +======= +>>>>>>> b85645996a573b19159ad184007484a99742cc5f logger.debug("================================================================"); MappedFileQueue mappedFileQueue = - new MappedFileQueue("target/unit_test_store/a/", 1024, null); + new MappedFileQueue("target/unit_test_store/a/", 1024, null); for (int i = 0; i < 1024; i++) { MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0); @@ -78,20 +76,28 @@ public void test_getLastMapedFile() { logger.debug("MappedFileQueue.getLastMappedFile() OK"); } + @Test public void test_findMapedFileByOffset() { // four-byte string. final String fixedMsg = "abcd"; +<<<<<<< HEAD +======= +>>>>>>> b85645996a573b19159ad184007484a99742cc5f logger.debug("================================================================"); MappedFileQueue mappedFileQueue = - new MappedFileQueue("target/unit_test_store/b/", 1024, null); + 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()); +<<<<<<< HEAD + // logger.debug("appendMessage " + bytes); +======= +>>>>>>> b85645996a573b19159ad184007484a99742cc5f assertTrue(result); } @@ -100,28 +106,32 @@ public void test_findMapedFileByOffset() { MappedFile mappedFile = mappedFileQueue.findMappedFileByOffset(0); assertTrue(mappedFile != null); assertEquals(mappedFile.getFileFromOffset(), 0); - + mappedFile = mappedFileQueue.findMappedFileByOffset(100); assertTrue(mappedFile != null); assertEquals(mappedFile.getFileFromOffset(), 0); - + mappedFile = mappedFileQueue.findMappedFileByOffset(1024); assertTrue(mappedFile != null); assertEquals(mappedFile.getFileFromOffset(), 1024); - + mappedFile = mappedFileQueue.findMappedFileByOffset(1024 + 100); assertTrue(mappedFile != null); assertEquals(mappedFile.getFileFromOffset(), 1024); - + mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 2); assertTrue(mappedFile != null); assertEquals(mappedFile.getFileFromOffset(), 1024 * 2); - + mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 2 + 100); assertTrue(mappedFile != null); assertEquals(mappedFile.getFileFromOffset(), 1024 * 2); +<<<<<<< HEAD + +======= // over mapped memory size. +>>>>>>> b85645996a573b19159ad184007484a99742cc5f mappedFile = mappedFileQueue.findMappedFileByOffset(1024 * 4); assertTrue(mappedFile == null); @@ -136,10 +146,13 @@ public void test_findMapedFileByOffset() { @Test public void test_commit() { final String fixedMsg = "0123456789abcdef"; +<<<<<<< HEAD +======= +>>>>>>> b85645996a573b19159ad184007484a99742cc5f logger.debug("================================================================"); MappedFileQueue mappedFileQueue = - new MappedFileQueue("target/unit_test_store/c/", 1024, null); + new MappedFileQueue("target/unit_test_store/c/", 1024, null); for (int i = 0; i < 1024; i++) { MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0); @@ -152,27 +165,27 @@ public void test_commit() { boolean result = mappedFileQueue.flush(0); assertFalse(result); assertEquals(1024 * 1, mappedFileQueue.getFlushedWhere()); - + result = mappedFileQueue.flush(0); assertFalse(result); assertEquals(1024 * 2, mappedFileQueue.getFlushedWhere()); - + result = mappedFileQueue.flush(0); assertFalse(result); assertEquals(1024 * 3, mappedFileQueue.getFlushedWhere()); - + result = mappedFileQueue.flush(0); assertFalse(result); assertEquals(1024 * 4, mappedFileQueue.getFlushedWhere()); - + result = mappedFileQueue.flush(0); assertFalse(result); assertEquals(1024 * 5, mappedFileQueue.getFlushedWhere()); - + result = mappedFileQueue.flush(0); assertFalse(result); assertEquals(1024 * 6, mappedFileQueue.getFlushedWhere()); - + mappedFileQueue.shutdown(1000); mappedFileQueue.destroy(); logger.debug("MappedFileQueue.flush() OK"); @@ -181,10 +194,13 @@ public void test_commit() { @Test public void test_getMapedMemorySize() { final String fixedMsg = "abcd"; +<<<<<<< HEAD +======= +>>>>>>> b85645996a573b19159ad184007484a99742cc5f logger.debug("================================================================"); MappedFileQueue mappedFileQueue = - new MappedFileQueue("target/unit_test_store/d/", 1024, null); + new MappedFileQueue("target/unit_test_store/d/", 1024, null); for (int i = 0; i < 1024; i++) { MappedFile mappedFile = mappedFileQueue.getLastMappedFile(0); diff --git a/store/src/test/java/org/apache/rocketmq/store/index/IndexFileTest.java b/store/src/test/java/org/apache/rocketmq/store/index/IndexFileTest.java index 956f86743f8..f596833ddec 100644 --- a/store/src/test/java/org/apache/rocketmq/store/index/IndexFileTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/index/IndexFileTest.java @@ -1,4 +1,4 @@ -/* +/** * 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. @@ -8,32 +8,42 @@ * * 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. + * 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 vintagewang@apache.org $ + * $Id: IndexFileTest.java 1831 2013-05-16 01:39:51Z shijia.wxr $ */ -package org.apache.rocketmq.store.index; +package com.alibaba.rocketmq.store.index; + +import org.junit.Test; import java.util.ArrayList; import java.util.List; -import org.junit.Test; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; + public class IndexFileTest { private static final int HASH_SLOT_NUM = 100; private static final int INDEX_NUM = 400; @Test public void test_put_index() throws Exception { +<<<<<<< HEAD + 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); + } + +======= IndexFile indexFile = new IndexFile("100", HASH_SLOT_NUM, INDEX_NUM, 0, 0); for (long i = 0; i < (INDEX_NUM - 1); i++) { boolean putResult = indexFile.putKey(Long.toString(i), i, System.currentTimeMillis()); @@ -41,30 +51,44 @@ public void test_put_index() throws Exception { } // put over index file capacity. +>>>>>>> b85645996a573b19159ad184007484a99742cc5f boolean putResult = indexFile.putKey(Long.toString(400), 400, System.currentTimeMillis()); assertFalse(putResult); - + indexFile.destroy(0); } + @Test public void test_put_get_index() throws Exception { +<<<<<<< HEAD + 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); + } +======= IndexFile indexFile = new IndexFile("200", HASH_SLOT_NUM, INDEX_NUM, 0, 0); - + for (long i = 0; i < (INDEX_NUM - 1); i++) { boolean putResult = indexFile.putKey(Long.toString(i), i, System.currentTimeMillis()); assertTrue(putResult); } // put over index file capacity. +>>>>>>> b85645996a573b19159ad184007484a99742cc5f boolean putResult = indexFile.putKey(Long.toString(400), 400, System.currentTimeMillis()); assertFalse(putResult); - + final List phyOffsets = new ArrayList(); indexFile.selectPhyOffset(phyOffsets, "60", 10, 0, Long.MAX_VALUE, true); assertFalse(phyOffsets.isEmpty()); assertEquals(1, phyOffsets.size()); +<<<<<<< HEAD +======= +>>>>>>> b85645996a573b19159ad184007484a99742cc5f indexFile.destroy(0); } }