[hotfix][examples-table] Add missing format validation

deepfield · Jul 13, 2020 · e4f431c · e4f431c
1 parent a028ba2
commit e4f431c
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 0 deletions.
diff --git a/docs/dev/table/sourceSinks.md b/docs/dev/table/sourceSinks.md
@@ -398,6 +398,7 @@ import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.connector.format.DecodingFormat;
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.factories.DeserializationFormatFactory;
 import org.apache.flink.table.factories.DynamicTableFactory;
 
@@ -429,6 +430,10 @@ public class ChangelogCsvFormatFactory implements DeserializationFormatFactory {
  public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
  DynamicTableFactory.Context context,
  ReadableConfig formatOptions) {
+ // either implement your custom validation logic here ...
+ // or use the provided helper method
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
  // get the validated options
  final String columnDelimiter = formatOptions.get(COLUMN_DELIMITER);
 

diff --git a/docs/dev/table/sourceSinks.zh.md b/docs/dev/table/sourceSinks.zh.md
@@ -398,6 +398,7 @@ import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.connector.format.DecodingFormat;
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.factories.DeserializationFormatFactory;
 import org.apache.flink.table.factories.DynamicTableFactory;
 
@@ -429,6 +430,10 @@ public class ChangelogCsvFormatFactory implements DeserializationFormatFactory {
  public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
  DynamicTableFactory.Context context,
  ReadableConfig formatOptions) {
+ // either implement your custom validation logic here ...
+ // or use the provided helper method
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
  // get the validated options
  final String columnDelimiter = formatOptions.get(COLUMN_DELIMITER);
 

diff --git a/.../main/java/org/apache/flink/table/examples/java/connectors/ChangelogCsvFormatFactory.java b/.../main/java/org/apache/flink/table/examples/java/connectors/ChangelogCsvFormatFactory.java
@@ -67,6 +67,10 @@ public Set<ConfigOption<?>> optionalOptions() {
 
  @Override
  public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+ // either implement your custom validation logic here ...
+ // or use the provided helper method
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
  // get the validated options
  final String columnDelimiter = formatOptions.get(COLUMN_DELIMITER);