[GitHub] [flink] brachi-wernick commented on a change in pull request #16598: [FLINK-23501][GCP PubSub] add table source API for GCP PubSub

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

brachi-wernick commented on a change in pull request #16598:
URL: https://github.com/apache/flink/pull/16598#discussion_r677542021



##########
File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubTableSourceFactory.java
##########
@@ -0,0 +1,78 @@
+package org.apache.flink.streaming.connectors.gcp.pubsub.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+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.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Internal
+public class PubSubTableSourceFactory implements DynamicTableSourceFactory {
+
+    // define all options statically
+    public static final ConfigOption<String> PROJECT_NAME =
+            ConfigOptions.key("projectName").stringType().noDefaultValue();
+
+    public static final ConfigOption<String> TOPIC =
+            ConfigOptions.key("topic").stringType().noDefaultValue();
+
+    public static final ConfigOption<String> FORMAT =
+            ConfigOptions.key("format").stringType().noDefaultValue();
+
+    @Override
+    public String factoryIdentifier() {
+        return "pubsub";
+    }
+
+    @Override
+    public Set<ConfigOption<?>> requiredOptions() {
+        final Set<ConfigOption<?>> options = new HashSet<>();
+        options.add(PROJECT_NAME);
+        options.add(TOPIC);
+        options.add(FORMAT);
+        return options;
+    }
+
+    @Override
+    public Set<ConfigOption<?>> optionalOptions() {
+        return new HashSet<>();
+    }
+
+    @Override
+    public DynamicTableSource createDynamicTableSource(Context context) {
+        // either implement your custom validation logic here ...
+        // or use the provided helper utility
+        final FactoryUtil.TableFactoryHelper helper =
+                FactoryUtil.createTableFactoryHelper(this, context);
+
+        // validate all options
+        helper.validateExcept("json.");

Review comment:
       When there is json config in the create table sql, the validation here failed.
   These are the json config I see:
   ```
    -- 'json.map-null-key.mode' = 'FAIL', -- Optional flag to control the handling mode when serializing null key for map data, FAIL by default. Option DROP will drop null key entries for map data. Option LITERAL will use 'map-null-key.literal' as key literal.
     -- 'json.encode.decimal-as-plain-number' = 'false', -- Optional flag to specify whether to encode all decimals as plain numbers instead of possible scientific notations, false by default.
     -- 'json.map-null-key.literal' = 'null', -- Optional flag to specify string literal for null keys when 'map-null-key.mode' is LITERAL, "null" by default.
     -- 'json.fail-on-missing-field' = 'false', -- Optional flag to specify whether to fail if a field is missing or not, false by default.
     -- 'json.ignore-parse-errors' = 'false', -- Optional flag to skip fields and rows with parse errors instead of failing; fields are set to null in case of errors, false by default.
     -- 'json.timestamp-format.standard' = 'SQL', -- Optional flag to specify timestamp format, SQL by default. Option ISO-8601 will parse input timestamp in "yyyy-MM-ddTHH:mm:ss.s{precision}" format and output timestamp in the same format. Option SQL will parse input timestamp in "yyyy-MM-dd HH:mm:ss.s{precision}" format and output timestamp in the same format.
     ```
   Any other idea how to address it?
    




-- 
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: issues-unsubscribe@flink.apache.org

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