brachi-wernick commented on a change in pull request #16598: URL: https://github.com/apache/flink/pull/16598#discussion_r700523755
########## File path: docs/content/docs/connectors/table/pubsub.md ########## @@ -0,0 +1,145 @@ +--- +title: PubSub +weight: 20 +type: docs +aliases: + - /dev/table/connectors/pubsub.html +--- +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Google Cloud PubSub SQL Connector + +{{< label "Scan Source: Unbounded" >}} +{{< label "Sink: Streaming Append Mode" >}} Review comment: mistake (copy paste from another connector so forget to clean it 😬 ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubDynamicTableFactory.java ########## @@ -0,0 +1,67 @@ +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.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; + +/** Factory for creating {@link PubsubDynamicSource}. */ +@Internal +public class PubSubDynamicTableFactory implements DynamicTableSourceFactory { + + @Override + public String factoryIdentifier() { + return PubSubConnectorConfigOptions.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + final Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PubSubConnectorConfigOptions.PROJECT_NAME); + options.add(PubSubConnectorConfigOptions.SUBSCRIPTION); + options.add(FactoryUtil.FORMAT); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return new HashSet<>(); + } + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + + // discover a suitable decoding format + final DecodingFormat<DeserializationSchema<RowData>> decodingFormat = + helper.discoverDecodingFormat( + DeserializationFormatFactory.class, FactoryUtil.FORMAT); + + // validate all options + helper.validate(); + + // get the validated options + final ReadableConfig options = helper.getOptions(); + final String project = options.get(PubSubConnectorConfigOptions.PROJECT_NAME); + final String topic = options.get(PubSubConnectorConfigOptions.SUBSCRIPTION); Review comment: oops, sure ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubSubConnectorConfigOptions.java ########## @@ -0,0 +1,27 @@ +package org.apache.flink.streaming.connectors.gcp.pubsub.table; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; + +/** + * Options for PubSub tables supported by the {@code CREATE TABLE ... WITH ...} clause of the Flink + * SQL dialect and the Flink Table API. + */ +@PublicEvolving +public class PubSubConnectorConfigOptions { + + public static final ConfigOption<String> PROJECT_NAME = + ConfigOptions.key("projectName") + .stringType() + .noDefaultValue() + .withDescription("Name of the PubSub project backing this table."); + + public static final ConfigOption<String> SUBSCRIPTION = + ConfigOptions.key("subscription") + .stringType() + .noDefaultValue() + .withDescription("Name of the PubSub subscription backing this table."); + + public static final String IDENTIFIER = "pubsub"; Review comment: So if it will be moved to the factory class, do we still need this option class? can I remove it? not sure i understand its purpose.. (For now I removed it lmk if it is needed) ########## File path: flink-connectors/flink-connector-gcp-pubsub/src/main/java/org/apache/flink/streaming/connectors/gcp/pubsub/table/PubsubDynamicSource.java ########## @@ -0,0 +1,80 @@ +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.streaming.connectors.gcp.pubsub.PubSubSource; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; +import org.apache.flink.table.connector.source.SourceFunctionProvider; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.DataType; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; + +/** PubSub-backed {@link ScanTableSource}. */ +@Internal +public class PubsubDynamicSource implements ScanTableSource { + + private static Logger logger = LoggerFactory.getLogger(PubsubDynamicSource.class); + + /** Name of the PubSub project backing this table. */ + private final String project; + /** Name of the PubSub subscription backing this table. */ + private final String subscription; + /** Scan format for decoding records from PubSub. */ + private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat; + /** Data type that describes the final output of the source. */ + private final DataType producedDataType; + + public PubsubDynamicSource( + String project, + String subscription, + DecodingFormat<DeserializationSchema<RowData>> decodingFormat, + DataType producedDataType) { + + this.project = project; + this.subscription = subscription; + this.decodingFormat = decodingFormat; + this.producedDataType = producedDataType; + } + + @Override + public ChangelogMode getChangelogMode() { + return ChangelogMode.insertOnly(); + } + + @Override + public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) { + // create runtime classes that are shipped to the cluster + + final DeserializationSchema<RowData> deserializer = + decodingFormat.createRuntimeDecoder(runtimeProviderContext, producedDataType); + + try { + PubSubSource<RowData> source = + PubSubSource.newBuilder() + .withDeserializationSchema(deserializer) + .withProjectName(project) + .withSubscriptionName(subscription) Review comment: I think, most of the time, when you read from PubSub your cluster exists in GCP and you probably configure the state with GCS, so you already define the credentials. Same for Kinesis, your cluster probably deployed in AWS with s3 state and you already define the credentials.(I can't see in Kinesis that they get credentials... maybe it should be addressed also there) This can be relevant when you run in one cloud, let's say AWS ,and want to read data from GCP PubSub and vice versa.. or you run on premises cluster that reads data from kinesis/PubSub.. for those cases yes. we need to find a way to read credentials in some secured way, because tables can be done by external teams (self service) -- 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]
