[server] Fix compatibility issue when only [coordinator|tablet-server].host is set (#907)

Author: wangfengpro

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

@@ -106,15 +106,12 @@ && getHost(conf, serverType).isPresent()) {
         }
 
         Optional<String> maybeHost = getHost(conf, serverType);
-        Optional<String> maybePort = getPort(conf, serverType);
 
         // backward compatibility
-        if (maybeHost.isPresent() && maybePort.isPresent()) {
+        if (maybeHost.isPresent()) {
+            String port = getPort(conf, serverType);
             return Collections.singletonList(
-                    new Endpoint(
-                            maybeHost.get(),
-                            Integer.parseInt(maybePort.get()),
-                            DEFAULT_LISTENER_NAME));
+                    new Endpoint(maybeHost.get(), Integer.parseInt(port), DEFAULT_LISTENER_NAME));
         }
 
         throw new IllegalArgumentException(
@@ -195,10 +192,10 @@ private static Optional<String> getHost(Configuration conf, ServerType serverTyp
                 : conf.getOptional(ConfigOptions.TABLET_SERVER_HOST);
     }
 
-    private static Optional<String> getPort(Configuration conf, ServerType serverType) {
+    private static String getPort(Configuration conf, ServerType serverType) {
         return serverType == ServerType.COORDINATOR
-                ? conf.getOptional(ConfigOptions.COORDINATOR_PORT)
-                : conf.getOptional(ConfigOptions.TABLET_SERVER_PORT);
+                ? conf.get(ConfigOptions.COORDINATOR_PORT)
+                : conf.get(ConfigOptions.TABLET_SERVER_PORT);
     }
 
     /**

fluss-common/src/test/java/com/alibaba/fluss/cluster/EndpointTest.java

@@ -125,12 +125,23 @@ void testCoordinatorEndpointsCompatibility(ServerType serverType) {
         Configuration configuration = new Configuration();
         assertThatThrownBy(() -> Endpoint.loadBindEndpoints(configuration, serverType))
                 .hasMessageContaining("The 'bind.listeners' is not configured.");
-        configuration.setString(ConfigOptions.INTERNAL_LISTENER_NAME, "INTERNAL");
+
+        // if bind.listeners is not set, use deprecated [coordinator|tablet.server].host config
+        // options even though [coordinator|tablet.server].port is not set.
         configuration.setString(
                 serverType == ServerType.COORDINATOR
                         ? ConfigOptions.COORDINATOR_HOST
                         : ConfigOptions.TABLET_SERVER_HOST,
                 "my_host");
+        assertThat(Endpoint.loadBindEndpoints(configuration, serverType))
+                .containsExactlyElementsOf(
+                        Collections.singletonList(
+                                new Endpoint(
+                                        "my_host",
+                                        serverType == ServerType.COORDINATOR ? 9123 : 0,
+                                        "FLUSS")));
+
+        configuration.setString(ConfigOptions.INTERNAL_LISTENER_NAME, "INTERNAL");
         configuration.setString(
                 serverType == ServerType.COORDINATOR
                         ? ConfigOptions.COORDINATOR_PORT