Introduce LakeTieringDecoratorPlugin to make it pluggable

Author: luoyuxia

fluss-flink/fluss-flink-tiering/src/main/java/org/apache/fluss/flink/tiering/FlussLakeTiering.java

@@ -27,6 +27,7 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
 import java.util.Map;
+import java.util.ServiceLoader;
 
 import static org.apache.flink.runtime.executiongraph.failover.FailoverStrategyFactoryLoader.FULL_RESTART_STRATEGY_NAME;
 import static org.apache.fluss.flink.tiering.source.TieringSourceOptions.DATA_LAKE_CONFIG_PREFIX;
@@ -39,10 +40,10 @@
  * <p>This class is responsible for parsing configuration parameters, initializing the Flink
  * execution environment, and coordinating the construction of the tiering pipeline.
  *
- * <p>Design Motivation: By decoupling the logic from {@link FlussLakeTieringEntrypoint} into this
- * class, extensibility is significantly improved. Developers can now extend this class to customize
- * configuration extraction (e.g., injecting internal security tokens) without duplicating the core
- * entrypoint boilerplate.
+ * <p>Extensibility: Customization of Flink execution environment and configurations is supported
+ * through the {@link LakeTieringDecoratorPlugin} SPI mechanism. Different environments (e.g.,
+ * internal vs. public cloud) can provide their own decorator implementations without touching core
+ * code, ensuring better maintainability, testability, and forward compatibility.
  */
 public class FlussLakeTiering {
 
@@ -51,17 +52,17 @@ public class FlussLakeTiering {
 
     protected final StreamExecutionEnvironment execEnv;
     protected final String dataLake;
-    protected final Map<String, String> flussConfigMap;
-    protected final Map<String, String> lakeConfigMap;
-    protected final Map<String, String> lakeTieringConfigMap;
+    protected final Configuration flussConfig;
+    protected final Configuration lakeConfig;
+    protected final Configuration lakeTieringConfig;
 
     public FlussLakeTiering(String[] args) {
         // parse params
         final MultipleParameterToolAdapter params = MultipleParameterToolAdapter.fromArgs(args);
         Map<String, String> paramsMap = params.toMap();
 
         // extract fluss config
-        flussConfigMap = extractAndRemovePrefix(paramsMap, FLUSS_CONF_PREFIX);
+        Map<String, String> flussConfigMap = extractAndRemovePrefix(paramsMap, FLUSS_CONF_PREFIX);
         // we need to get bootstrap.servers
         String bootstrapServers = flussConfigMap.get(ConfigOptions.BOOTSTRAP_SERVERS.key());
         if (bootstrapServers == null) {
@@ -70,6 +71,7 @@ public FlussLakeTiering(String[] args) {
                             "The bootstrap server to fluss is not configured, please configure %s",
                             FLUSS_CONF_PREFIX + ConfigOptions.BOOTSTRAP_SERVERS.key()));
         }
+        this.flussConfig = Configuration.fromMap(flussConfigMap);
 
         dataLake = paramsMap.get(ConfigOptions.DATALAKE_FORMAT.key());
         if (dataLake == null) {
@@ -78,12 +80,15 @@ public FlussLakeTiering(String[] args) {
         }
 
         // extract lake config
-        lakeConfigMap =
+        Map<String, String> lakeConfigMap =
                 extractAndRemovePrefix(
                         paramsMap, String.format("%s%s.", DATA_LAKE_CONFIG_PREFIX, dataLake));
+        this.lakeConfig = Configuration.fromMap(lakeConfigMap);
 
         // extract tiering service config
-        lakeTieringConfigMap = extractPrefix(paramsMap, LAKE_TIERING_CONFIG_PREFIX);
+        Map<String, String> lakeTieringConfigMap =
+                extractPrefix(paramsMap, LAKE_TIERING_CONFIG_PREFIX);
+        this.lakeTieringConfig = Configuration.fromMap(lakeTieringConfigMap);
 
         // now, we must use full restart strategy if any task is failed,
         // since committer is stateless, if tiering committer is failover, committer
@@ -97,18 +102,45 @@ public FlussLakeTiering(String[] args) {
     }
 
     protected void run() throws Exception {
+        // Load and apply all available decorator plugins
+        loadAndApplyDecoratorPlugins();
+
         // build and run lake tiering job
         JobClient jobClient =
                 LakeTieringJobBuilder.newBuilder(
-                                execEnv,
-                                Configuration.fromMap(flussConfigMap),
-                                Configuration.fromMap(lakeConfigMap),
-                                Configuration.fromMap(lakeTieringConfigMap),
-                                dataLake)
+                                execEnv, flussConfig, lakeConfig, lakeTieringConfig, dataLake)
                         .build();
 
         System.out.printf(
                 "Starting data tiering service from Fluss to %s, jobId is %s.....%n",
                 dataLake, jobClient.getJobID());
     }
+
+    /**
+     * Loads all available {@link LakeTieringDecoratorPlugin} implementations and applies their
+     * decorators in sequence.
+     *
+     * <p>All available plugins will be loaded and their decorators will be called in the order they
+     * are discovered by the ServiceLoader. This allows multiple decorators to be applied
+     * sequentially, where each decorator can further customize the Flink execution environment and
+     * configurations.
+     */
+    protected void loadAndApplyDecoratorPlugins() {
+        ServiceLoader<LakeTieringDecoratorPlugin> serviceLoader =
+                ServiceLoader.load(
+                        LakeTieringDecoratorPlugin.class,
+                        LakeTieringDecoratorPlugin.class.getClassLoader());
+        serviceLoader.stream()
+                .map(ServiceLoader.Provider::get)
+                .forEach(
+                        plugin -> {
+                            String identifier = plugin.identifier();
+                            System.out.printf(
+                                    "Applying LakeTieringDecoratorPlugin with identifier: %s%n",
+                                    identifier);
+                            LakeTieringDecorator decorator = plugin.createLakeTieringDecorator();
+                            decorator.decorate(
+                                    execEnv, flussConfig, lakeConfig, lakeTieringConfig, dataLake);
+                        });
+    }
 }

fluss-flink/fluss-flink-tiering/src/main/java/org/apache/fluss/flink/tiering/LakeTieringDecorator.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.fluss.flink.tiering;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.config.Configuration;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+/**
+ * Interface for customizing Flink execution environment and configurations for lake tiering jobs.
+ *
+ * <p>Implementations of this interface can customize the Flink execution environment and
+ * configurations as needed for specific deployment environments (e.g., injecting internal security
+ * tokens, setting environment-specific configurations).
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public interface LakeTieringDecorator {
+
+    /**
+     * Customizes the Flink execution environment and configurations for the lake tiering job.
+     *
+     * <p>This method is called before building the tiering job, allowing implementations to modify
+     * the Flink execution environment or any of the provided configurations as needed.
+     *
+     * @param env the Flink StreamExecutionEnvironment to customize
+     * @param flussConfig the Fluss configuration (may be modified)
+     * @param dataLakeConfig the data lake configuration (may be modified)
+     * @param lakeTieringConfig the lake tiering configuration (may be modified)
+     * @param dataLakeFormat the data lake format identifier (e.g., "paimon", "iceberg")
+     */
+    void decorate(
+            StreamExecutionEnvironment env,
+            Configuration flussConfig,
+            Configuration dataLakeConfig,
+            Configuration lakeTieringConfig,
+            String dataLakeFormat);
+}

