[ISSUE #8961] Automatic recognition of address scheme in Topic Route …

…by host (#8962)

* automatic recognition of address scheme in topic route by host.

common/src/main/java/org/apache/rocketmq/common/utils/IPAddressUtils.java

@@ -35,6 +35,14 @@ public static boolean isValidIp(String ip) {
         return VALIDATOR.isValid(ip);
     }
 
+    public static boolean isValidIPv4(String ip) {
+        return VALIDATOR.isValidInet4Address(ip);
+    }
+
+    public static boolean isValidIPv6(String ip) {
+        return VALIDATOR.isValidInet6Address(ip);
+    }
+
     public static boolean isValidCidr(String cidr) {
         return isValidIPv4Cidr(cidr) || isValidIPv6Cidr(cidr);
     }

proxy/src/main/java/org/apache/rocketmq/proxy/common/Address.java

@@ -18,6 +18,7 @@
 
 import com.google.common.net.HostAndPort;
 import java.util.Objects;
+import org.apache.rocketmq.common.utils.IPAddressUtils;
 
 public class Address {
 
@@ -31,6 +32,11 @@ public enum AddressScheme {
     private AddressScheme addressScheme;
     private HostAndPort hostAndPort;
 
+    public Address(HostAndPort hostAndPort) {
+        this.addressScheme = buildScheme(hostAndPort);
+        this.hostAndPort = hostAndPort;
+    }
+
     public Address(AddressScheme addressScheme, HostAndPort hostAndPort) {
         this.addressScheme = addressScheme;
         this.hostAndPort = hostAndPort;
@@ -52,6 +58,20 @@ public void setHostAndPort(HostAndPort hostAndPort) {
         this.hostAndPort = hostAndPort;
     }
 
+    private AddressScheme buildScheme(HostAndPort hostAndPort) {
+        if (hostAndPort == null) {
+            return AddressScheme.UNRECOGNIZED;
+        }
+        String address = hostAndPort.getHost();
+        if (IPAddressUtils.isValidIPv4(address)) {
+            return AddressScheme.IPv4;
+        }
+        if (IPAddressUtils.isValidIPv6(address)) {
+            return AddressScheme.IPv6;
+        }
+        return AddressScheme.DOMAIN_NAME;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/GetTopicRouteActivity.java

@@ -50,7 +50,7 @@ protected RemotingCommand processRequest0(ChannelHandlerContext ctx, RemotingCom
             (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
         List<Address> addressList = new ArrayList<>();
         // AddressScheme is just a placeholder and will not affect topic route result in this case.
-        addressList.add(new Address(Address.AddressScheme.IPv4, HostAndPort.fromParts(proxyConfig.getRemotingAccessAddr(), proxyConfig.getRemotingListenPort())));
+        addressList.add(new Address(HostAndPort.fromParts(proxyConfig.getRemotingAccessAddr(), proxyConfig.getRemotingListenPort())));
         ProxyTopicRouteData proxyTopicRouteData = messagingProcessor.getTopicRouteDataForProxy(context, addressList, requestHeader.getTopic());
         TopicRouteData topicRouteData = proxyTopicRouteData.buildTopicRouteData();
 

proxy/src/main/java/org/apache/rocketmq/proxy/service/route/ProxyTopicRouteData.java

@@ -43,7 +43,7 @@ public ProxyTopicRouteData(TopicRouteData topicRouteData) {
             brokerData.getBrokerAddrs().forEach((brokerId, brokerAddr) -> {
                 HostAndPort brokerHostAndPort = HostAndPort.fromString(brokerAddr);
 
-                proxyBrokerData.getBrokerAddrs().put(brokerId, Lists.newArrayList(new Address(Address.AddressScheme.IPv4, brokerHostAndPort)));
+                proxyBrokerData.getBrokerAddrs().put(brokerId, Lists.newArrayList(new Address(brokerHostAndPort)));
             });
             this.brokerDatas.add(proxyBrokerData);
         }
@@ -61,7 +61,7 @@ public ProxyTopicRouteData(TopicRouteData topicRouteData, int port) {
                 HostAndPort brokerHostAndPort = HostAndPort.fromString(brokerAddr);
                 HostAndPort proxyHostAndPort = HostAndPort.fromParts(brokerHostAndPort.getHost(), port);
 
-                proxyBrokerData.getBrokerAddrs().put(brokerId, Lists.newArrayList(new Address(Address.AddressScheme.IPv4, proxyHostAndPort)));
+                proxyBrokerData.getBrokerAddrs().put(brokerId, Lists.newArrayList(new Address(proxyHostAndPort)));
             });
             this.brokerDatas.add(proxyBrokerData);
         }

proxy/src/test/java/org/apache/rocketmq/proxy/common/AddressTest.java

@@ -0,0 +1,60 @@
+/*
+ * 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.proxy.common;
+
+import com.google.common.net.HostAndPort;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+public class AddressTest {
+
+    @Test
+    public void testConstructorWithIPv4() {
+        HostAndPort hostAndPort = HostAndPort.fromString("192.168.1.1:8080");
+        Address address = new Address(hostAndPort);
+
+        assertEquals(Address.AddressScheme.IPv4, address.getAddressScheme());
+        assertEquals(hostAndPort, address.getHostAndPort());
+    }
+
+    @Test
+    public void testConstructorWithIPv6() {
+        HostAndPort hostAndPort = HostAndPort.fromString("[2001:db8::1]:8080");
+        Address address = new Address(hostAndPort);
+
+        assertEquals(Address.AddressScheme.IPv6, address.getAddressScheme());
+        assertEquals(hostAndPort, address.getHostAndPort());
+    }
+
+    @Test
+    public void testConstructorWithDomainName() {
+        HostAndPort hostAndPort = HostAndPort.fromString("example.com:8080");
+        Address address = new Address(hostAndPort);
+
+        assertEquals(Address.AddressScheme.DOMAIN_NAME, address.getAddressScheme());
+        assertEquals(hostAndPort, address.getHostAndPort());
+    }
+
+    @Test
+    public void testConstructorWithNullHostAndPort() {
+        Address address = new Address(null);
+
+        assertEquals(Address.AddressScheme.UNRECOGNIZED, address.getAddressScheme());
+        assertNull(address.getHostAndPort());
+    }
+}