You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2021/11/24 18:04:40 UTC
[GitHub] [hudi] yihua commented on a change in pull request #4090: [HUDI-2840] Fixed DeltaStreaemer to properly respect configuration passed t/h properties file
yihua commented on a change in pull request #4090:
URL: https://github.com/apache/hudi/pull/4090#discussion_r756282739
##########
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:
@nsivabalan This is changed to get the `FileSystem` instance each time for loading a props file (`fs` is passed in from the constructor before). Since it's only for configuration, it shouldn't incur too much overhead?
##########
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:
I believe the defaults are set when building the clustering and write config so it's not necessary to `getBooleanOrDefault()`, since there is always a value corresponding to each config key.
##########
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()
+ .withPath(cfg.targetBasePath)
+ .combineInput(cfg.filterDupes, combineBeforeUpsert)
+ .withCompactionConfig(
+ HoodieCompactionConfig.newBuilder()
+ .withPayloadClass(cfg.payloadClassName)
+ .withInlineCompaction(cfg.isInlineCompactionEnabled())
+ .build()
+ )
+ .withPayloadConfig(
+ HoodiePayloadConfig.newBuilder()
+ .withPayloadOrderingField(cfg.sourceOrderingField)
+ .build())
.forTable(cfg.targetTableName)
- .withAutoCommit(autoCommit).withProps(props);
+ .withAutoCommit(autoCommit)
+ .withProps(props);
- if (null != schema) {
- builder = builder.withSchema(schema.toString());
+ if (schema != null) {
+ builder.withSchema(schema.toString());
}
+
HoodieWriteConfig config = builder.build();
// set default value for {@link HoodieWriteCommitKafkaCallbackConfig} if needed.
if (config.writeCommitCallbackOn() && HoodieWriteCommitKafkaCallback.class.getName().equals(config.getCallbackClass())) {
HoodieWriteCommitKafkaCallbackConfig.setCallbackKafkaConfigIfNeeded(config);
}
+ HoodieClusteringConfig clusteringConfig = HoodieClusteringConfig.from(props);
+
// Validate what deltastreamer assumes of write-config to be really safe
ValidationUtils.checkArgument(config.inlineCompactionEnabled() == cfg.isInlineCompactionEnabled(),
String.format("%s should be set to %s", INLINE_COMPACT.key(), cfg.isInlineCompactionEnabled()));
- ValidationUtils.checkArgument(config.inlineClusteringEnabled() == cfg.isInlineClusteringEnabled(),
- String.format("%s should be set to %s", INLINE_CLUSTERING.key(), cfg.isInlineClusteringEnabled()));
- ValidationUtils.checkArgument(config.isAsyncClusteringEnabled() == cfg.isAsyncClusteringEnabled(),
- String.format("%s should be set to %s", ASYNC_CLUSTERING_ENABLE.key(), cfg.isAsyncClusteringEnabled()));
+ ValidationUtils.checkArgument(config.inlineClusteringEnabled() == clusteringConfig.isInlineClusteringEnabled(),
+ String.format("%s should be set to %s", INLINE_CLUSTERING.key(), clusteringConfig.isInlineClusteringEnabled()));
+ ValidationUtils.checkArgument(config.isAsyncClusteringEnabled() == clusteringConfig.isAsyncClusteringEnabled(),
+ String.format("%s should be set to %s", ASYNC_CLUSTERING_ENABLE.key(), clusteringConfig.isAsyncClusteringEnabled()));
Review comment:
To clarify, is this to make sure the properties loaded from the file matches what's in the write config after config resolution?
##########
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:
Most changes in this PR are related to this so just to make sure there is no hidden implication.
##########
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
Review comment:
nit: typo, `injection` -> `injecting`
--
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