fluss-flink/fluss-flink-tiering/src/main/java/org/apache/fluss/flink/tiering/LakeTieringDecoratorPlugin.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.fluss.flink.tiering;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.plugin.Plugin;
+
+/**
+ * A Plugin to create instances of {@link LakeTieringDecorator}.
+ *
+ * <p>This plugin mechanism allows different environments (e.g., internal vs. public cloud) to
+ * provide their own decorator implementations for customizing Flink execution environment and
+ * configurations without touching core code, ensuring better maintainability, testability, and
+ * forward compatibility.
+ *
+ * <p>Multiple different plugin implementations can be loaded and applied simultaneously. However,
+ * the loading order of plugins is not guaranteed and may vary between runs. Plugin implementations
+ * should not depend on the loading order and should be designed to work correctly regardless of
+ * when they are applied relative to other plugins.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public interface LakeTieringDecoratorPlugin extends Plugin {
+
+    /**
+     * Returns a unique identifier among {@link LakeTieringDecoratorPlugin} implementations.
+     *
+     * @return the identifier
+     */
+    String identifier();
+
+    /**
+     * Creates a new instance of {@link LakeTieringDecorator}.
+     *
+     * @return the lake tiering decorator instance
+     */
+    LakeTieringDecorator createLakeTieringDecorator();
+}

fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/FlussLakeTieringTest.java

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.fluss.flink.tiering;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link FlussLakeTiering}. */
+class FlussLakeTieringTest {
+
+    @Test
+    void testMultipleDecoratorsApplied() {
+        String[] args = {
+            "--fluss.bootstrap.servers",
+            "localhost:9123",
+            "--datalake.paimon.metastore",
+            "rest",
+            "--datalake.paimon.warehous",
+            "fluss_test",
+            "--datalake.format",
+            "paimon",
+        };
+        FlussLakeTiering tiering = new FlussLakeTiering(args);
+
+        // Apply decorators - method now uses internal Configuration fields
+        tiering.loadAndApplyDecoratorPlugins();
+
+        // Verify that TestLakeTieringDecorator modified the configurations
+        assertThat(tiering.flussConfig.getRawValue(TestLakeTieringDecorator.TEST_FLUSS_CONFIG_KEY))
+                .isPresent()
+                .contains(TestLakeTieringDecorator.TEST_FLUSS_CONFIG_VALUE);
+        assertThat(
+                        tiering.lakeConfig.getRawValue(
+                                TestLakeTieringDecorator.TEST_DATA_LAKE_CONFIG_KEY))
+                .isPresent()
+                .contains(TestLakeTieringDecorator.TEST_DATA_LAKE_CONFIG_VALUE);
+        assertThat(
+                        tiering.lakeTieringConfig.getRawValue(
+                                TestLakeTieringDecorator.TEST_LAKE_TIERING_CONFIG_KEY))
+                .isPresent()
+                .contains(TestLakeTieringDecorator.TEST_LAKE_TIERING_CONFIG_VALUE);
+
+        // Verify that TestLakeTieringDecorator2 also modified the configurations
+        assertThat(tiering.flussConfig.getRawValue(TestLakeTieringDecorator2.TEST_FLUSS_CONFIG_KEY))
+                .isPresent()
+                .contains(TestLakeTieringDecorator2.TEST_FLUSS_CONFIG_VALUE);
+        assertThat(
+                        tiering.lakeConfig.getRawValue(
+                                TestLakeTieringDecorator2.TEST_DATA_LAKE_CONFIG_KEY))
+                .isPresent()
+                .contains(TestLakeTieringDecorator2.TEST_DATA_LAKE_CONFIG_VALUE);
+        assertThat(
+                        tiering.lakeTieringConfig.getRawValue(
+                                TestLakeTieringDecorator2.TEST_LAKE_TIERING_CONFIG_KEY))
+                .isPresent()
+                .contains(TestLakeTieringDecorator2.TEST_LAKE_TIERING_CONFIG_VALUE);
+    }
+}

fluss-flink/fluss-flink-tiering/src/test/java/org/apache/fluss/flink/tiering/TestLakeTieringDecorator.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.fluss.flink.tiering;
+
+import org.apache.fluss.config.Configuration;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+/** Test implementation of {@link LakeTieringDecorator}. */
+class TestLakeTieringDecorator implements LakeTieringDecorator {
+
+    static final String TEST_FLUSS_CONFIG_KEY = "test.decorator1.fluss.config";
+    static final String TEST_FLUSS_CONFIG_VALUE = "decorator1-value";
+    static final String TEST_DATA_LAKE_CONFIG_KEY = "test.decorator1.datalake.config";
+    static final String TEST_DATA_LAKE_CONFIG_VALUE = "decorator1-datalake-value";
+    static final String TEST_LAKE_TIERING_CONFIG_KEY = "test.decorator1.tiering.config";
+    static final String TEST_LAKE_TIERING_CONFIG_VALUE = "decorator1-tiering-value";
+
+    @Override
+    public void decorate(
+            StreamExecutionEnvironment env,
+            Configuration flussConfig,
+            Configuration dataLakeConfig,
+            Configuration lakeTieringConfig,
+            String dataLakeFormat) {
+        // Modify configurations to verify decorator is called
+        flussConfig.setString(TEST_FLUSS_CONFIG_KEY, TEST_FLUSS_CONFIG_VALUE);
+        dataLakeConfig.setString(TEST_DATA_LAKE_CONFIG_KEY, TEST_DATA_LAKE_CONFIG_VALUE);
+        lakeTieringConfig.setString(TEST_LAKE_TIERING_CONFIG_KEY, TEST_LAKE_TIERING_CONFIG_VALUE);
+    }
+}