[HUDI-8093] Replace Hadoop Configuration with StorageConfiguration in hudi-client-common module (apache#11792)

Co-authored-by: Shawn Chang <[email protected]>

Author: CTTY

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java

@@ -41,15 +41,14 @@
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.exception.HoodieWriteConflictException;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
 import org.apache.hudi.metadata.HoodieTableMetadataWriter;
 import org.apache.hudi.metrics.HoodieMetrics;
 import org.apache.hudi.storage.HoodieStorage;
 import org.apache.hudi.storage.HoodieStorageUtils;
+import org.apache.hudi.storage.StorageConfiguration;
 import org.apache.hudi.table.HoodieTable;
 
 import com.codahale.metrics.Timer;
-import org.apache.hadoop.conf.Configuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -70,7 +69,7 @@ public abstract class BaseHoodieClient implements Serializable, AutoCloseable {
   private static final long serialVersionUID = 1L;
   protected final transient HoodieStorage storage;
   protected final transient HoodieEngineContext context;
-  protected final transient Configuration hadoopConf;
+  protected final transient StorageConfiguration<?> storageConf;
   protected final transient HoodieMetrics metrics;
   protected final HoodieWriteConfig config;
   protected final String basePath;
@@ -92,8 +91,8 @@ protected BaseHoodieClient(HoodieEngineContext context, HoodieWriteConfig client
 
   protected BaseHoodieClient(HoodieEngineContext context, HoodieWriteConfig clientConfig,
       Option<EmbeddedTimelineService> timelineServer) {
-    this.hadoopConf = context.getStorageConf().unwrapAs((Configuration.class));
-    this.storage = HoodieStorageUtils.getStorage(clientConfig.getBasePath(), HadoopFSUtils.getStorageConf(hadoopConf));
+    this.storageConf = context.getStorageConf();
+    this.storage = HoodieStorageUtils.getStorage(clientConfig.getBasePath(), storageConf);
     this.context = context;
     this.basePath = clientConfig.getBasePath();
     this.config = clientConfig;
@@ -105,7 +104,7 @@ protected BaseHoodieClient(HoodieEngineContext context, HoodieWriteConfig client
     this.metrics = new HoodieMetrics(config, storage);
     this.txnManager = new TransactionManager(config, storage);
     this.timeGenerator = TimeGenerators.getTimeGenerator(
-        config.getTimeGeneratorConfig(), HadoopFSUtils.getStorageConf(hadoopConf));
+        config.getTimeGeneratorConfig(), storageConf);
     startEmbeddedServerView();
     initWrapperFSMetrics();
     runClientInitCallbacks();
@@ -183,8 +182,10 @@ protected void initWrapperFSMetrics() {
 
   protected HoodieTableMetaClient createMetaClient(boolean loadActiveTimelineOnLoad) {
     return HoodieTableMetaClient.builder()
-        .setConf(HadoopFSUtils.getStorageConfWithCopy(hadoopConf)).setBasePath(config.getBasePath())
-        .setLoadActiveTimelineOnLoad(loadActiveTimelineOnLoad).setConsistencyGuardConfig(config.getConsistencyGuardConfig())
+        .setConf(storageConf.newInstance())
+        .setBasePath(config.getBasePath())
+        .setLoadActiveTimelineOnLoad(loadActiveTimelineOnLoad)
+        .setConsistencyGuardConfig(config.getConsistencyGuardConfig())
         .setLayoutVersion(Option.of(new TimelineLayoutVersion(config.getTimelineLayoutVersion())))
         .setTimeGeneratorConfig(config.getTimeGeneratorConfig())
         .setFileSystemRetryConfig(config.getFileSystemRetryConfig())

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java

@@ -59,14 +59,14 @@
 import org.apache.hudi.exception.HoodieRollbackException;
 import org.apache.hudi.metadata.HoodieTableMetadata;
 import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.storage.StorageConfiguration;
 import org.apache.hudi.table.HoodieTable;
 import org.apache.hudi.table.action.HoodieWriteMetadata;
 import org.apache.hudi.table.action.compact.CompactHelpers;
 import org.apache.hudi.table.action.rollback.RollbackUtils;
 import org.apache.hudi.table.marker.WriteMarkersFactory;
 
 import com.codahale.metrics.Timer;
-import org.apache.hadoop.conf.Configuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -165,7 +165,7 @@ protected void setPendingInflightAndRequestedInstants(Set<String> pendingInfligh
   protected void preCommit(HoodieCommitMetadata metadata) {
     // Create a Hoodie table after startTxn which encapsulated the commits and files visible.
     // Important to create this after the lock to ensure the latest commits show up in the timeline without need for reload
-    HoodieTable table = createTable(config, hadoopConf);
+    HoodieTable table = createTable(config, storageConf);
     resolveWriteConflict(table, metadata, this.pendingInflightAndRequestedInstants);
   }
 
@@ -198,7 +198,7 @@ private void inlineCompaction(HoodieTable table, Option<Map<String, String>> ext
    * @return Collection of Write Status
    */
   protected HoodieWriteMetadata<O> logCompact(String logCompactionInstantTime, boolean shouldComplete) {
-    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf().unwrapAs(Configuration.class));
+    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf());
 
     // Check if a commit or compaction instant with a greater timestamp is on the timeline.
     // If an instant is found then abort log compaction, since it is no longer needed.
@@ -288,7 +288,7 @@ public Option<String> scheduleCompaction(Option<Map<String, String>> extraMetada
    * @return Collection of Write Status
    */
   protected HoodieWriteMetadata<O> compact(String compactionInstantTime, boolean shouldComplete) {
-    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf().unwrapAs(Configuration.class));
+    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf());
     HoodieTimeline pendingCompactionTimeline = table.getActiveTimeline().filterPendingCompactionTimeline();
     HoodieInstant inflightInstant = HoodieTimeline.getCompactionInflightInstant(compactionInstantTime);
     if (pendingCompactionTimeline.containsInstant(inflightInstant)) {
@@ -313,7 +313,7 @@ protected HoodieWriteMetadata<O> compact(String compactionInstantTime, boolean s
    */
   public void commitCompaction(String compactionInstantTime, HoodieCommitMetadata metadata, Option<Map<String, String>> extraMetadata) {
     extraMetadata.ifPresent(m -> m.forEach(metadata::addMetadata));
-    completeCompaction(metadata, createTable(config, context.getStorageConf().unwrapAs(Configuration.class)), compactionInstantTime);
+    completeCompaction(metadata, createTable(config, context.getStorageConf()), compactionInstantTime);
   }
 
   /**
@@ -446,7 +446,7 @@ public boolean scheduleClusteringAtInstant(String instantTime, Option<Map<String
    * @return Collection of Write Status
    */
   public HoodieWriteMetadata<O> cluster(String clusteringInstant, boolean shouldComplete) {
-    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf().unwrapAs(Configuration.class));
+    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf());
     HoodieTimeline pendingClusteringTimeline = table.getActiveTimeline().filterPendingReplaceOrClusteringTimeline();
     Option<HoodieInstant> inflightInstantOpt = ClusteringUtils.getInflightClusteringInstant(clusteringInstant, table.getActiveTimeline());
     if (inflightInstantOpt.isPresent()) {
@@ -481,7 +481,7 @@ public HoodieWriteMetadata<O> cluster(String clusteringInstant, boolean shouldCo
   }
 
   public boolean purgePendingClustering(String clusteringInstant) {
-    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf().unwrapAs(Configuration.class));
+    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf());
     Option<HoodieInstant> inflightInstantOpt = ClusteringUtils.getInflightClusteringInstant(clusteringInstant, table.getActiveTimeline());
     if (inflightInstantOpt.isPresent()) {
       table.rollbackInflightClustering(inflightInstantOpt.get(), commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false), true);
@@ -498,7 +498,7 @@ public boolean purgePendingClustering(String clusteringInstant) {
    * @return HoodieWriteMetadata
    */
   public HoodieWriteMetadata<T> managePartitionTTL(String instantTime) {
-    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf().unwrapAs(Configuration.class));
+    HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf());
     return table.managePartitionTTL(context, instantTime);
   }
 
@@ -633,7 +633,7 @@ protected Option<String> scheduleTableServiceInternal(String instantTime, Option
     }
 
     Option<String> option = Option.empty();
-    HoodieTable<?, ?, ?, ?> table = createTable(config, hadoopConf);
+    HoodieTable<?, ?, ?, ?> table = createTable(config, storageConf);
 
     switch (tableServiceType) {
       case ARCHIVE:
@@ -675,7 +675,7 @@ protected Option<String> scheduleTableServiceInternal(String instantTime, Option
     return option;
   }
 
-  protected abstract HoodieTable<?, I, ?, T> createTable(HoodieWriteConfig config, Configuration hadoopConf);
+  protected abstract HoodieTable<?, I, ?, T> createTable(HoodieWriteConfig config, StorageConfiguration<?> storageConf);
 
   /**
    * Executes a clustering plan on a table, serially before or after an insert/upsert action.
@@ -754,7 +754,7 @@ public HoodieCleanMetadata clean(String cleanInstantTime, boolean scheduleInline
     CleanerUtils.rollbackFailedWrites(config.getFailedWritesCleanPolicy(),
         HoodieTimeline.CLEAN_ACTION, () -> rollbackFailedWrites());
 
-    HoodieTable table = createTable(config, hadoopConf);
+    HoodieTable table = createTable(config, storageConf);
     if (config.allowMultipleCleans() || !table.getActiveTimeline().getCleanerTimeline().filterInflightsAndRequested().firstInstant().isPresent()) {
       LOG.info("Cleaner started");
       // proceed only if multiple clean schedules are enabled or if there are no pending cleans.
@@ -892,7 +892,7 @@ protected Map<String, Option<HoodiePendingRollbackInfo>> getPendingRollbackInfos
    * @return {@code true} if rollback happens; {@code false} otherwise.
    */
   protected boolean rollbackFailedIndexingCommits() {
-    HoodieTable table = createTable(config, hadoopConf);
+    HoodieTable table = createTable(config, storageConf);
     List<String> instantsToRollback = getFailedIndexingCommitsToRollbackForMetadataTable(table.getMetaClient());
     Map<String, Option<HoodiePendingRollbackInfo>> pendingRollbacks = getPendingRollbackInfos(table.getMetaClient());
     instantsToRollback.forEach(entry -> pendingRollbacks.putIfAbsent(entry, Option.empty()));
@@ -929,7 +929,7 @@ && isIndexingCommit(dataIndexTimeline, instant.getTimestamp()))
    * @return true if rollback was triggered. false otherwise.
    */
   protected Boolean rollbackFailedWrites() {
-    HoodieTable table = createTable(config, hadoopConf);
+    HoodieTable table = createTable(config, storageConf);
     List<String> instantsToRollback = getInstantsToRollback(table.getMetaClient(), config.getFailedWritesCleanPolicy(), Option.empty());
     Map<String, Option<HoodiePendingRollbackInfo>> pendingRollbacks = getPendingRollbackInfos(table.getMetaClient());
     instantsToRollback.forEach(entry -> pendingRollbacks.putIfAbsent(entry, Option.empty()));
@@ -1051,7 +1051,7 @@ public boolean rollback(final String commitInstantTime, Option<HoodiePendingRoll
     LOG.info("Begin rollback of instant " + commitInstantTime);
     final Timer.Context timerContext = this.metrics.getRollbackCtx();
     try {
-      HoodieTable table = createTable(config, hadoopConf);
+      HoodieTable table = createTable(config, storageConf);
       Option<HoodieInstant> commitInstantOpt = Option.fromJavaOptional(table.getActiveTimeline().getCommitsTimeline().getInstantsAsStream()
           .filter(instant -> HoodieActiveTimeline.EQUALS.test(instant.getTimestamp(), commitInstantTime))
           .findFirst());
@@ -1096,7 +1096,7 @@ public boolean rollback(final String commitInstantTime, Option<HoodiePendingRoll
    */
   public void rollbackFailedBootstrap() {
     LOG.info("Rolling back pending bootstrap if present");
-    HoodieTable table = createTable(config, hadoopConf);
+    HoodieTable table = createTable(config, storageConf);
     HoodieTimeline inflightTimeline = table.getMetaClient().getCommitsTimeline().filterPendingExcludingMajorAndMinorCompaction();
     Option<String> instant = Option.fromJavaOptional(
         inflightTimeline.getReverseOrderedInstants().map(HoodieInstant::getTimestamp).findFirst());

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java

@@ -64,7 +64,6 @@
 import org.apache.hudi.exception.HoodieRestoreException;
 import org.apache.hudi.exception.HoodieRollbackException;
 import org.apache.hudi.exception.HoodieSavepointException;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.internal.schema.InternalSchema;
 import org.apache.hudi.internal.schema.Type;
@@ -697,7 +696,7 @@ public void restoreToSavepoint(String savepointTime) {
         // or before the oldest compaction on MDT.
         // We cannot restore to before the oldest compaction on MDT as we don't have the basefiles before that time.
         HoodieTableMetaClient mdtMetaClient = HoodieTableMetaClient.builder()
-            .setConf(HadoopFSUtils.getStorageConfWithCopy(hadoopConf))
+            .setConf(storageConf.newInstance())
             .setBasePath(getMetadataTableBasePath(config.getBasePath())).build();
         Option<HoodieInstant> oldestMdtCompaction = mdtMetaClient.getCommitTimeline().filterCompletedInstants().firstInstant();
         boolean deleteMDT = false;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java

@@ -31,7 +31,6 @@
 import org.apache.hudi.storage.StorageConfiguration;
 import org.apache.hudi.timeline.service.TimelineService;
 
-import org.apache.hadoop.conf.Configuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -181,15 +180,15 @@ private void startServer(TimelineServiceCreator timelineServiceCreator) throws I
 
     this.serviceConfig = timelineServiceConfBuilder.build();
 
-    server = timelineServiceCreator.create(context, storageConf.unwrapCopyAs(Configuration.class), serviceConfig,
+    server = timelineServiceCreator.create(context, storageConf.newInstance(), serviceConfig,
         HoodieStorageUtils.getStorage(writeConfig.getBasePath(), storageConf.newInstance()), viewManager);
     serverPort = server.startService();
     LOG.info("Started embedded timeline server at " + hostAddr + ":" + serverPort);
   }
 
   @FunctionalInterface
   interface TimelineServiceCreator {
-    TimelineService create(HoodieEngineContext context, Configuration hadoopConf, TimelineService.Config timelineServerConf,
+    TimelineService create(HoodieEngineContext context, StorageConfiguration<?> storageConf, TimelineService.Config timelineServerConf,
                            HoodieStorage storage, FileSystemViewManager globalFileSystemViewManager) throws IOException;
   }
 

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/BaseActionExecutor.java

@@ -29,17 +29,16 @@
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.metadata.HoodieTableMetadataWriter;
+import org.apache.hudi.storage.StorageConfiguration;
 import org.apache.hudi.table.HoodieTable;
 
-import org.apache.hadoop.conf.Configuration;
-
 import java.io.Serializable;
 
 public abstract class BaseActionExecutor<T, I, K, O, R> implements Serializable {
 
   private static final long serialVersionUID = 1L;
   protected final transient HoodieEngineContext context;
-  protected final transient Configuration hadoopConf;
+  protected final transient StorageConfiguration<?> storageConf;
 
   protected final HoodieWriteConfig config;
 
@@ -49,7 +48,7 @@ public abstract class BaseActionExecutor<T, I, K, O, R> implements Serializable
 
   public BaseActionExecutor(HoodieEngineContext context, HoodieWriteConfig config, HoodieTable<T, I, K, O> table, String instantTime) {
     this.context = context;
-    this.hadoopConf = context.getStorageConf().unwrapAs(Configuration.class);
+    this.storageConf = context.getStorageConf();
     this.config = config;
     this.table = table;
     this.instantTime = instantTime;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/bootstrap/BootstrapUtils.java

@@ -25,7 +25,6 @@
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.hadoop.fs.HadoopFSUtils;
 
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocatedFileStatus;
@@ -80,7 +79,7 @@ public static List<Pair<String, List<HoodieFileStatus>>> getAllLeafFoldersWithFi
       result.addAll(context.flatMap(subDirectories, directory -> {
         PathFilter pathFilter = getFilePathFilter(baseFileExtension);
         Path path = new Path(directory);
-        FileSystem fileSystem = path.getFileSystem(new Configuration());
+        FileSystem fileSystem = HadoopFSUtils.getFs(path, HadoopFSUtils.getStorageConf());
         RemoteIterator<LocatedFileStatus> itr = fileSystem.listFiles(path, true);
         List<Pair<HoodieFileStatus, Pair<Integer, String>>> res = new ArrayList<>();
         while (itr.hasNext()) {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/RunIndexActionExecutor.java

@@ -36,7 +36,6 @@
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieIndexException;
 import org.apache.hudi.exception.HoodieMetadataException;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
 import org.apache.hudi.metadata.HoodieMetadataMetrics;
 import org.apache.hudi.metadata.HoodieTableMetadataWriter;
 import org.apache.hudi.metadata.MetadataPartitionType;
@@ -158,7 +157,7 @@ public Option<HoodieIndexCommitMetadata> execute() {
           // reconcile with metadata table timeline
           String metadataBasePath = getMetadataTableBasePath(table.getMetaClient().getBasePath().toString());
           HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder()
-              .setConf(HadoopFSUtils.getStorageConfWithCopy(hadoopConf))
+              .setConf(storageConf.newInstance())
               .setBasePath(metadataBasePath).build();
           Set<String> metadataCompletedTimestamps = getCompletedArchivedAndActiveInstantsAfter(indexUptoInstant, metadataMetaClient).stream()
               .map(HoodieInstant::getTimestamp).collect(Collectors.toSet());

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/DirectWriteMarkers.java

@@ -39,7 +39,6 @@
 import org.apache.hudi.storage.StoragePathInfo;
 import org.apache.hudi.table.HoodieTable;
 
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocatedFileStatus;
@@ -118,7 +117,7 @@ public Set<String> createdAndMergedDataPaths(HoodieEngineContext context, int pa
       context.setJobStatus(this.getClass().getSimpleName(), "Obtaining marker files for all created, merged paths");
       dataFiles.addAll(context.flatMap(subDirectories, directory -> {
         Path path = new Path(directory);
-        FileSystem fileSystem = HadoopFSUtils.getFs(path, storageConf.unwrapAs(Configuration.class));
+        FileSystem fileSystem = HadoopFSUtils.getFs(path, storageConf);
         RemoteIterator<LocatedFileStatus> itr = fileSystem.listFiles(path, true);
         List<String> result = new ArrayList<>();
         while (itr.hasNext()) {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkersFactory.java

@@ -24,7 +24,6 @@
 import org.apache.hudi.storage.StorageSchemes;
 import org.apache.hudi.table.HoodieTable;
 
-import org.apache.hadoop.conf.Configuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -53,7 +52,7 @@ public static WriteMarkers get(MarkerType markerType, HoodieTable table, String
         }
         String basePath = table.getMetaClient().getBasePath().toString();
         if (StorageSchemes.HDFS.getScheme().equals(
-            HadoopFSUtils.getFs(basePath, table.getContext().getStorageConf().unwrapCopyAs(Configuration.class)).getScheme())) {
+            HadoopFSUtils.getFs(basePath, table.getContext().getStorageConf(), true).getScheme())) {
           LOG.warn("Timeline-server-based markers are not supported for HDFS: "
               + "base path " + basePath + ".  Falling back to direct markers.");
           return new DirectWriteMarkers(table, instantTime);