[GitHub] [hudi] vinothchandar commented on a change in pull request #4090: [HUDI-2840] Fixed DeltaStreaemer to properly respect configuration passed t/h properties file

[GitBox <gi...@apache.org>]

vinothchandar commented on a change in pull request #4090:
URL: https://github.com/apache/hudi/pull/4090#discussion_r756360237



##########
File path: hudi-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
##########
@@ -119,13 +123,13 @@ public void addPropsFromFile(Path filePath) {
     if (visitedFilePaths.contains(filePath.toString())) {
       throw new IllegalStateException("Loop detected; file " + filePath + " already referenced");
     }
-    FileSystem fileSystem;
-    try {
-      fileSystem = fs != null ? fs : filePath.getFileSystem(new Configuration());
-    } catch (IOException e) {
-      throw new IllegalArgumentException("Cannot get the file system from file path", e);
-    }
-    try (BufferedReader reader = new BufferedReader(new InputStreamReader(fileSystem.open(filePath)))) {
+
+    FileSystem fs = FSUtils.getFs(
+        filePath.toString(),
+        Option.ofNullable(hadoopConfig).orElseGet(Configuration::new)
+    );

Review comment:
       FileSystem instance is also cached at the hadoop client level. so this itself may be ok. 

##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
##########
@@ -1150,11 +1150,11 @@ public int getArchiveDeleteParallelism() {
   }
 
   public boolean inlineClusteringEnabled() {
-    return getBoolean(HoodieClusteringConfig.INLINE_CLUSTERING);
+    return getBooleanOrDefault(HoodieClusteringConfig.INLINE_CLUSTERING);
   }
 
   public boolean isAsyncClusteringEnabled() {
-    return getBoolean(HoodieClusteringConfig.ASYNC_CLUSTERING_ENABLE);
+    return getBooleanOrDefault(HoodieClusteringConfig.ASYNC_CLUSTERING_ENABLE);

Review comment:
       +1 

##########
File path: hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
##########
@@ -635,7 +637,7 @@ public DeltaSync getDeltaSync() {
                     HoodieTimeline.COMPACTION_ACTION, scheduledCompactionInstantAndRDD.get().getLeft().get()));
                 asyncCompactService.get().waitTillPendingAsyncServiceInstantsReducesTo(cfg.maxPendingCompactions);
               }
-              if (cfg.isAsyncClusteringEnabled()) {
+              if (HoodieClusteringConfig.from(props).isAsyncClusteringEnabled()) {

Review comment:
       Can we retain the resolved clustering config above as a member?

##########
File path: hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
##########
@@ -680,36 +682,49 @@ private HoodieWriteConfig getHoodieClientConfig(SchemaProvider schemaProvider) {
   private HoodieWriteConfig getHoodieClientConfig(Schema schema) {
     final boolean combineBeforeUpsert = true;
     final boolean autoCommit = false;
+
+    // NOTE: Provided that we're injection combined properties
+    //       (from {@code props}, including CLI overrides), there's no
+    //       need to explicitly set up some configuration aspects that
+    //       are based on these (for ex Clustering configuration)
     HoodieWriteConfig.Builder builder =
-        HoodieWriteConfig.newBuilder().withPath(cfg.targetBasePath).combineInput(cfg.filterDupes, combineBeforeUpsert)
-            .withCompactionConfig(HoodieCompactionConfig.newBuilder().withPayloadClass(cfg.payloadClassName)
-                // Inline compaction is disabled for continuous mode. otherwise enabled for MOR
-                .withInlineCompaction(cfg.isInlineCompactionEnabled()).build())
-            .withClusteringConfig(HoodieClusteringConfig.newBuilder()
-                .withInlineClustering(cfg.isInlineClusteringEnabled())
-                .withAsyncClustering(cfg.isAsyncClusteringEnabled()).build())
-            .withPayloadConfig(HoodiePayloadConfig.newBuilder().withPayloadOrderingField(cfg.sourceOrderingField)
-                .build())
+        HoodieWriteConfig.newBuilder()

Review comment:
       please avoid reformatting code in the same PR, unless you are changing lines. Prefer to deal with code formatting as a separate topic. 

##########
File path: hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
##########
@@ -635,7 +637,7 @@ public DeltaSync getDeltaSync() {
                     HoodieTimeline.COMPACTION_ACTION, scheduledCompactionInstantAndRDD.get().getLeft().get()));
                 asyncCompactService.get().waitTillPendingAsyncServiceInstantsReducesTo(cfg.maxPendingCompactions);
               }
-              if (cfg.isAsyncClusteringEnabled()) {
+              if (HoodieClusteringConfig.from(props).isAsyncClusteringEnabled()) {

Review comment:
       this does a full load each time?

##########
File path: hudi-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
##########
@@ -58,7 +61,8 @@
   // props read from hudi-defaults.conf
   private static TypedProperties GLOBAL_PROPS = loadGlobalProps();
 
-  private final FileSystem fs;
+  @Nullable
+  private final Configuration hadoopConfig;

Review comment:
       In general we can keep `SerializableConfiguration` that way even if it serializes from driver to executor, it may be ok. But I think we don't do this, otherwise we ll face errros with `fs` to begin with




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org