Adapt local-cluster.sh and Dockerfile for bind.listeners config option

docker/docker-entrypoint.sh

@@ -15,28 +15,39 @@
 # limitations under the License.
 #
 
-CONF_FILE="${FLUSS_HOME}/conf/server.yaml"
+COMMON_CONF_FILE="${FLUSS_HOME}/conf/common.yaml"
 
 prepare_configuration() {
+    additional_conf_file="${FLUSS_HOME}/conf/$1"
+
+    # backward compatability: allow to use old [coordinator|tablet-server].host option in FLUSS_PROPERTIES
+    sed -i '/bind.listeners:/d' "${COMMON_CONF_FILE}" "$additional_conf_file"
+
     if [ -n "${FLUSS_PROPERTIES}" ]; then
-        echo "${FLUSS_PROPERTIES}" >> "${CONF_FILE}"
+        # copy over all configuration options of FLUSS_PROPERTIES to the common and the additional configuration file
+        # since we cannot tell which configuration options are specific and which are not
+        echo "#==============================================================================" | tee -a "${COMMON_CONF_FILE}" "$additional_conf_file"
+        echo "# Configuration Options from FLUSS_PROPERTIES Environment Variable" | tee -a "${COMMON_CONF_FILE}" "$additional_conf_file"
+        echo "#=============================================================================="  | tee -a "${COMMON_CONF_FILE}" "$additional_conf_file"
+        echo "${FLUSS_PROPERTIES}" | tee -a "${COMMON_CONF_FILE}" "$additional_conf_file"
     fi
-    envsubst < "${CONF_FILE}" > "${CONF_FILE}.tmp" && mv "${CONF_FILE}.tmp" "${CONF_FILE}"
+    envsubst < "${COMMON_CONF_FILE}" > "${COMMON_CONF_FILE}.tmp" && mv "${COMMON_CONF_FILE}.tmp" "${COMMON_CONF_FILE}"
+    envsubst < "$additional_conf_file" > "$additional_conf_file.tmp" && mv "$additional_conf_file.tmp" "$additional_conf_file"
 }
 
-prepare_configuration
-
 args=("$@")
 
 if [ "$1" = "help" ]; then
   printf "Usage: $(basename "$0") (coordinatorServer|tabletServer)\n"
   printf "    Or $(basename "$0") help\n\n"
   exit 0
 elif [ "$1" = "coordinatorServer" ]; then
+  prepare_configuration "coordinator-server.yaml"
   args=("${args[@]:1}")
   echo "Starting Coordinator Server"
   exec "$FLUSS_HOME/bin/coordinator-server.sh" start-foreground "${args[@]}"
 elif [ "$1" = "tabletServer" ]; then
+  prepare_configuration "tablet-server.yaml"
   args=("${args[@]:1}")
   echo "Starting Tablet Server"
   exec "$FLUSS_HOME/bin/tablet-server.sh" start-foreground "${args[@]}"

fluss-common/src/main/java/com/alibaba/fluss/config/GlobalConfiguration.java

@@ -29,6 +29,9 @@
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 
 /* This file is based on source code of Apache Flink Project (https://flink.apache.org/), licensed by the Apache
  * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
@@ -46,19 +49,20 @@ public class GlobalConfiguration {
 
     private static final Logger LOG = LoggerFactory.getLogger(GlobalConfiguration.class);
 
-    public static final String FLUSS_CONF_FILENAME = "server.yaml";
+    private static final String[] FLUSS_CONF_FILENAMES =
+            new String[] {"server.yaml", "common.yaml"};
 
     // --------------------------------------------------------------------------------------------
 
     private GlobalConfiguration() {}
 
     /**
-     * Load the configuration files from the config file specified by key {@link
+     * Loads the configuration files from the config file specified by key {@link
      * #SERVER_CONFIG_FILE} in {@code dynamicProperties}. If no config file is specified, it'll load
-     * the configuration from the specified {@code defaultConfigDir}.
+     * the common configuration from the specified {@code defaultConfigDir}.
      *
      * <p>If the {@code dynamicProperties} is not null, then it is added to the loaded
-     * configuration.
+     * configuration. Dynamic configuration options take precedence over file configuration options.
      *
      * @param defaultConfigDir directory to load the configuration from when no config file is
      *     specified in the dynamic properties
@@ -67,16 +71,42 @@ private GlobalConfiguration() {}
      */
     public static Configuration loadConfiguration(
             final String defaultConfigDir, @Nullable final Configuration dynamicProperties) {
+        return loadConfiguration(defaultConfigDir, Collections.emptyList(), dynamicProperties);
+    }
 
