lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-3946] Validate option path in flink hudi sink (#5397)

Browse Source
This commit is contained in:
Ruguo Yu
2022-04-25 10:13:47 +08:00
committed by GitHub
parent bda3db078e
commit d994c58cc0
1 changed files with 5 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
View File

@@ -20,6 +20,7 @@ package org.apache.hudi.table;
import org.apache.hudi.common.model.DefaultHoodieRecordPayload; import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
import org.apache.hudi.common.model.EventTimeAvroPayload; import org.apache.hudi.common.model.EventTimeAvroPayload;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.OptionsResolver; import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.exception.HoodieValidationException; import org.apache.hudi.exception.HoodieValidationException;
@@ -53,6 +54,8 @@ import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.Set; import java.util.Set;
import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
/** /**
* Hoodie data source/sink factory. * Hoodie data source/sink factory.
*/ */
@@ -81,6 +84,8 @@ public class HoodieTableFactory implements DynamicTableSourceFactory, DynamicTab
@Override @Override
public DynamicTableSink createDynamicTableSink(Context context) { public DynamicTableSink createDynamicTableSink(Context context) {
Configuration conf = FlinkOptions.fromMap(context.getCatalogTable().getOptions()); Configuration conf = FlinkOptions.fromMap(context.getCatalogTable().getOptions());
checkArgument(!StringUtils.isNullOrEmpty(conf.getString(FlinkOptions.PATH)),
"Option [path] should not be empty.");
ResolvedSchema schema = context.getCatalogTable().getResolvedSchema(); ResolvedSchema schema = context.getCatalogTable().getResolvedSchema();
sanityCheck(conf, schema); sanityCheck(conf, schema);
setupConfOptions(conf, context.getObjectIdentifier().getObjectName(), context.getCatalogTable(), schema); setupConfOptions(conf, context.getObjectIdentifier().getObjectName(), context.getCatalogTable(), schema);