[ISSUE #9497] Fix IndexOutOfBoundsException in getEarliestMessageTime when running in IPv6 environment (#9498)

Co-authored-by: shixiaoxiao <shixiaoxiao@bytedance.com>

Author: woaishixiaoxiao

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

@@ -179,6 +179,18 @@ public static String normalizeHostAddress(final InetAddress localHost) {
         }
     }
 
+    public static String denormalizeHostAddress(final String bracketedAddress) {
+        if (bracketedAddress == null) {
+            return null;
+        }
+
+        if (bracketedAddress.startsWith("[") && bracketedAddress.endsWith("]")) {
+            return bracketedAddress.substring(1, bracketedAddress.length() - 1);
+        } else {
+            return bracketedAddress;
+        }
+    }
+
     public static SocketAddress string2SocketAddress(final String addr) {
         int split = addr.lastIndexOf(":");
         String host = addr.substring(0, split);
@@ -211,4 +223,18 @@ private static boolean isBridge(NetworkInterface networkInterface) {
         }
         return false;
     }
+
+    // valid various ipv6 format like:
+    // with scope 2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth0
+    // with bracketed [2001:0db8:85a3:0000:0000:8a2e:0370:7334]
+    public static boolean validCommonInet6Address(String ipOrCidr) {
+        String  ipWithoutBracketed = denormalizeHostAddress(ipOrCidr);
+        if (ipWithoutBracketed != null && ipWithoutBracketed.length() != 0) {
+            InetAddressValidator validator = InetAddressValidator.getInstance();
+            if (validator.isValidInet6Address(ipWithoutBracketed.split("%")[0])) {
+                return true;
+            }
+        }
+        return false;
+    }
 }

common/src/test/java/org/apache/rocketmq/common/utils/IPAddressUtilsTest.java

@@ -18,6 +18,8 @@
 
 import org.junit.Test;
 
+import static org.apache.rocketmq.common.utils.NetworkUtil.validCommonInet6Address;
+
 public class IPAddressUtilsTest {
 
     @Test
@@ -72,4 +74,19 @@ public void isValidIPOrCidr() {
         assert IPAddressUtils.isValidIPOrCidr(ipv4Cidr);
         assert IPAddressUtils.isValidIPOrCidr(ipv6Cidr);
     }
+
+    @Test
+    public void isValidIPv6Common() {
+        String ipv6WithoutScope = "2001:0db8:85a3:0000:0000:8a2e:0370:7334";
+        assert validCommonInet6Address(ipv6WithoutScope);
+        String ipv6WithScope = "2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth0";
+        assert validCommonInet6Address(ipv6WithScope);
+        String ipv6WithBracketedAndScope = "[2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth0]";
+        assert validCommonInet6Address(ipv6WithBracketedAndScope);
+        String ipv4 = "192.168.1.0";
+        assert !validCommonInet6Address(ipv4);
+        String ipv4Cidr = "192.168.1.0/24";
+        assert !validCommonInet6Address(ipv4Cidr);
+    }
+
 }

store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

@@ -57,7 +57,6 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.validator.routines.InetAddressValidator;
 import org.apache.rocketmq.common.AbstractBrokerRunnable;
 import org.apache.rocketmq.common.BoundaryType;
 import org.apache.rocketmq.common.BrokerConfig;
@@ -80,6 +79,7 @@
 import org.apache.rocketmq.common.sysflag.MessageSysFlag;
 import org.apache.rocketmq.common.topic.TopicValidator;
 import org.apache.rocketmq.common.utils.CleanupPolicyUtils;
+import org.apache.rocketmq.common.utils.NetworkUtil;
 import org.apache.rocketmq.common.utils.QueueTypeUtils;
 import org.apache.rocketmq.common.utils.ServiceProvider;
 import org.apache.rocketmq.common.utils.ThreadUtils;
@@ -1174,8 +1174,7 @@ public long getEarliestMessageTime() {
             minPhyOffset += DLedgerEntry.BODY_OFFSET;
         }
         int size = MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSITION + 8;
-        InetAddressValidator validator = InetAddressValidator.getInstance();
-        if (validator.isValidInet6Address(this.brokerConfig.getBrokerIP1())) {
+        if (NetworkUtil.validCommonInet6Address(this.brokerConfig.getBrokerIP1())) {
             size = MessageDecoder.MESSAGE_STORE_TIMESTAMP_POSITION + 20;
         }
         return this.getCommitLog().pickupStoreTimestamp(minPhyOffset, size);