[HUDI-3336][HUDI-FLINK]Support custom hadoop config for flink

Author: cuibo01

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java

@@ -738,16 +738,14 @@ public static org.apache.hadoop.conf.Configuration getParquetConf(
     return copy;
   }
 
-  // Keep the redundant to avoid too many modifications.
+  /**
+   * Create a new hadoop configuration that is initialized with the given flink configuration.
+   */
   public static org.apache.hadoop.conf.Configuration getHadoopConf(Configuration conf) {
-    if (conf == null) {
-      return FlinkClientUtil.getHadoopConf();
-    } else {
-      org.apache.hadoop.conf.Configuration hadoopConf = FlinkClientUtil.getHadoopConf();
-      Map<String, String> options = getPropertiesWithPrefix(conf.toMap(), HADOOP_PREFIX);
-      options.forEach((k, v) -> hadoopConf.set(k, v));
-      return hadoopConf;
-    }
+    org.apache.hadoop.conf.Configuration hadoopConf = FlinkClientUtil.getHadoopConf();
+    Map<String, String> options = getPropertiesWithPrefix(conf.toMap(), HADOOP_PREFIX);
+    options.forEach((k, v) -> hadoopConf.set(k, v));
+    return hadoopConf;
   }
 
   /**
@@ -771,9 +769,7 @@ private static boolean hasPropertyOptions(Map<String, String> options, String pr
 
   /**
    * Creates a new configuration that is initialized with the options of the given map.
-   * @deprecated Use {@link Configuration#fromMap(Map)} instead.
    */
-  @Deprecated
   public static Configuration fromMap(Map<String, String> map) {
     final Configuration configuration = new Configuration();
     map.forEach(configuration::setString);

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/BulkInsertWriteFunction.java

@@ -113,7 +113,7 @@ public BulkInsertWriteFunction(Configuration config, RowType rowType) {
   public void open(Configuration parameters) throws IOException {
     this.taskID = getRuntimeContext().getIndexOfThisSubtask();
     this.writeClient = StreamerUtil.createWriteClient(this.config, getRuntimeContext());
-    this.ckpMetadata = CkpMetadata.getInstance(config.getString(FlinkOptions.PATH));
+    this.ckpMetadata = CkpMetadata.getInstance(config);
     this.initInstant = lastPendingInstant();
     sendBootstrapEvent();
     initWriterHelper();

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/meta/CkpMetadata.java

@@ -18,9 +18,11 @@
 
 package org.apache.hudi.sink.meta;
 
+import org.apache.flink.configuration.Configuration;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.util.StreamerUtil;
 
@@ -70,8 +72,8 @@ public class CkpMetadata implements Serializable {
   private List<CkpMessage> messages;
   private List<String> instantCache;
 
-  private CkpMetadata(String basePath) {
-    this(FSUtils.getFs(basePath, StreamerUtil.getHadoopConf()), basePath);
+  private CkpMetadata(Configuration config) {
+    this(FSUtils.getFs(config.getString(FlinkOptions.PATH), FlinkOptions.getHadoopConf(config)), config.getString(FlinkOptions.PATH));
   }
 
   private CkpMetadata(FileSystem fs, String basePath) {
@@ -196,8 +198,8 @@ public boolean isAborted(String instant) {
   // -------------------------------------------------------------------------
   //  Utilities
   // -------------------------------------------------------------------------
-  public static CkpMetadata getInstance(String basePath) {
-    return new CkpMetadata(basePath);
+  public static CkpMetadata getInstance(Configuration config) {
+    return new CkpMetadata(config);
   }
 
   public static CkpMetadata getInstance(FileSystem fs, String basePath) {

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java

@@ -140,15 +140,6 @@ public static DFSPropertiesConfiguration readConfig(org.apache.hadoop.conf.Confi
     return conf;
   }
 
-  /**
-   * Keep the redundant to avoid too many modifications.
-   * @deprecated Use {@link FlinkOptions#getHadoopConf(Configuration)} instead.
-   */
-  @Deprecated
-  public static org.apache.hadoop.conf.Configuration getHadoopConf() {
-    return FlinkOptions.getHadoopConf(new Configuration());
-  }
-
   /**
    * Mainly used for tests.
    */
@@ -352,14 +343,6 @@ public static HoodieTableMetaClient createMetaClient(String basePath, org.apache
     return HoodieTableMetaClient.builder().setBasePath(basePath).setConf(hadoopConf).build();
   }
 
-  /**
-   * Creates the meta client.
-   */
-  @Deprecated
-  public static HoodieTableMetaClient createMetaClient(String basePath) {
-    return createMetaClient(basePath, FlinkClientUtil.getHadoopConf());
-  }
-
   /**
    * Creates the meta client.
    */

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/meta/TestCkpMetadata.java

@@ -19,6 +19,7 @@
 package org.apache.hudi.sink.meta;
 
 import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.util.StreamerUtil;
 import org.apache.hudi.utils.TestConfigurations;
 
@@ -47,7 +48,7 @@ public class TestCkpMetadata {
   @BeforeEach
   public void beforeEach() throws Exception {
     String basePath = tempFile.getAbsolutePath();
-    FileSystem fs = FSUtils.getFs(tempFile.getAbsolutePath(), StreamerUtil.getHadoopConf());
+    FileSystem fs = FSUtils.getFs(tempFile.getAbsolutePath(), FlinkOptions.getHadoopConf(new Configuration()));
 
     Configuration conf = TestConfigurations.getDefaultConf(basePath);
     StreamerUtil.initTableIfNotExists(conf);