Add multiple endpoint listener for fluss server.

fluss-common/src/main/java/com/alibaba/fluss/cluster/Endpoint.java

@@ -0,0 +1,232 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.cluster;
+
+import com.alibaba.fluss.annotation.Internal;
+import com.alibaba.fluss.config.ConfigOptions;
+import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.utils.StringUtils;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+/**
+ * Endpoint is what fluss server is listened for. It includes host, port and listener name. Listener
+ * name is used for routing, all the fluss servers can have the same listener names to listen. For
+ * example, coordinator server and tablet sever can use internal listener to communicate with each
+ * other. And If a client connect to a server with a host and port, it can only see the other
+ * server's same listener.
+ */
+@Internal
+public class Endpoint {
+    private static final Pattern ENDPOINT_PARSE_EXP =
+            Pattern.compile("^(.*)://\\[?([0-9a-zA-Z\\-%._:]*)\\]?:(-?[0-9]+)");
+
+    private final String host;
+    private final int port;
+    private final String listenerName;
+
+    public Endpoint(String host, int port, String listenerName) {
+        this.host = host;
+        this.port = port;
+        this.listenerName = listenerName;
+    }
+
+    public String getHost() {
+        return host;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public String getListenerName() {
+        return listenerName;
+    }
+
+    public static List<Endpoint> loadBindEndpoints(Configuration conf, ServerType serverType) {
+        if (conf.getOptional(ConfigOptions.BIND_LISTENERS).isPresent()) {
+            String listeners = conf.getString(ConfigOptions.BIND_LISTENERS);
+            List<Endpoint> endpoints = fromListenersString(listeners, true);
+            String internalListenerName = conf.get(ConfigOptions.INTERNAL_LISTENER_NAME);
+            if (endpoints.stream()
+                    .noneMatch(
+                            endpoint -> endpoint.getListenerName().equals(internalListenerName))) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "Internal listener name %s is not included in listeners %s",
+                                internalListenerName, listeners));
+            }
+            return endpoints;
+        }
+
+        if (conf.getOptional(ConfigOptions.INTERNAL_LISTENER_NAME).isPresent()) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "%s cannot be set without %s",
+                            ConfigOptions.INTERNAL_LISTENER_NAME.key(),
+                            ConfigOptions.BIND_LISTENERS.key()));
+        }
+
+        String host = getHost(conf, serverType);
+        String port = getPort(conf, serverType);
+
+        if (host != null && port != null) {
+            return Collections.singletonList(
+                    new Endpoint(
+                            host,
+                            Integer.parseInt(port),
+                            ConfigOptions.INTERNAL_LISTENER_NAME.defaultValue()));
+        }
+
+        throw new IllegalArgumentException("No server listeners are configured");
+    }
+
+    public static List<Endpoint> loadAdvertisedEndpoints(
+            List<Endpoint> bindEndpoints, Configuration conf) {
+        List<Endpoint> advertisedEndpoints =
+                fromListenersString(conf.getString(ConfigOptions.ADVERTISED_LISTENERS), false);
+        Set<String> bindListenerNames =
+                bindEndpoints.stream().map(Endpoint::getListenerName).collect(Collectors.toSet());
+        Map<String, Endpoint> advertisedEndpointMap = new HashMap<>();
+        for (Endpoint advertisedEndpoint : advertisedEndpoints) {
+            if (bindListenerNames.contains(advertisedEndpoint.listenerName)) {
+                advertisedEndpointMap.put(advertisedEndpoint.listenerName, advertisedEndpoint);
+            } else {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "advertised listener name %s is not included in listeners %s",
+                                advertisedEndpoint.listenerName,
+                                Endpoint.toListenersString(bindEndpoints)));
+            }
+        }
+
+        return bindEndpoints.stream()
+                .map(
+                        endpoint ->
+                                advertisedEndpointMap.getOrDefault(
+                                        endpoint.getListenerName(), endpoint))
+                .collect(Collectors.toList());
+    }
+
+    public static List<Endpoint> fromListenersString(String listeners) {
+        return fromListenersString(listeners, true);
+    }
+
+    private static List<Endpoint> fromListenersString(
+            String listeners, boolean requireDistinctPorts) {
+        if (StringUtils.isNullOrWhitespaceOnly(listeners)) {
+            return Collections.emptyList();
+        }
+
+        List<Endpoint> endpoints =
+                Arrays.stream(listeners.split(","))
+                        .map(Endpoint::fromConnectionString)
+                        .collect(Collectors.toList());
+
+        Set<String> distinctListenerNames =
+                endpoints.stream().map(Endpoint::getListenerName).collect(Collectors.toSet());
+        if (endpoints.size() != distinctListenerNames.size()) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Each listener must have a different name, listeners: %s", listeners));
+        }
+
+        if (requireDistinctPorts) {
+            List<Integer> portsExcludingZero =
+                    endpoints.stream()
+                            .map(Endpoint::getPort)
+                            .filter(port -> port != 0)
+                            .collect(Collectors.toList());
+            Set<Integer> distinctPorts = new HashSet<>(portsExcludingZero);
+            if (portsExcludingZero.size() != distinctPorts.size()) {
+                throw new IllegalArgumentException(
+                        String.format(
+                                "There is more than one endpoint with the same port, listeners: %s",
+                                listeners));
+            }
+        }
+
+        return endpoints;
+    }
+
+    private static String getHost(Configuration conf, ServerType serverType) {
+        return serverType == ServerType.COORDINATOR
+                ? conf.get(ConfigOptions.COORDINATOR_HOST)
+                : conf.get(ConfigOptions.TABLET_SERVER_HOST);
+    }
+
+    private static String getPort(Configuration conf, ServerType serverType) {
+        return serverType == ServerType.COORDINATOR
+                ? conf.get(ConfigOptions.COORDINATOR_PORT)
+                : conf.get(ConfigOptions.TABLET_SERVER_PORT);
+    }
+
+    /**
+     * Create Endpoint object from {@code connectionString}.
+     *
+     * @param connectionString the format is listener_name://host:port or listener_name://[ipv6
+     *     host]:port for example: INTERNAL://my_host:9092, CLIENT://my_host:9093 or
+     *     REPLICATION://[::1]:9094
+     */
+    private static Endpoint fromConnectionString(String connectionString) {
+        Matcher matcher = ENDPOINT_PARSE_EXP.matcher(connectionString.trim());
+        if (!matcher.matches()) {
+            throw new IllegalArgumentException("Invalid endpoint format: " + connectionString);
+        }
+
+        return new Endpoint(matcher.group(2), Integer.parseInt(matcher.group(3)), matcher.group(1));
+    }
+
+    public static String toListenersString(List<Endpoint> endpoints) {
+        return endpoints.stream().map(Endpoint::connectionString).collect(Collectors.joining(","));
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        Endpoint endpoint = (Endpoint) o;
+        return port == endpoint.port
+                && Objects.equals(host, endpoint.host)
+                && Objects.equals(listenerName, endpoint.listenerName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(host, port, listenerName);
+    }
+
+    public String connectionString() {
+        return listenerName + "://" + host + ":" + port;
+    }
+
+    @Override
+    public String toString() {
+        return connectionString();
+    }
+}

