dannycranmer commented on code in PR #84: URL: https://github.com/apache/flink-connector-aws/pull/84#discussion_r1247594258
########## flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/model/Metadata.java: ########## @@ -0,0 +1,61 @@ +/* + * 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. + */ + +package org.apache.flink.connector.kinesis.source.model; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.types.DataType; + +import java.util.Arrays; +import java.util.Objects; + +/** Internal type for enumerating available metadata. */ +public enum Metadata { Review Comment: Missing compatibility annotation (`@Internal`) ########## flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/config/KinesisStreamsSourceConfigUtil.java: ########## @@ -58,4 +64,51 @@ public static Date parseStreamTimestampStartingPosition(final Configuration sour return new Date((long) (Double.parseDouble(timestamp) * 1000)); } } + + /** + * Validate configuration properties for {@link + * org.apache.flink.connector.kinesis.source.KinesisStreamsSource}. + */ + public static void validateStreamSourceConfiguration(Configuration config) { + checkNotNull(config, "config can not be null"); + + Properties consumerProperties = new Properties(); + config.addAllToProperties(consumerProperties); + validateAwsConfiguration(consumerProperties); + + if (!(config.containsKey(AWSConfigConstants.AWS_REGION) + || config.containsKey(AWSConfigConstants.AWS_ENDPOINT))) { + // per validation in AwsClientBuilder + throw new IllegalArgumentException( + String.format( + "For KinesisStreamsSource AWS region ('%s') and/or AWS endpoint ('%s') must be set in the config.", + AWSConfigConstants.AWS_REGION, AWSConfigConstants.AWS_ENDPOINT)); + } Review Comment: Seems like this should be duplicated, can we consolidate into a common place for all table APIs? ########## flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/model/Metadata.java: ########## @@ -0,0 +1,61 @@ +/* + * 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. + */ + +package org.apache.flink.connector.kinesis.source.model; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.types.DataType; + +import java.util.Arrays; +import java.util.Objects; + +/** Internal type for enumerating available metadata. */ +public enum Metadata { Review Comment: This class is named very generically, and the package name does not tell me it is related to Table API. Can we rename/move package? ########## flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/table/KinesisDynamicTableSourceFactory.java: ########## @@ -0,0 +1,91 @@ +/* + * 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. + */ + +package org.apache.flink.connector.kinesis.table; + +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.connector.kinesis.table.util.KinesisStreamsConnectorSourceOptionsUtil; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +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 java.util.HashSet; +import java.util.Set; + +import static org.apache.flink.connector.kinesis.table.KinesisConnectorOptions.STREAM_ARN; +import static org.apache.flink.table.factories.FactoryUtil.FORMAT; + +/** Factory for creating {@link KinesisDynamicSource}. */ +public class KinesisDynamicTableSourceFactory implements DynamicTableSourceFactory { + public static final String IDENTIFIER = "kinesis-source"; Review Comment: This is not a decision we can make so lightly. For the KDS sink we superseded the previous sink with the new one. We reused `kinesis` identifier and maintained backwards compatibility for config options, where possible. Now we have externalised connectors we have 2 options, but regardless, I do not want to change the identifier: 1. Minor version (4.x) with backwards compatibility for old source config 2. Major version (5.x) with breaking changes, no support for old config ########## flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/table/KinesisDynamicTableSourceFactory.java: ########## @@ -0,0 +1,91 @@ +/* + * 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. + */ + +package org.apache.flink.connector.kinesis.table; + +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.connector.kinesis.table.util.KinesisStreamsConnectorSourceOptionsUtil; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +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 java.util.HashSet; +import java.util.Set; + +import static org.apache.flink.connector.kinesis.table.KinesisConnectorOptions.STREAM_ARN; +import static org.apache.flink.table.factories.FactoryUtil.FORMAT; + +/** Factory for creating {@link KinesisDynamicSource}. */ +public class KinesisDynamicTableSourceFactory implements DynamicTableSourceFactory { + public static final String IDENTIFIER = "kinesis-source"; + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + + ReadableConfig tableOptions = helper.getOptions(); + + DecodingFormat<DeserializationSchema<RowData>> decodingFormat = + helper.discoverDecodingFormat(DeserializationFormatFactory.class, FORMAT); + + ResolvedCatalogTable catalogTable = context.getCatalogTable(); + + KinesisStreamsConnectorSourceOptionsUtil kinesisStreamsConnectorSourceOptionsUtil = + new KinesisStreamsConnectorSourceOptionsUtil( + catalogTable.getOptions(), tableOptions.get(STREAM_ARN)); + + Configuration sourceConfig = + kinesisStreamsConnectorSourceOptionsUtil.getValidatedSourceConfigurations(); + + KinesisDynamicSource.KinesisDynamicTableSourceBuilder builder = + new KinesisDynamicSource.KinesisDynamicTableSourceBuilder(); + + builder.setStream(tableOptions.get(STREAM_ARN)) + .setDecodingFormat(decodingFormat) + .setConsumedDataType(context.getPhysicalRowDataType()) + .setSourceConfig(sourceConfig); + + KinesisDynamicSource kinesisDynamicSource = builder.build(); + return kinesisDynamicSource; + } + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + final Set<ConfigOption<?>> options = new HashSet<>(); + options.add(STREAM_ARN); Review Comment: This would be a breaking change for a minor version update ########## flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/table/KinesisDynamicSource.java: ########## @@ -0,0 +1,244 @@ +/* + * 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. + */ + +package org.apache.flink.connector.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.connector.kinesis.source.KinesisStreamsSource; +import org.apache.flink.connector.kinesis.source.model.Metadata; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.datastream.DataStreamSource; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.ProviderContext; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.source.DataStreamScanProvider; +import org.apache.flink.table.connector.source.DynamicTableSource; +import org.apache.flink.table.connector.source.ScanTableSource; +import org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.DataType; +import org.apache.flink.util.Preconditions; + +import javax.annotation.Nullable; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; + +/** Kinesis-backed {@link ScanTableSource}. */ +public class KinesisDynamicSource implements ScanTableSource, SupportsReadingMetadata { Review Comment: Missing compatibility annotation ########## flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/config/KinesisStreamsSourceConfigUtil.java: ########## @@ -58,4 +64,51 @@ public static Date parseStreamTimestampStartingPosition(final Configuration sour return new Date((long) (Double.parseDouble(timestamp) * 1000)); } } + + /** + * Validate configuration properties for {@link + * org.apache.flink.connector.kinesis.source.KinesisStreamsSource}. + */ + public static void validateStreamSourceConfiguration(Configuration config) { + checkNotNull(config, "config can not be null"); + + Properties consumerProperties = new Properties(); + config.addAllToProperties(consumerProperties); + validateAwsConfiguration(consumerProperties); + + if (!(config.containsKey(AWSConfigConstants.AWS_REGION) + || config.containsKey(AWSConfigConstants.AWS_ENDPOINT))) { + // per validation in AwsClientBuilder + throw new IllegalArgumentException( + String.format( + "For KinesisStreamsSource AWS region ('%s') and/or AWS endpoint ('%s') must be set in the config.", + AWSConfigConstants.AWS_REGION, AWSConfigConstants.AWS_ENDPOINT)); + } + + if (config.contains(KinesisStreamsSourceConfigConstants.STREAM_INITIAL_POSITION)) { + KinesisStreamsSourceConfigConstants.InitialPosition initPosType = + config.get(KinesisStreamsSourceConfigConstants.STREAM_INITIAL_POSITION); + + // specified initial timestamp in stream when using AT_TIMESTAMP + if (initPosType == KinesisStreamsSourceConfigConstants.InitialPosition.AT_TIMESTAMP) { + if (!config.contains(STREAM_INITIAL_TIMESTAMP)) { + throw new IllegalArgumentException( + "Please set value for initial timestamp ('" + + STREAM_INITIAL_TIMESTAMP + + "') when using AT_TIMESTAMP initial position."); + } + validateOptionalDateProperty( + consumerProperties, + String.valueOf(STREAM_INITIAL_TIMESTAMP), + config.getString(STREAM_TIMESTAMP_DATE_FORMAT), + "Invalid value given for initial timestamp for AT_TIMESTAMP initial position in stream. " + + "Must be a valid format: yyyy-MM-dd'T'HH:mm:ss.SSSXXX or non-negative double value. For example, 2016-04-04T19:58:46.480-00:00 or 1459799926.480 ."); Review Comment: This error message is only valid when `STREAM_TIMESTAMP_DATE_FORMAT` is not given. Can we use `STREAM_TIMESTAMP_DATE_FORMAT` in the error message? -- 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]
