alexeykudinkin commented on a change in pull request #4090:
URL: https://github.com/apache/hudi/pull/4090#discussion_r756395759
##########
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:
Correct
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