-        File yamlConfigFile = null;
+    /**
+     * Loads the configuration as described in {@link GlobalConfiguration#loadConfiguration(String,
+     * Configuration)}, but allows to specify a list of files that should be loaded from {@code
+     * defaultConfigDir} in addition to the common configuration, in case the user does not specify
+     * {@link #SERVER_CONFIG_FILE} in {@code dynamicProperties}.
+     *
+     * <p>The configuration files are read in the specified order. If multiple configuration files
+     * are given, and a configuration option is present in at least two of them, the configuration
+     * option in the <i>latest</i> configuration file that contains the option takes precedence.
+     * Additionally, dynamic configuration options take precedence over configuration options given
+     * in <i>any</i> file.
+     *
+     * @param defaultConfigDir see {@link GlobalConfiguration#loadConfiguration(String,
+     *     Configuration)}
+     * @param additionalDefaultFiles a list of additional config files that should be loaded from
+     *     defaultConfigDir that will be read in the given order
+     * @param dynamicProperties see {@link GlobalConfiguration#loadConfiguration(String,
+     *     Configuration)}
+     */
+    public static Configuration loadConfiguration(
+            final String defaultConfigDir,
+            final List<String> additionalDefaultFiles,
+            @Nullable final Configuration dynamicProperties) {
+
+        List<File> yamlConfigFiles = new ArrayList<>();
 
         // first, try to get the config file name from the dynamic properties
         // user passed
         if (dynamicProperties != null && dynamicProperties.contains(SERVER_CONFIG_FILE)) {
             // get the config file name passed by user
             String configFileName = dynamicProperties.getString(SERVER_CONFIG_FILE);
             dynamicProperties.removeConfig(SERVER_CONFIG_FILE);
-            yamlConfigFile = new File(configFileName);
+            File yamlConfigFile = new File(configFileName);
             if (!yamlConfigFile.exists() && !yamlConfigFile.isFile()) {
                 throw new IllegalConfigurationException(
                         "The given configuration file name '"
@@ -85,9 +115,10 @@ public static Configuration loadConfiguration(
                                 + yamlConfigFile.getAbsolutePath()
                                 + ") does not describe an existing file.");
             }
+            yamlConfigFiles.add(yamlConfigFile);
         }
 
-        if (yamlConfigFile == null) {
+        if (yamlConfigFiles.isEmpty()) {
             // try to load from the default conf dir
             if (defaultConfigDir == null) {
                 throw new IllegalArgumentException(
@@ -102,11 +133,41 @@ public static Configuration loadConfiguration(
                                 + confDirFile.getAbsolutePath()
                                 + ") does not describe an existing directory.");
             }
-            // get Fluss yaml configuration file from dir
-            yamlConfigFile = new File(confDirFile, FLUSS_CONF_FILENAME);
+
+            // get Fluss yaml configuration files from dir
+            final File serverYamlFile = new File(confDirFile, FLUSS_CONF_FILENAMES[0]);
+            final File commonYamlFile = new File(confDirFile, FLUSS_CONF_FILENAMES[1]);
+
+            // 1. check if old and new configuration files are mixed which is not supported
+            if (serverYamlFile.exists() && commonYamlFile.exists()) {
+                throw new IllegalConfigurationException(
+                        "Only one of "
+                                + FLUSS_CONF_FILENAMES[0]
+                                + " and "
+                                + FLUSS_CONF_FILENAMES[1]
+                                + " may be specified.");
+            }
+
+            // 2. backward compatability, use server.yaml
+            if (serverYamlFile.exists()) {
+                yamlConfigFiles.add(new File(confDirFile, FLUSS_CONF_FILENAMES[0]));
+            }
+
+            // 3. latest configuration setup: load common.yaml and additionally specified, dedicated
+            // configuration files
+            if (commonYamlFile.exists()) {
+                yamlConfigFiles.add(new File(confDirFile, FLUSS_CONF_FILENAMES[1]));
+
+                for (String additionalDefaultFile : additionalDefaultFiles) {
+                    yamlConfigFiles.add(new File(confDirFile, additionalDefaultFile));
+                }
+            }
         }
 
-        Configuration configuration = loadYAMLResource(yamlConfigFile);
+        Configuration configuration = loadYAMLResource(yamlConfigFiles.remove(0));
+        for (File yamlConfigFile : yamlConfigFiles) {
+            configuration.addAll(loadYAMLResource(yamlConfigFile));
+        }
 
         logConfiguration("Loading", configuration);
 

fluss-common/src/test/java/com/alibaba/fluss/config/GlobalConfigurationTest.java

@@ -0,0 +1,200 @@
+/*
+ *  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.config;
+
+import com.alibaba.fluss.exception.IllegalConfigurationException;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link com.alibaba.fluss.config.GlobalConfiguration}. */
+public class GlobalConfigurationTest {
+
+    private static final String OLD_COMMON_CONFIG_FILE_NAME = "server.yaml";
+    private static final String NEW_COMMON_CONFIG_FILE_NAME = "common.yaml";
+
+    @Test
+    void testLoadConfigurationWithoutAdditionalFiles(@TempDir Path tempFolder) throws Exception {
+        String confDir = tempFolder.toAbsolutePath().toString();
+
+        // backward compatability for old common configuration file
+        Path serverYaml = tempFolder.resolve(OLD_COMMON_CONFIG_FILE_NAME);
+        Files.write(serverYaml, Collections.singleton("coordinator.host: localhost"));
+
+        Configuration configuration = GlobalConfiguration.loadConfiguration(confDir, null);
+        assertThat(configuration.get(ConfigOptions.COORDINATOR_HOST)).isEqualTo("localhost");
+
+        // backward compatability for old common configuration file + precedence for dynamic
+        // properties
+        Configuration dynamicConfig = new Configuration();
+        dynamicConfig.set(ConfigOptions.COORDINATOR_HOST, "example.com");
+        configuration = GlobalConfiguration.loadConfiguration(confDir, dynamicConfig);
+        assertThat(configuration.get(ConfigOptions.COORDINATOR_HOST)).isEqualTo("example.com");
+
+        // old and new common configuration file should not be present at the same time
+        Path commonYaml = tempFolder.resolve(NEW_COMMON_CONFIG_FILE_NAME);
+        Files.write(commonYaml, Collections.singleton("bind.listeners: FLUSS://localhost:9124"));
+        assertThatThrownBy(() -> GlobalConfiguration.loadConfiguration(confDir, null))
+                .isInstanceOf(IllegalConfigurationException.class)
+                .hasMessageContaining("Only one of");
+
+        // new common configuration file
+        Files.delete(serverYaml);
+        configuration = GlobalConfiguration.loadConfiguration(confDir, null);
+        assertThat(configuration.get(ConfigOptions.BIND_LISTENERS))
+                .isEqualTo("FLUSS://localhost:9124");
+        assertThat(configuration.get(ConfigOptions.COORDINATOR_HOST)).isEqualTo(null);
+
+        // new common configuration file + precedence for dynamic properties
+        dynamicConfig = new Configuration();
+        dynamicConfig.set(ConfigOptions.BIND_LISTENERS, "FLUSS://example.com:9124");
+        configuration = GlobalConfiguration.loadConfiguration(confDir, dynamicConfig);
+        assertThat(configuration.get(ConfigOptions.BIND_LISTENERS))
+                .isEqualTo("FLUSS://example.com:9124");
+        assertThat(configuration.get(ConfigOptions.COORDINATOR_HOST)).isEqualTo(null);
+    }
+
+    @Test
+    void testLoadConfigurationWithAdditionalFileSimple(@TempDir Path tempFolder) throws Exception {
+        String confDir = tempFolder.toAbsolutePath().toString();
+
+        // additional file should be read when old config is present
+        Path serverYaml = tempFolder.resolve(OLD_COMMON_CONFIG_FILE_NAME);
+        Files.write(
+                serverYaml,
+                Arrays.asList(
+                        "remote.data.dir: /tmp/fluss-remote-data",
+                        "bind.listeners: FLUSS://localhost:9124"));
+        Path additionalYaml = tempFolder.resolve("additional-file.yaml");
+        Files.write(
+                additionalYaml,
+                Arrays.asList(
+                        "bind.listeners: FLUSS://example.com:9124",
+                        "default.replication.factor: 5"));
+        Configuration configuration =
+                GlobalConfiguration.loadConfiguration(
+                        confDir, Collections.singletonList("additional-file.yaml"), null);
+
+        assertThat(configuration.get(ConfigOptions.REMOTE_DATA_DIR))
+                .isEqualTo("/tmp/fluss-remote-data");
+        assertThat(configuration.get(ConfigOptions.BIND_LISTENERS))
+                .isEqualTo("FLUSS://localhost:9124");
+        assertThat(configuration.get(ConfigOptions.DEFAULT_REPLICATION_FACTOR))
+                .isEqualTo(ConfigOptions.DEFAULT_REPLICATION_FACTOR.defaultValue());
+
+        Files.delete(serverYaml);
+
+        // additional file should be read when new config is present
+        Path commonYaml = tempFolder.resolve(NEW_COMMON_CONFIG_FILE_NAME);
+        Files.write(
+                commonYaml,
+                Arrays.asList(
+                        "remote.data.dir: /tmp/fluss-remote-data",
+                        "bind.listeners: FLUSS://localhost:9124"));
+
+        configuration =
+                GlobalConfiguration.loadConfiguration(
+                        confDir, Collections.singletonList("additional-file.yaml"), null);
+
+        assertThat(configuration.get(ConfigOptions.REMOTE_DATA_DIR))
+                .isEqualTo("/tmp/fluss-remote-data");
+        assertThat(configuration.get(ConfigOptions.BIND_LISTENERS))
+                .isEqualTo("FLUSS://example.com:9124");
+        assertThat(configuration.get(ConfigOptions.DEFAULT_REPLICATION_FACTOR)).isEqualTo(5);
+    }
+
+    @Test
+    void testLoadConfigurationWithAdditionalFilesRespectsPrecedence(@TempDir Path tempFolder)
+            throws Exception {
+        String confDir = tempFolder.toAbsolutePath().toString();
+
+        ConfigOption<Integer> keyWillNotBeOverwritten =
+                ConfigBuilder.key("willNotBeOverwritten").intType().noDefaultValue();
+        ConfigOption<Integer> keyWillBeOverwrittenByFirstAdditionalFile =
+                ConfigBuilder.key("willBeOverwrittenByFirstAdditionalFile")
+                        .intType()
+                        .noDefaultValue();
+        ConfigOption<Integer> keyWillBeOverwrittenBySecondAdditionalFile =
+                ConfigBuilder.key("willBeOverwrittenBySecondAdditionalFile")
+                        .intType()
+                        .noDefaultValue();
+        ConfigOption<Integer> keyWillBeOverwrittenByDynamicProperty =
+                ConfigBuilder.key("willBeOverwrittenByDynamicProperty").intType().noDefaultValue();
+
+        Path commonYaml = tempFolder.resolve(NEW_COMMON_CONFIG_FILE_NAME);
+        Files.write(
+                commonYaml,
+                Arrays.asList(
+                        "willNotBeOverwritten: 0",
+                        "willBeOverwrittenByFirstAdditionalFile: 1",
+                        "willBeOverwrittenBySecondAdditionalFile: 2"));
+
+        // only common, no additional file
+        Configuration configuration =
+                GlobalConfiguration.loadConfiguration(confDir, Collections.emptyList(), null);
+        assertThat(configuration.get(keyWillNotBeOverwritten)).isEqualTo(0);
+        assertThat(configuration.get(keyWillBeOverwrittenByFirstAdditionalFile)).isEqualTo(1);
+        assertThat(configuration.get(keyWillBeOverwrittenBySecondAdditionalFile)).isEqualTo(2);
+
+        // first additional file
+        Path firstAdditionalYaml = tempFolder.resolve("first-additional.yaml");
+        Files.write(
+                firstAdditionalYaml,
+                Arrays.asList(
+                        "willBeOverwrittenByFirstAdditionalFile: 11",
+                        "willBeOverwrittenByDynamicProperty: 12"));
+        List<String> additionalFiles = new ArrayList<>(2);
+        additionalFiles.add("first-additional.yaml");
+        configuration = GlobalConfiguration.loadConfiguration(confDir, additionalFiles, null);
+        assertThat(configuration.get(keyWillNotBeOverwritten)).isEqualTo(0);
+        assertThat(configuration.get(keyWillBeOverwrittenByFirstAdditionalFile)).isEqualTo(11);
+        assertThat(configuration.get(keyWillBeOverwrittenBySecondAdditionalFile)).isEqualTo(2);
+        assertThat(configuration.get(keyWillBeOverwrittenByDynamicProperty)).isEqualTo(12);
+
+        // second additional file
+        Path secondAdditionalYaml = tempFolder.resolve("second-additional.yaml");
+        Files.write(
+                secondAdditionalYaml,
+                Collections.singleton("willBeOverwrittenBySecondAdditionalFile: 20"));
+        additionalFiles.add("second-additional.yaml");
+        configuration = GlobalConfiguration.loadConfiguration(confDir, additionalFiles, null);
+        assertThat(configuration.get(keyWillNotBeOverwritten)).isEqualTo(0);
+        assertThat(configuration.get(keyWillBeOverwrittenByFirstAdditionalFile)).isEqualTo(11);
+        assertThat(configuration.get(keyWillBeOverwrittenBySecondAdditionalFile)).isEqualTo(20);
+        assertThat(configuration.get(keyWillBeOverwrittenByDynamicProperty)).isEqualTo(12);
+
+        // dynamic property
+        Configuration dynamicProperties = new Configuration();
+        dynamicProperties.set(keyWillBeOverwrittenByDynamicProperty, 32);
+        configuration =
+                GlobalConfiguration.loadConfiguration(confDir, additionalFiles, dynamicProperties);
+        assertThat(configuration.get(keyWillNotBeOverwritten)).isEqualTo(0);
+        assertThat(configuration.get(keyWillBeOverwrittenByFirstAdditionalFile)).isEqualTo(11);
+        assertThat(configuration.get(keyWillBeOverwrittenBySecondAdditionalFile)).isEqualTo(20);
+        assertThat(configuration.get(keyWillBeOverwrittenByDynamicProperty)).isEqualTo(32);
+    }
+}