bugfix: fix broker cannot connect caused by invalid ipv6 format

Author: cicada-lewis

consumer/consumer.go

@@ -971,6 +971,7 @@ func (dc *defaultConsumer) findConsumerList(topic string) []string {
 		req := &internal.GetConsumerListRequestHeader{
 			ConsumerGroup: dc.consumerGroup,
 		}
+		brokerAddr = utils.AdaptIPv6(brokerAddr) // adapt IPv6 address to a format that can be used with net.DialContext
 		cmd := remote.NewRemotingCommand(internal.ReqGetConsumerListByGroup, req, nil)
 		res, err := dc.client.InvokeSync(context.Background(), brokerAddr, cmd, 3*time.Second) // TODO 超时机制有问题
 		if err != nil {

internal/utils/net.go

@@ -20,10 +20,12 @@ package utils
 import (
 	"bytes"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/v2/errors"
 	"net"
 	"strconv"
+	"strings"
 	"time"
+
+	"github.com/apache/rocketmq-client-go/v2/errors"
 )
 
 var (
@@ -68,3 +70,42 @@ func FakeIP() []byte {
 func GetAddressByBytes(data []byte) string {
 	return net.IPv4(data[0], data[1], data[2], data[3]).String()
 }
+
+
+// AdaptIPv6 adapts an IPv6 address to a format that can be used with net.DialContext, it will return the original address if the address is not a valid IPv6 address
+// addr must contain a valid IPv6 address and a port number
+func AdaptIPv6(addr string) string {
+	var host, port string
+	var err error
+	host, port, err = net.SplitHostPort(addr)
+	if err != nil {
+		// if the address is not in the format of host:port, return the original address 
+		if addrErr, ok := err.(*net.AddrError); ok && addrErr.Err == "too many colons in address" {
+			if i := strings.LastIndex(addr, ":"); i > 0 {
+				host = addr[:i]
+				port = addr[i+1:]
+			}
+		} else {
+			return addr 
+		}
+	}
+
+	if _, err := strconv.Atoi(port); err != nil {
+		// if the port is not a valid port, return the original address
+		return addr 
+	}
+
+	if ip := net.ParseIP(host); ip == nil || ip.To4() != nil {
+		// if the host is not a valid IP address or is an IPv4 address, return the original address
+		return addr
+	} 
+
+	// if the host is in the format of [host]:port, return the original address 
+	if strings.HasPrefix(host, "[") && strings.HasSuffix(host, "]") {
+		return net.JoinHostPort(host[1:len(host)-1], port)
+	}
+
+
+	return net.JoinHostPort(host, port)
+	
+}

internal/utils/net_test.go

@@ -17,8 +17,128 @@ limitations under the License.
 
 package utils
 
-import "testing"
+import (
+	"testing"
+)
 
 func TestLocalIP2(t *testing.T) {
 	t.Log(LocalIP)
 }
+
+func TestAdaptIPv6(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    string
+		expected string
+	}{
+		{
+			name:     "valid IPv6 address with port",
+			input:    "2001:db8::1:8080",
+			expected: "[2001:db8::1]:8080",
+		},
+		{
+			name:     "valid IPv6 address with port - short format",
+			input:    "::1:8080",
+			expected: "[::1]:8080",
+		},
+		{
+			name:     "valid IPv6 address with port - full format",
+			input:    "2001:0db8:85a3:0000:0000:8a2e:0370:7334:8080",
+			expected: "[2001:0db8:85a3:0000:0000:8a2e:0370:7334]:8080",
+		},
+		{
+			name:     "already bracketed IPv6 address",
+			input:    "[2001:db8::1]:8080",
+			expected: "[2001:db8::1]:8080",
+		},
+		{
+			name:     "already bracketed IPv6 address - short format",
+			input:    "[::1]:8080",
+			expected: "[::1]:8080",
+		},
+		{
+			name:     "IPv4 address should return original",
+			input:    "192.168.1.1:8080",
+			expected: "192.168.1.1:8080",
+		},
+		{
+			name:     "IPv4 address should return original - localhost",
+			input:    "127.0.0.1:8080",
+			expected: "127.0.0.1:8080",
+		},
+		{
+			name:     "invalid address format - no port",
+			input:    "2001:db8::1",
+			expected: "2001:db8::1",
+		},
+		{
+			name:     "invalid address format - invalid port",
+			input:    "2001:db8::1:invalid",
+			expected: "2001:db8::1:invalid",
+		},
+		{
+			name:     "invalid address format - empty string",
+			input:    "",
+			expected: "",
+		},
+		{
+			name:     "invalid address format - port only",
+			input:    ":8080",
+			expected: ":8080",
+		},
+		{
+			name:     "invalid address format - colon only",
+			input:    ":",
+			expected: ":",
+		},
+		{
+			name:     "invalid address format - multiple colons but no valid IPv6",
+			input:    "invalid:address:8080",
+			expected: "invalid:address:8080",
+		},
+		{
+			name:     "domain name should return original",
+			input:    "localhost:8080",
+			expected: "localhost:8080",
+		},
+		{
+			name:     "domain name should return original - with dots",
+			input:    "example.com:8080",
+			expected: "example.com:8080",
+		},
+		{
+			name:     "IPv6 address with large port number",
+			input:    "2001:db8::1:65535",
+			expected: "[2001:db8::1]:65535",
+		},
+		{
+			name:     "IPv6 address with small port number",
+			input:    "2001:db8::1:1",
+			expected: "[2001:db8::1]:1",
+		},
+		{
+			name:     "IPv6 address with zero port number",
+			input:    "2001:db8::1:0",
+			expected: "[2001:db8::1]:0",
+		},
+		{
+			name:     "complex IPv6 address",
+			input:    "fe80::1%lo0:8080",
+			expected: "fe80::1%lo0:8080",
+		},
+		{
+			name:     "IPv6 address with zone identifier",
+			input:    "fe80::1%eth0:8080",
+			expected: "fe80::1%eth0:8080",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := AdaptIPv6(tt.input)
+			if result != tt.expected {
+				t.Errorf("AdaptIPv6(%q) = %q, expected %q", tt.input, result, tt.expected)
+			}
+		})
+	}
+}