when the client host have more than 2 ip address, the ClientConfig get the last one ipv4 always. It will make some unpredictable result.

.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

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,13 +56,20 @@ 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();
             logger.info("Broker is started");
 
             brokerController.shutdown();
             logger.info("Broker is stopped");
+>>>>>>> b85645996a573b19159ad184007484a99742cc5f
         }
     }
 }

build.sh

@@ -0,0 +1 @@
+mvn install -DskipTests=true -Dmaven.javadoc.skip=true -B -V -X

client/src/main/java/org/apache/rocketmq/client/ClientConfig.java

@@ -20,13 +20,18 @@
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.remoting.common.RemotingUtil;
 
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * Client Common configuration
  */
 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<String> 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<String> computeClientIP(String ipPattern) {
+        List<String> rstIPArray = new ArrayList<String>(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 <computeClientIP> to avoid it ,or use <setClientIP>");
+            }
+            catch (Exception e) {
+                e.printStackTrace();
+            }
+        return rstIPArray;
+    }
     public void setClientIP(String clientIP) {
+        isClientIPReplaced = true;
         this.clientIP = clientIP;
     }
 

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();
     }
 

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<MessageExt> 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();
+    }
+}

merge.sh

@@ -0,0 +1,3 @@
+git fetch upstream
+git checkout master
+git merge upstream/master

pom.xml

@@ -32,7 +32,7 @@
     <artifactId>rocketmq-all</artifactId>
     <version>4.0.0-SNAPSHOT</version>
     <packaging>pom</packaging>
-    <name>Apache RocketMQ git s${project.version}</name>
+    <name>rocketmq-all ${project.version}</name>
     <url>http://rocketmq.incubator.apache.org/</url>
 
     <prerequisites>
@@ -42,8 +42,7 @@
     <scm>
         <url>https://git-wip-us.apache.org/repos/asf/incubator-rocketmq.git</url>
         <connection>scm:git:https://git-wip-us.apache.org/repos/asf/incubator-rocketmq.git</connection>
-        <developerConnection>scm:git:https://git-wip-us.apache.org/repos/asf/incubator-rocketmq.git
-        </developerConnection>
+        <developerConnection>scm:git:https://git-wip-us.apache.org/repos/asf/incubator-rocketmq.git</developerConnection>
     </scm>
 
     <mailingLists>
@@ -158,11 +157,6 @@
         <!-- compiler settings properties -->
         <maven.compiler.source>1.7</maven.compiler.source>
         <maven.compiler.target>1.7</maven.compiler.target>
-
-        <!-- URL of the ASF SonarQube server -->
-        <sonar.host.url>https://builds.apache.org/analysis</sonar.host.url>
-        <!-- Exclude all generated code -->
-        <sonar.exclusions>file:**/generated-sources/**</sonar.exclusions>
     </properties>
 
     <modules>
@@ -343,11 +337,6 @@
                 <artifactId>findbugs-maven-plugin</artifactId>
                 <version>3.0.4</version>
             </plugin>
-            <plugin>
-                <groupId>org.sonarsource.scanner.maven</groupId>
-                <artifactId>sonar-maven-plugin</artifactId>
-                <version>3.0.2</version>
-            </plugin>
         </plugins>
     </build>
 

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<String> getLocalAddressArray()
+    {
+        ArrayList<String> rstIP = new ArrayList<String>();
         try {
             // Traversal Network interface to get the first non-loopback and non-private address
             Enumeration<NetworkInterface> 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<String> address = getLocalAddressArray();
+        if (address != null && !address.isEmpty()) {
+            return address.get(0);
+        }
+
+        return null;
+    }
 
     public static String normalizeHostAddress(final InetAddress localHost) {
         if (localHost instanceof Inet6Address) {

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);
+    }
+}