fluss-common/src/main/java/com/alibaba/fluss/cluster/MetadataCache.java

@@ -18,11 +18,10 @@
 
 import com.alibaba.fluss.annotation.Internal;
 
-import javax.annotation.Nullable;
-
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
 
 /**
  * The metadata cache to cache the cluster metadata info.
@@ -37,8 +36,7 @@ public interface MetadataCache {
      *
      * @return the coordinator server node
      */
-    @Nullable
-    ServerNode getCoordinatorServer();
+    Optional<ServerNode> getCoordinatorServer(String listenerName);
 
     /**
      * Check whether the tablet server id related tablet server node is alive.
@@ -54,22 +52,24 @@ public interface MetadataCache {
      * @param serverId the tablet server id
      * @return the tablet server node
      */
-    @Nullable
-    ServerNode getTabletServer(int serverId);
+    Optional<ServerNode> getTabletServer(int serverId, String listenerName);
 
     /**
      * Get all alive tablet server nodes.
      *
      * @return all alive tablet server nodes
      */
-    Map<Integer, ServerNode> getAllAliveTabletServers();
+    Map<Integer, ServerNode> getAllAliveTabletServers(String listenerName);
+
+    Set<Integer> getAliveTabletServerIds();
 
     /** Get ids of all alive tablet server nodes. */
     default int[] getLiveServerIds() {
-        List<ServerNode> serverNodes = new ArrayList<>(getAllAliveTabletServers().values());
-        int[] server = new int[serverNodes.size()];
-        for (int i = 0; i < serverNodes.size(); i++) {
-            server[i] = serverNodes.get(i).id();
+        Set<Integer> aliveTabletServerIds = getAliveTabletServerIds();
+        int[] server = new int[aliveTabletServerIds.size()];
+        Iterator<Integer> iterator = aliveTabletServerIds.iterator();
+        for (int i = 0; i < aliveTabletServerIds.size(); i++) {
+            server[i] = iterator.next();
         }
         return server;
     }