twalthr commented on a change in pull request #13770: URL: https://github.com/apache/flink/pull/13770#discussion_r516768180
########## File path: docs/dev/table/connectors/kinesis.md ########## @@ -0,0 +1,334 @@ +--- +title: "Amazon Kinesis Data Streams SQL Connector" +nav-title: Kinesis +nav-parent_id: sql-connectors +nav-pos: 2 +--- +<!-- +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. +--> + +<span class="label label-primary">Scan Source: Unbounded</span> +<span class="label label-primary">Sink: Streaming Append Mode</span> + +* This will be replaced by the TOC +{:toc} + +The Kinesis connector allows for reading data from and writing data into [Amazon Kinesis Data Streams (KDS)](https://aws.amazon.com/kinesis/data-streams/). + +Dependencies +------------ + +To use the connector, add the following Maven dependency to your project: + +{% highlight xml %} +<dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-kinesis{{ site.scala_version_suffix }}</artifactId> + <version>{{site.version }}</version> +</dependency> +{% endhighlight %} + +How to create a Kinesis data stream table +----------------------------------------- + +Follow the instructions from the [Amazon KDS Developer Guide](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) to set up a Kinesis stream. +The following example shows how to create a table backed by a Kinesis data stream: + +<div class="codetabs" markdown="1"> +<div data-lang="SQL" markdown="1"> +{% highlight sql %} +CREATE TABLE KinesisTable ( + user_id BIGINT, + item_id BIGINT, + category_id BIGINT, + behavior STRING, + ts TIMESTAMP(3) +) PARTITIONED BY (user_id, item_id) WITH ( + 'connector' = 'kinesis', + 'stream' = 'user_behavior', + 'properties.aws.region' = 'us-east-2', + 'properties.flink.stream.initpos' = 'LATEST', + 'format' = 'csv' +) +{% endhighlight %} +</div> +</div> + +Available Metadata +------------------ + +The following metadata can be exposed as read-only (`VIRTUAL`) columns in a table definition. + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Column Name</th> + <th class="text-center" style="width: 45%">Column Type</th> + <th class="text-center" style="width: 35%">Description</th> + </tr> + </thead> + <tbody> + <tr> + <td><code><a href="https://docs.aws.amazon.com/kinesis/latest/APIReference/API_Record.html#Streams-Type-Record-ApproximateArrivalTimestamp";>ApproximateArrivalTimestamp</a></code></td> + <td><code>TIMESTAMP(3) WITH LOCAL TIMEZONE NOT NULL</code></td> + <td>The approximate time that the record was inserted into the stream.</td> Review comment: `The approximate time when the record was inserted into the stream.` ########## File path: docs/dev/table/connectors/kinesis.md ########## @@ -0,0 +1,334 @@ +--- +title: "Amazon Kinesis Data Streams SQL Connector" +nav-title: Kinesis +nav-parent_id: sql-connectors +nav-pos: 2 +--- +<!-- +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. +--> + +<span class="label label-primary">Scan Source: Unbounded</span> +<span class="label label-primary">Sink: Streaming Append Mode</span> + +* This will be replaced by the TOC +{:toc} + +The Kinesis connector allows for reading data from and writing data into [Amazon Kinesis Data Streams (KDS)](https://aws.amazon.com/kinesis/data-streams/). + +Dependencies +------------ + +To use the connector, add the following Maven dependency to your project: + +{% highlight xml %} +<dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-kinesis{{ site.scala_version_suffix }}</artifactId> + <version>{{site.version }}</version> +</dependency> +{% endhighlight %} + +How to create a Kinesis data stream table +----------------------------------------- + +Follow the instructions from the [Amazon KDS Developer Guide](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) to set up a Kinesis stream. +The following example shows how to create a table backed by a Kinesis data stream: + +<div class="codetabs" markdown="1"> +<div data-lang="SQL" markdown="1"> +{% highlight sql %} +CREATE TABLE KinesisTable ( + user_id BIGINT, + item_id BIGINT, + category_id BIGINT, + behavior STRING, + ts TIMESTAMP(3) +) PARTITIONED BY (user_id, item_id) WITH ( + 'connector' = 'kinesis', + 'stream' = 'user_behavior', + 'properties.aws.region' = 'us-east-2', + 'properties.flink.stream.initpos' = 'LATEST', + 'format' = 'csv' +) +{% endhighlight %} +</div> +</div> + +Available Metadata +------------------ + +The following metadata can be exposed as read-only (`VIRTUAL`) columns in a table definition. + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Column Name</th> + <th class="text-center" style="width: 45%">Column Type</th> + <th class="text-center" style="width: 35%">Description</th> + </tr> + </thead> + <tbody> + <tr> + <td><code><a href="https://docs.aws.amazon.com/kinesis/latest/APIReference/API_Record.html#Streams-Type-Record-ApproximateArrivalTimestamp";>ApproximateArrivalTimestamp</a></code></td> + <td><code>TIMESTAMP(3) WITH LOCAL TIMEZONE NOT NULL</code></td> + <td>The approximate time that the record was inserted into the stream.</td> + </tr> + <tr> + <td><code><a href="https://docs.aws.amazon.com/kinesis/latest/APIReference/API_Shard.html#Streams-Type-Shard-ShardId";>ShardId</a></code></td> + <td><code>VARCHAR(128) NOT NULL</code></td> + <td>The unique identifier of the shard within the stream from which the record was read.</td> + </tr> + <tr> + <td><code><a href="https://docs.aws.amazon.com/kinesis/latest/APIReference/API_Record.html#Streams-Type-Record-SequenceNumber";>SequenceNumber</a></code></td> + <td><code>VARCHAR(128) NOT NULL</code></td> + <td>The unique identifier of the record within its shard.</td> + </tr> + </tbody> +</table> + +The extended `CREATE TABLE` example demonstrates the syntax for exposing these metadata columns: + +<div class="codetabs" markdown="1"> +<div data-lang="SQL" markdown="1"> +{% highlight sql %} +CREATE TABLE KinesisTable ( + user_id BIGINT, + item_id BIGINT, + category_id BIGINT, + behavior STRING, + ts TIMESTAMP(3), + arrival_time TIMESTAMP(3) METADATA FROM 'ApproximateArrivalTimestamp' VIRTUAL, Review comment: We would like to give users a consistent property and metadata naming. Please follow the guidelines in `org.apache.flink.table.factories.Factory` as mentioned in `SupportsReadingMetadata#listReadableMetadata`. So `shard-id` instead of `ShardId` and `approximate-arrival-timestamp` instead of `ApproximateArrivalTimestamp`. Btw should we shorten it to `arrival-timestamp` or just `timestamp` like in Kafka? We try to keep the keys as concise as possible lately. ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisDynamicTableFactory.java ########## @@ -0,0 +1,167 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.api.common.serialization.SerializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.streaming.connectors.kinesis.util.KinesisConfigUtil; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.format.EncodingFormat; +import org.apache.flink.table.connector.sink.DynamicTableSink; +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.DynamicTableSinkFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.factories.SerializationFormatFactory; + +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Properties; +import java.util.Set; + +/** + * Factory for creating {@link KinesisDynamicSource} and {@link KinesisDynamicSink} instances. + */ +@Internal +public class KinesisDynamicTableFactory implements + DynamicTableSourceFactory, + DynamicTableSinkFactory { + public static final String IDENTIFIER = "kinesis"; + + // -------------------------------------------------------------------------------------------- + // DynamicTableSourceFactory + // -------------------------------------------------------------------------------------------- + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + FactoryUtil.TableFactoryHelper helper = FactoryUtil + .createTableFactoryHelper(this, context); + Properties connectorProperties = KinesisOptions Review comment: nit: this is in the middle of the `helper` calls, I would put it further down to where it is needed ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisOptions.java ########## @@ -0,0 +1,186 @@ + /* + * 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.streaming.connectors.kinesis.table; + + import org.apache.flink.annotation.Internal; + import org.apache.flink.configuration.ConfigOption; + import org.apache.flink.configuration.ConfigOptions; + import org.apache.flink.configuration.ReadableConfig; + import org.apache.flink.streaming.connectors.kinesis.FixedKinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.RandomKinesisPartitioner; + import org.apache.flink.table.api.ValidationException; + import org.apache.flink.table.catalog.CatalogTable; + import org.apache.flink.table.data.RowData; + import org.apache.flink.util.FlinkException; + import org.apache.flink.util.InstantiationUtil; + + import java.util.Map; + import java.util.Properties; + + /** + * Options for Kinesis tables supported by the {@code CREATE TABLE ... WITH ...} clause of the + * Flink SQL dialect and the Flink Table API. + */ + @Internal + public class KinesisOptions { + + private KinesisOptions() { + } + + // -------------------------------------------------------------------------------------------- + // Kinesis specific options + // -------------------------------------------------------------------------------------------- + + /** + * Prefix for properties delegated to the + * {@link org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer} and + * {@link org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer}. + */ + public static final String PROPERTIES_PREFIX = "properties."; + + public static final ConfigOption<String> STREAM = ConfigOptions + .key("stream") + .stringType() + .noDefaultValue() + .withDescription("Name of the Kinesis stream backing this table (required)"); + + // -------------------------------------------------------------------------------------------- + // Sink specific options + // -------------------------------------------------------------------------------------------- + + public static final ConfigOption<String> SINK_PARTITIONER = ConfigOptions + .key("sink.partitioner") + .stringType() + .noDefaultValue() + .withDescription( + "Optional output partitioning from Flink's partitions into Kinesis shards. " + + "Sinks that write to tables defined with the PARTITION BY clause " + + "always use a field-based partitioner and cannot define this option. " + + "Valid enumerations are: \n" + + "\"random\":" + + " (use a random partition key),\n" + + "\"fixed\":" + + " (each Flink partition ends up in at most one Kinesis shard),\n" + + "\"custom class name\":" + + " (use a custom KinesisPartitioner subclass)"); Review comment: use a fully qualified name such that user can find this class ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisOptions.java ########## @@ -0,0 +1,186 @@ + /* + * 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.streaming.connectors.kinesis.table; + + import org.apache.flink.annotation.Internal; + import org.apache.flink.configuration.ConfigOption; + import org.apache.flink.configuration.ConfigOptions; + import org.apache.flink.configuration.ReadableConfig; + import org.apache.flink.streaming.connectors.kinesis.FixedKinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.RandomKinesisPartitioner; + import org.apache.flink.table.api.ValidationException; + import org.apache.flink.table.catalog.CatalogTable; + import org.apache.flink.table.data.RowData; + import org.apache.flink.util.FlinkException; + import org.apache.flink.util.InstantiationUtil; + + import java.util.Map; + import java.util.Properties; + + /** + * Options for Kinesis tables supported by the {@code CREATE TABLE ... WITH ...} clause of the + * Flink SQL dialect and the Flink Table API. + */ + @Internal + public class KinesisOptions { + + private KinesisOptions() { + } + + // -------------------------------------------------------------------------------------------- + // Kinesis specific options + // -------------------------------------------------------------------------------------------- + + /** + * Prefix for properties delegated to the + * {@link org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer} and + * {@link org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer}. + */ + public static final String PROPERTIES_PREFIX = "properties."; + + public static final ConfigOption<String> STREAM = ConfigOptions + .key("stream") + .stringType() + .noDefaultValue() + .withDescription("Name of the Kinesis stream backing this table (required)"); + + // -------------------------------------------------------------------------------------------- + // Sink specific options + // -------------------------------------------------------------------------------------------- + + public static final ConfigOption<String> SINK_PARTITIONER = ConfigOptions + .key("sink.partitioner") + .stringType() + .noDefaultValue() + .withDescription( + "Optional output partitioning from Flink's partitions into Kinesis shards. " + + "Sinks that write to tables defined with the PARTITION BY clause " + + "always use a field-based partitioner and cannot define this option. " + + "Valid enumerations are: \n" + + "\"random\":" + + " (use a random partition key),\n" + + "\"fixed\":" + + " (each Flink partition ends up in at most one Kinesis shard),\n" + + "\"custom class name\":" + + " (use a custom KinesisPartitioner subclass)"); + + public static final ConfigOption<String> SINK_PARTITIONER_FIELD_DELIMITER = ConfigOptions + .key("sink.partitioner.field.delimiter") Review comment: call this `sink.partitioner-field-delimiter`? the reason for this is that we try to keep the options hierarchical (imagine a JSON representation). In this case `sink.partitioner` and `sink.partitioner.field` could not be represented. ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitioner.java ########## @@ -0,0 +1,236 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.Preconditions; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * <p>A {@link KinesisPartitioner} of {@link RowData} elements that constructs the partition key + * from a list of field names.</p> + * + * <p>The key is constructed by concatenating the string representations of a list of fields + * projected from an input element. A fixed prefix can be optionally configured in order to speed + * up the key construction process.</p> + * + * <p>Resulting partition key values are trimmed to the maximum length allowed by Kinesis.</p> + */ +@Internal +public class RowDataFieldsKinesisPartitioner extends KinesisPartitioner<RowData> { + + /** + * Allowed maximum length limit of a partition key. + * + * @link https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#API_PutRecord_RequestSyntax + */ + public static final int MAX_PARTITION_KEY_LENGTH = 256; + + /** + * Default delimiter for {@link RowDataFieldsKinesisPartitioner#delimiter}. + */ + public static final String DEFAULT_DELIMITER = String.valueOf('|'); + + /** + * The character used to delimit field values in the concatenated partition key string. + */ + private final String delimiter; + + /** + * A list of field names used to extract the partition key for a record that will be written to + * a Kinesis stream. + */ + private final List<String> fieldNames; + + /** + * A list of getter functions to dynamically extract the field values for all + * {@link RowDataFieldsKinesisPartitioner#fieldNames} from an input record. + */ + private final RowData.FieldGetter[] dynamicFieldGetters; + + /** + * A buffer used to accumulate the concatenation of all field values that form the partition + * key. + */ + private final StringBuilder keyBuffer = new StringBuilder(); + + /** + * A prefix of fixed values to be used instead of the corresponding + * {@link RowDataFieldsKinesisPartitioner#dynamicFieldGetters} entries. + */ + private String[] fixedPrefix = new String[]{}; + + /** + * The cumulative length of all values currently stored in the + * {@link RowDataFieldsKinesisPartitioner#fixedPrefix}. + */ + private int fixedPrefixLength = 0; + + public RowDataFieldsKinesisPartitioner(CatalogTable table) { + this(table, DEFAULT_DELIMITER); + } + + public RowDataFieldsKinesisPartitioner(CatalogTable table, String delimiter) { + Preconditions.checkNotNull(table, "table"); + Preconditions.checkNotNull(delimiter, "delimiter"); + Preconditions.checkArgument( + table.isPartitioned(), + "Cannot create a RowDataFieldsKinesisPartitioner for a non-partitioned table"); + Preconditions.checkArgument( + table.getPartitionKeys().size() == new HashSet<>(table.getPartitionKeys()).size(), + "The sequence of partition keys cannot contain duplicates"); + + TableSchema schema = table.getSchema(); + List<String> schemaFieldsList = Arrays.asList(schema.getFieldNames()); + + String[] badKeys = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + boolean hasField = schema.getTableColumn(fieldName).isPresent(); + return hasField ? Stream.empty() : Stream.of(fieldName); + }) + .toArray(String[]::new); + Preconditions.checkArgument( + badKeys.length == 0, + "The following partition keys are not present in the table: %s", + String.join(", ", badKeys)); + + this.delimiter = delimiter; + this.fieldNames = table.getPartitionKeys(); + this.dynamicFieldGetters = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + Optional<TableColumn> tableColumn = schema.getTableColumn(fieldName); + // convert Optional to Stream to unpack the value in the enclosing flatMap + return tableColumn.map(Stream::of).orElseGet(Stream::empty); + }) + .map(column -> { + LogicalType fieldType = column.getType().getLogicalType(); + int fieldPos = schemaFieldsList.indexOf(column.getName()); + return RowData.createFieldGetter(fieldType, fieldPos); + }) + .toArray(RowData.FieldGetter[]::new); + } + + @Override + public String getPartitionId(RowData element) { + keyBuffer.setLength(fixedPrefixLength); // reset the dynamic part of the key buffer + for (int i = fixedPrefix.length; i < dynamicFieldGetters.length; i++) { + RowData.FieldGetter getter = dynamicFieldGetters[i]; + Object fieldValue = getter.getFieldOrNull(element); + + keyBuffer.append(fieldValue); Review comment: are we limiting the supported data types for partitions somewhere? we cannot assume that every data type (esp in internal data structures) has a good toString method. If I check `SupportsPartitioning`, it seems we always limited this STRING type? ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisDynamicSource.java ########## @@ -0,0 +1,223 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer; +import org.apache.flink.streaming.connectors.kinesis.serialization.KinesisDeserializationSchema; +import org.apache.flink.streaming.connectors.kinesis.serialization.KinesisDeserializationSchemaWrapper; +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.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.Properties; +import java.util.stream.Collectors; + +import static org.apache.flink.streaming.connectors.kinesis.table.RowDataKinesisDeserializationSchema.Metadata; + +/** + * Kinesis-backed {@link ScanTableSource}. + */ +@Internal +public class KinesisDynamicSource implements ScanTableSource, SupportsReadingMetadata { + + /** List of read-only metadata fields that the source can provide upstream upon request. */ + private static final Map<String, DataType> READABLE_METADATA = + new HashMap<String, DataType>() {{ + for (Metadata metadata : Metadata.values()) { + put(metadata.name(), metadata.getDataType()); + } + }}; + + // -------------------------------------------------------------------------------------------- + // Mutable attributes + // -------------------------------------------------------------------------------------------- + + /** Data type that describes the final output of the source. */ + private DataType producedDataType; + + /** Metadata that is requested to be appended at the end of a physical source row. */ + private List<Metadata> requestedMetadataFields; + + // -------------------------------------------------------------------------------------------- + // Scan format attributes + // -------------------------------------------------------------------------------------------- + + /** Data type to configure the format. */ + private final DataType physicalDataType; + + /** Scan format for decoding records from Kinesis. */ + private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat; + + // -------------------------------------------------------------------------------------------- + // Kinesis-specific attributes + // -------------------------------------------------------------------------------------------- + + /** The Kinesis stream to consume. */ + private final String stream; + + /** Properties for the Kinesis consumer. */ + private final Properties consumerProperties; + + public KinesisDynamicSource( + DataType physicalDataType, + String stream, + Properties consumerProperties, + DecodingFormat<DeserializationSchema<RowData>> decodingFormat) { + this( + physicalDataType, + stream, + consumerProperties, + decodingFormat, + null, + Collections.emptyList()); + } + + public KinesisDynamicSource( + DataType physicalDataType, + String stream, + Properties consumerProperties, + DecodingFormat<DeserializationSchema<RowData>> decodingFormat, + @Nullable DataType producedDataType, + List<Metadata> requestedMetadataFields) { + + this.producedDataType = this.physicalDataType = Preconditions.checkNotNull( + physicalDataType, + "Produced data type must not be null."); + this.stream = Preconditions.checkNotNull( + stream, + "Stream must not be null."); + this.consumerProperties = Preconditions.checkNotNull( + consumerProperties, + "Properties must not be null."); + this.decodingFormat = Preconditions.checkNotNull( + decodingFormat, + "Decoding format must not be null."); + this.producedDataType = producedDataType; Review comment: remove this ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/FixedKinesisPartitioner.java ########## @@ -0,0 +1,77 @@ +/* + * 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.streaming.connectors.kinesis; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.util.Preconditions; + +import java.util.Objects; + +/** + * A partitioner ensuring that each internal Flink partition ends up in the same Kinesis partition. + * + * <p>This is achieved by using the index of the producer task as a {@code PartitionKey}.</p> + */ +@PublicEvolving +public class FixedKinesisPartitioner<T> extends KinesisPartitioner<T> { Review comment: nit: make partitioner classes final? ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisDynamicTableFactory.java ########## @@ -0,0 +1,167 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.api.common.serialization.SerializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.streaming.connectors.kinesis.util.KinesisConfigUtil; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.connector.format.DecodingFormat; +import org.apache.flink.table.connector.format.EncodingFormat; +import org.apache.flink.table.connector.sink.DynamicTableSink; +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.DynamicTableSinkFactory; +import org.apache.flink.table.factories.DynamicTableSourceFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.factories.SerializationFormatFactory; + +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Properties; +import java.util.Set; + +/** + * Factory for creating {@link KinesisDynamicSource} and {@link KinesisDynamicSink} instances. + */ +@Internal +public class KinesisDynamicTableFactory implements + DynamicTableSourceFactory, + DynamicTableSinkFactory { + public static final String IDENTIFIER = "kinesis"; + + // -------------------------------------------------------------------------------------------- + // DynamicTableSourceFactory + // -------------------------------------------------------------------------------------------- + + @Override + public DynamicTableSource createDynamicTableSource(Context context) { + FactoryUtil.TableFactoryHelper helper = FactoryUtil + .createTableFactoryHelper(this, context); + Properties connectorProperties = KinesisOptions + .getConnectorProperties(context.getCatalogTable().getOptions()); + ReadableConfig tableOptions = helper.getOptions(); + + // initialize the table format early in order to register its consumedOptionKeys + // in the TableFactoryHelper, as those are needed for correct option validation + DecodingFormat<DeserializationSchema<RowData>> decodingFormat = helper + .discoverDecodingFormat(DeserializationFormatFactory.class, FactoryUtil.FORMAT); + + // Validate option data types + helper.validateExcept(KinesisOptions.PROPERTIES_PREFIX); + // Validate option values + validateTableOptionsAndConnectorProperties(tableOptions, connectorProperties); + + return new KinesisDynamicSource( + context.getCatalogTable().getSchema().toPhysicalRowDataType(), + tableOptions.get(KinesisOptions.STREAM), + KinesisOptions.getConnectorProperties(context.getCatalogTable().getOptions()), Review comment: reuse variable ########## File path: docs/dev/table/connectors/kinesis.md ########## @@ -0,0 +1,334 @@ +--- +title: "Amazon Kinesis Data Streams SQL Connector" +nav-title: Kinesis +nav-parent_id: sql-connectors +nav-pos: 2 +--- +<!-- +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. +--> + +<span class="label label-primary">Scan Source: Unbounded</span> +<span class="label label-primary">Sink: Streaming Append Mode</span> + +* This will be replaced by the TOC +{:toc} + +The Kinesis connector allows for reading data from and writing data into [Amazon Kinesis Data Streams (KDS)](https://aws.amazon.com/kinesis/data-streams/). + +Dependencies +------------ + +To use the connector, add the following Maven dependency to your project: + +{% highlight xml %} +<dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-kinesis{{ site.scala_version_suffix }}</artifactId> + <version>{{site.version }}</version> +</dependency> +{% endhighlight %} + +How to create a Kinesis data stream table +----------------------------------------- + +Follow the instructions from the [Amazon KDS Developer Guide](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) to set up a Kinesis stream. +The following example shows how to create a table backed by a Kinesis data stream: + +<div class="codetabs" markdown="1"> +<div data-lang="SQL" markdown="1"> +{% highlight sql %} +CREATE TABLE KinesisTable ( + user_id BIGINT, + item_id BIGINT, + category_id BIGINT, + behavior STRING, + ts TIMESTAMP(3) +) PARTITIONED BY (user_id, item_id) WITH ( + 'connector' = 'kinesis', + 'stream' = 'user_behavior', + 'properties.aws.region' = 'us-east-2', + 'properties.flink.stream.initpos' = 'LATEST', + 'format' = 'csv' +) +{% endhighlight %} +</div> +</div> + +Available Metadata +------------------ + +The following metadata can be exposed as read-only (`VIRTUAL`) columns in a table definition. + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Column Name</th> + <th class="text-center" style="width: 45%">Column Type</th> + <th class="text-center" style="width: 35%">Description</th> + </tr> + </thead> + <tbody> + <tr> + <td><code><a href="https://docs.aws.amazon.com/kinesis/latest/APIReference/API_Record.html#Streams-Type-Record-ApproximateArrivalTimestamp";>ApproximateArrivalTimestamp</a></code></td> + <td><code>TIMESTAMP(3) WITH LOCAL TIMEZONE NOT NULL</code></td> + <td>The approximate time that the record was inserted into the stream.</td> + </tr> + <tr> + <td><code><a href="https://docs.aws.amazon.com/kinesis/latest/APIReference/API_Shard.html#Streams-Type-Shard-ShardId";>ShardId</a></code></td> + <td><code>VARCHAR(128) NOT NULL</code></td> + <td>The unique identifier of the shard within the stream from which the record was read.</td> + </tr> + <tr> + <td><code><a href="https://docs.aws.amazon.com/kinesis/latest/APIReference/API_Record.html#Streams-Type-Record-SequenceNumber";>SequenceNumber</a></code></td> + <td><code>VARCHAR(128) NOT NULL</code></td> + <td>The unique identifier of the record within its shard.</td> + </tr> + </tbody> +</table> + +The extended `CREATE TABLE` example demonstrates the syntax for exposing these metadata columns: + +<div class="codetabs" markdown="1"> +<div data-lang="SQL" markdown="1"> +{% highlight sql %} +CREATE TABLE KinesisTable ( + user_id BIGINT, + item_id BIGINT, + category_id BIGINT, + behavior STRING, + ts TIMESTAMP(3), + arrival_time TIMESTAMP(3) METADATA FROM 'ApproximateArrivalTimestamp' VIRTUAL, + shard_id VARCHAR(128) NOT NULL METADATA FROM 'ShardId' VIRTUAL, + sequence_number VARCHAR(128) NOT NULL METADATA FROM 'SequenceNumber' VIRTUAL +) PARTITIONED BY (user_id, item_id) WITH ( + 'connector' = 'kinesis', + 'stream' = 'user_behavior', + 'properties.aws.region' = 'us-east-2', + 'properties.flink.stream.initpos' = 'LATEST', + 'format' = 'csv' +) +{% endhighlight %} +</div> +</div> + +Connector Options +----------------- + +<table class="table table-bordered"> + <thead> + <tr> + <th class="text-left" style="width: 25%">Option</th> + <th class="text-center" style="width: 8%">Required</th> + <th class="text-center" style="width: 7%">Default</th> + <th class="text-center" style="width: 10%">Type</th> + <th class="text-center" style="width: 50%">Description</th> + </tr> + </thead> + <tbody> + <tr> + <td><h5>connector</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Specify what connector to use. For Kinesis use <code>'kinesis'</code>.</td> + </tr> + <tr> + <td><h5>stream</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>Name of the Kinesis data stream backing this table.</td> + </tr> + <tr> + <td><h5>format</h5></td> + <td>required</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>The format used to deserialize and serialize Kinesis data stream records. See <a href="#data-type-mapping">Data Type Mapping</a> for details.</td> + </tr> + <tr> + <td><h5>sink.partitioner</h5></td> + <td>optional</td> + <td style="word-wrap: break-word;">random or row-based</td> + <td>String</td> + <td>Optional output partitioning from Flink's partitions into Kinesis shards. See <a href="#sink-partitioning">Sink Partitioning</a> for details.</td> + </tr> + <tr> + <td><h5>sink.partitioner.field.delimiter</h5></td> + <td>optional</td> + <td style="word-wrap: break-word;">|</td> + <td>String</td> + <td>Optional field delimiter for a fields-based partitioner derived from a PARTITION BY clause. See <a href="#sink-partitioning">Sink Partitioning</a> for details.</td> + </tr> + <tr> + <td><h5>properties.aws.region</h5></td> + <td>optional</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>The AWS region where the stream is defined. Either this or <code>properties.aws.endpoint</code> are required.</td> + </tr> + <tr> + <td><h5>properties.aws.endpoint</h5></td> + <td>optional</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td>The AWS endpoint for Kinesis (derived from the AWS region setting if not set). Either this or <code>properties.aws.region</code> are required.</td> + </tr> + <tr> + <td><h5>properties.aws.credentials.provider</h5></td> + <td>optional</td> + <td style="word-wrap: break-word;">AUTO</td> + <td>String</td> + <td>A credentials provider to use when authenticating against the Kinesis endpoint. See <a href="#authentication">Authentication</a> for details.</td> + </tr> + <tr> + <td><h5>properties.flink.stream.initpos</h5></td> + <td>optional</td> + <td style="word-wrap: break-word;">LATEST</td> + <td>String</td> + <td>Initial position to be used when reading from the table. See <a href="#start-reading-position">Start Reading Position</a> for details.</td> + </tr> + <tr> + <td><h5>properties.flink.stream.recordpublisher</h5></td> + <td>optional</td> + <td style="word-wrap: break-word;">POLLING</td> + <td>String</td> + <td>The `RecordPublisher` type to use for sources. See <a href="#enhanced-fan-out">Enhanced Fan-Out</a> for details.</td> + </tr> + <tr> + <td><h5>properties.*</h5></td> + <td>optional</td> + <td style="word-wrap: break-word;">(none)</td> + <td>String</td> + <td> + Other properties to pass to the FlinkKinesisConsumer or FlinkKinesisProducer constructors. + See the constants defined in + <ul> + <li><code>AWSConfigConstants</code>,</li> + <li><code>ConsumerConfigConstants</code>, and</li> + <li><code>ProducerConfigConstants</code></li> + </ul> + for detailed information on the available suffixes. + </td> + </tr> + </tbody> +</table> + +Features +-------- + +### Authorization + +Make sure to [create an appropriate IAM policy](https://docs.aws.amazon.com/streams/latest/dev/controlling-access.html) to allow reading from / writing to the Kinesis data streams. + +### Authentication + +Depending on your deployment you would choose a different Credentials Provider to allow access to Kinesis. +By default, the `AUTO` Credentials Provider is used. +If the access key ID and secret key are set in the deployment configuration, this results in using the `BASIC` provider. + +A specific [AWSCredentialsProvider](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/AWSCredentialsProvider.html) can be **optionally** set using the `properties.aws.credentials.provider` setting. +Supported values are: + +* `AUTO` - Use the default AWS Credentials Provider chain that searches for credentials in the following order: `ENV_VARS`, `SYS_PROPS`, `WEB_IDENTITY_TOKEN`, `PROFILE`, and EC2/ECS credentials provider. +* `BASIC` - Use access key ID and secret key supplied as configuration. +* `ENV_VAR` - Use `AWS_ACCESS_KEY_ID` & `AWS_SECRET_ACCESS_KEY` environment variables. +* `SYS_PROP` - Use Java system properties `aws.accessKeyId` and `aws.secretKey`. +* `PROFILE` - Use an AWS credentials profile to create the AWS credentials. +* `ASSUME_ROLE` - Create AWS credentials by assuming a role. The credentials for assuming the role must be supplied. +* `WEB_IDENTITY_TOKEN` - Create AWS credentials by assuming a role using Web Identity Token. + +### Start Reading Position + +You can configure table sources to start reading a table-backing Kinesis data stream from a specific position through the `properties.flink.stream.initpos` option. Review comment: I'm wondering if we should also improve those properties for the table connector. `properties.flink.stream.initpos` is very long and has even `flink` in its name. Ideally, we split the properties into `scan.` and `sink.`. See https://cwiki.apache.org/confluence/display/FLINK/FLIP-122%3A+New+Connector+Property+Keys+for+New+Factory for more information. ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisOptions.java ########## @@ -0,0 +1,186 @@ + /* Review comment: this class uses wrong indention ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisOptions.java ########## @@ -0,0 +1,186 @@ + /* + * 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.streaming.connectors.kinesis.table; + + import org.apache.flink.annotation.Internal; + import org.apache.flink.configuration.ConfigOption; + import org.apache.flink.configuration.ConfigOptions; + import org.apache.flink.configuration.ReadableConfig; + import org.apache.flink.streaming.connectors.kinesis.FixedKinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.RandomKinesisPartitioner; + import org.apache.flink.table.api.ValidationException; + import org.apache.flink.table.catalog.CatalogTable; + import org.apache.flink.table.data.RowData; + import org.apache.flink.util.FlinkException; + import org.apache.flink.util.InstantiationUtil; + + import java.util.Map; + import java.util.Properties; + + /** + * Options for Kinesis tables supported by the {@code CREATE TABLE ... WITH ...} clause of the + * Flink SQL dialect and the Flink Table API. + */ + @Internal + public class KinesisOptions { Review comment: make it default scoped? it is internal anyways ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisOptions.java ########## @@ -0,0 +1,186 @@ + /* + * 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.streaming.connectors.kinesis.table; + + import org.apache.flink.annotation.Internal; + import org.apache.flink.configuration.ConfigOption; + import org.apache.flink.configuration.ConfigOptions; + import org.apache.flink.configuration.ReadableConfig; + import org.apache.flink.streaming.connectors.kinesis.FixedKinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.RandomKinesisPartitioner; + import org.apache.flink.table.api.ValidationException; + import org.apache.flink.table.catalog.CatalogTable; + import org.apache.flink.table.data.RowData; + import org.apache.flink.util.FlinkException; + import org.apache.flink.util.InstantiationUtil; + + import java.util.Map; + import java.util.Properties; + + /** + * Options for Kinesis tables supported by the {@code CREATE TABLE ... WITH ...} clause of the + * Flink SQL dialect and the Flink Table API. + */ + @Internal + public class KinesisOptions { + + private KinesisOptions() { + } + + // -------------------------------------------------------------------------------------------- + // Kinesis specific options + // -------------------------------------------------------------------------------------------- + + /** + * Prefix for properties delegated to the + * {@link org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer} and + * {@link org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer}. + */ + public static final String PROPERTIES_PREFIX = "properties."; + + public static final ConfigOption<String> STREAM = ConfigOptions + .key("stream") + .stringType() + .noDefaultValue() + .withDescription("Name of the Kinesis stream backing this table (required)"); + + // -------------------------------------------------------------------------------------------- + // Sink specific options + // -------------------------------------------------------------------------------------------- + + public static final ConfigOption<String> SINK_PARTITIONER = ConfigOptions + .key("sink.partitioner") + .stringType() + .noDefaultValue() + .withDescription( + "Optional output partitioning from Flink's partitions into Kinesis shards. " + + "Sinks that write to tables defined with the PARTITION BY clause " + + "always use a field-based partitioner and cannot define this option. " + + "Valid enumerations are: \n" + + "\"random\":" + + " (use a random partition key),\n" + + "\"fixed\":" + + " (each Flink partition ends up in at most one Kinesis shard),\n" + + "\"custom class name\":" + + " (use a custom KinesisPartitioner subclass)"); + + public static final ConfigOption<String> SINK_PARTITIONER_FIELD_DELIMITER = ConfigOptions + .key("sink.partitioner.field.delimiter") + .stringType() + .defaultValue("|") + .withDescription( + "Optional field delimiter for fields-based partitioner " + + "derived from a PARTITION BY clause (\"|\" by default)"); + + // -------------------------------------------------------------------------------------------- + // Option enumerations + // -------------------------------------------------------------------------------------------- + + public static final String SINK_PARTITIONER_VALUE_FIXED = "fixed"; + + public static final String SINK_PARTITIONER_VALUE_RANDOM = "random"; + + // -------------------------------------------------------------------------------------------- + // Utilities + // -------------------------------------------------------------------------------------------- + + public static Properties getConnectorProperties(Map<String, String> tableOptions) { + final Properties properties = new Properties(); + + tableOptions.keySet().stream() + .filter(key -> key.startsWith(PROPERTIES_PREFIX)) + .forEach(key -> { + final String value = tableOptions.get(key); + final String subKey = key.substring((PROPERTIES_PREFIX).length()); + properties.put(subKey, value); + }); + + return properties; + } + + /** + * Constructs the kinesis partitioner for a {@code targetTable} based on the currently set + * {@code tableOptions}. + * + * <p>The following rules are applied with decreasing precedence order.</p> + * + * <ul> + * <li>IF {@code targetTable} is partitioned, return a {@code RowDataKinesisPartitioner}.</li> + * <li>If the partitioner type is not set, return a {@link RandomKinesisPartitioner}.</li> + * <li>If a specific partitioner type alias is used, instantiate the corresponding type</li> + * <li>Interpret the partitioner type as a classname of a user-defiend partitioner.</li> + * </ul> + * + * @param tableOptions A read-only set of config options that determines the partitioner type. + * @param targetTable A catalog version of the table backing the partitioner. + * @param classLoader A {@link ClassLoader} to use for loading user-defined partitioner classes. + */ + public static KinesisPartitioner<RowData> getKinesisPartitioner( + ReadableConfig tableOptions, + CatalogTable targetTable, + ClassLoader classLoader) { + + if (targetTable.isPartitioned()) { + String delimiter = tableOptions.get(SINK_PARTITIONER_FIELD_DELIMITER); + return new RowDataFieldsKinesisPartitioner(targetTable, delimiter); + } else if (!tableOptions.getOptional(SINK_PARTITIONER).isPresent()) { + return new RandomKinesisPartitioner<>(); + } else { + String partitioner = tableOptions.getOptional(SINK_PARTITIONER).get(); + if (SINK_PARTITIONER_VALUE_FIXED.equals(partitioner)) { + return new FixedKinesisPartitioner<>(); + } else if (SINK_PARTITIONER_VALUE_RANDOM.equals(partitioner)) { + return new RandomKinesisPartitioner<>(); + } else { // interpret the option value as a fully-qualified class name + return initializePartitioner(partitioner, classLoader); + } + } + + } + + /** + * Returns a class value with the given class name. + */ + private static <T> KinesisPartitioner<T> initializePartitioner( + String name, + ClassLoader classLoader) { + try { + Class<?> clazz = Class.forName(name, true, classLoader); + if (!KinesisPartitioner.class.isAssignableFrom(clazz)) { Review comment: use `instantiate(Class<T> clazz, Class<? super T> castTo)` ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitioner.java ########## @@ -0,0 +1,236 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.Preconditions; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * <p>A {@link KinesisPartitioner} of {@link RowData} elements that constructs the partition key + * from a list of field names.</p> Review comment: nit: `</p>` are not necessary in JavaDocs ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisOptions.java ########## @@ -0,0 +1,186 @@ + /* + * 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.streaming.connectors.kinesis.table; + + import org.apache.flink.annotation.Internal; + import org.apache.flink.configuration.ConfigOption; + import org.apache.flink.configuration.ConfigOptions; + import org.apache.flink.configuration.ReadableConfig; + import org.apache.flink.streaming.connectors.kinesis.FixedKinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; + import org.apache.flink.streaming.connectors.kinesis.RandomKinesisPartitioner; + import org.apache.flink.table.api.ValidationException; + import org.apache.flink.table.catalog.CatalogTable; + import org.apache.flink.table.data.RowData; + import org.apache.flink.util.FlinkException; + import org.apache.flink.util.InstantiationUtil; + + import java.util.Map; + import java.util.Properties; + + /** + * Options for Kinesis tables supported by the {@code CREATE TABLE ... WITH ...} clause of the + * Flink SQL dialect and the Flink Table API. + */ + @Internal + public class KinesisOptions { + + private KinesisOptions() { + } + + // -------------------------------------------------------------------------------------------- + // Kinesis specific options + // -------------------------------------------------------------------------------------------- + + /** + * Prefix for properties delegated to the + * {@link org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer} and + * {@link org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer}. + */ + public static final String PROPERTIES_PREFIX = "properties."; + + public static final ConfigOption<String> STREAM = ConfigOptions + .key("stream") + .stringType() + .noDefaultValue() + .withDescription("Name of the Kinesis stream backing this table (required)"); + + // -------------------------------------------------------------------------------------------- + // Sink specific options + // -------------------------------------------------------------------------------------------- + + public static final ConfigOption<String> SINK_PARTITIONER = ConfigOptions + .key("sink.partitioner") + .stringType() + .noDefaultValue() + .withDescription( + "Optional output partitioning from Flink's partitions into Kinesis shards. " + + "Sinks that write to tables defined with the PARTITION BY clause " + + "always use a field-based partitioner and cannot define this option. " + + "Valid enumerations are: \n" + + "\"random\":" + + " (use a random partition key),\n" + + "\"fixed\":" + + " (each Flink partition ends up in at most one Kinesis shard),\n" + + "\"custom class name\":" + + " (use a custom KinesisPartitioner subclass)"); + + public static final ConfigOption<String> SINK_PARTITIONER_FIELD_DELIMITER = ConfigOptions + .key("sink.partitioner.field.delimiter") + .stringType() + .defaultValue("|") + .withDescription( + "Optional field delimiter for fields-based partitioner " + + "derived from a PARTITION BY clause (\"|\" by default)"); + + // -------------------------------------------------------------------------------------------- + // Option enumerations + // -------------------------------------------------------------------------------------------- + + public static final String SINK_PARTITIONER_VALUE_FIXED = "fixed"; + + public static final String SINK_PARTITIONER_VALUE_RANDOM = "random"; + + // -------------------------------------------------------------------------------------------- + // Utilities + // -------------------------------------------------------------------------------------------- + + public static Properties getConnectorProperties(Map<String, String> tableOptions) { + final Properties properties = new Properties(); + + tableOptions.keySet().stream() + .filter(key -> key.startsWith(PROPERTIES_PREFIX)) + .forEach(key -> { + final String value = tableOptions.get(key); + final String subKey = key.substring((PROPERTIES_PREFIX).length()); + properties.put(subKey, value); + }); + + return properties; + } + + /** + * Constructs the kinesis partitioner for a {@code targetTable} based on the currently set + * {@code tableOptions}. + * + * <p>The following rules are applied with decreasing precedence order.</p> + * + * <ul> + * <li>IF {@code targetTable} is partitioned, return a {@code RowDataKinesisPartitioner}.</li> Review comment: nit: some typos `IF`, `user-defiend` ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitioner.java ########## @@ -0,0 +1,236 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.Preconditions; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * <p>A {@link KinesisPartitioner} of {@link RowData} elements that constructs the partition key + * from a list of field names.</p> + * + * <p>The key is constructed by concatenating the string representations of a list of fields + * projected from an input element. A fixed prefix can be optionally configured in order to speed + * up the key construction process.</p> + * + * <p>Resulting partition key values are trimmed to the maximum length allowed by Kinesis.</p> + */ +@Internal +public class RowDataFieldsKinesisPartitioner extends KinesisPartitioner<RowData> { + Review comment: add a `serialVersionUID` to all partitioners ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataKinesisDeserializationSchema.java ########## @@ -0,0 +1,122 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.streaming.connectors.kinesis.serialization.KinesisDeserializationSchema; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.JoinedRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.DataType; +import org.apache.flink.util.Preconditions; + +import java.io.IOException; +import java.util.List; + +/** + * A {@link KinesisDeserializationSchema} adaptor for {@link RowData} records that delegates + * physical data deserialization to an inner {@link DeserializationSchema} and appends requested + * metadata to the end of the deserialized {@link RowData} record. + */ +@Internal +public class RowDataKinesisDeserializationSchema implements KinesisDeserializationSchema<RowData> { + + /** Internal type for enumerating available metadata. */ + protected enum Metadata { + ApproximateArrivalTimestamp(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3).notNull()), + SequenceNumber(DataTypes.VARCHAR(128).notNull()), + ShardId(DataTypes.VARCHAR(128).notNull()); + + private final DataType dataType; + + Metadata(DataType dataType) { + this.dataType = dataType; + } + + public DataType getDataType() { + return this.dataType; + } + } + + private static final long serialVersionUID = 5551095193778230749L; + + /** A {@link DeserializationSchema} to deserialize the physical part of the row. */ + private final DeserializationSchema<RowData> physicalDeserializer; + + /** The type of the produced {@link RowData} records (physical data with appended metadata]. */ + private final TypeInformation<RowData> producedTypeInfo; + + /** Metadata fields to be appended to the physical {@link RowData} in the produced records. */ + private final List<Metadata> requestedMetadataFields; + + public RowDataKinesisDeserializationSchema( + DeserializationSchema<RowData> physicalDeserializer, + TypeInformation<RowData> producedTypeInfo, + List<Metadata> requestedMetadataFields) { + this.physicalDeserializer = Preconditions.checkNotNull(physicalDeserializer); + this.producedTypeInfo = Preconditions.checkNotNull(producedTypeInfo); + this.requestedMetadataFields = Preconditions.checkNotNull(requestedMetadataFields); + } + + @Override + public void open(DeserializationSchema.InitializationContext context) throws Exception { + physicalDeserializer.open(context); + } + + @Override + public RowData deserialize( + byte[] recordValue, + String partitionKey, + String seqNum, + long approxArrivalTimestamp, + String stream, + String shardId) throws IOException { + + RowData physicalRow = physicalDeserializer.deserialize(recordValue); Review comment: We should call `deserialize(byte[] message, Collector<T> out)` instead. ########## File path: flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/TestFormatFactory.java ########## @@ -140,8 +142,28 @@ public int hashCode() { } } + /** + * {@link DeserializationSchema} for testing. + */ + public static class DeserializationSchemaMock implements DeserializationSchema<RowData> { Review comment: keep it private for now? ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitioner.java ########## @@ -0,0 +1,236 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.Preconditions; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * <p>A {@link KinesisPartitioner} of {@link RowData} elements that constructs the partition key + * from a list of field names.</p> + * + * <p>The key is constructed by concatenating the string representations of a list of fields + * projected from an input element. A fixed prefix can be optionally configured in order to speed + * up the key construction process.</p> + * + * <p>Resulting partition key values are trimmed to the maximum length allowed by Kinesis.</p> + */ +@Internal +public class RowDataFieldsKinesisPartitioner extends KinesisPartitioner<RowData> { + + /** + * Allowed maximum length limit of a partition key. + * + * @link https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#API_PutRecord_RequestSyntax + */ + public static final int MAX_PARTITION_KEY_LENGTH = 256; + + /** + * Default delimiter for {@link RowDataFieldsKinesisPartitioner#delimiter}. + */ + public static final String DEFAULT_DELIMITER = String.valueOf('|'); + + /** + * The character used to delimit field values in the concatenated partition key string. + */ + private final String delimiter; + + /** + * A list of field names used to extract the partition key for a record that will be written to + * a Kinesis stream. + */ + private final List<String> fieldNames; + + /** + * A list of getter functions to dynamically extract the field values for all + * {@link RowDataFieldsKinesisPartitioner#fieldNames} from an input record. + */ + private final RowData.FieldGetter[] dynamicFieldGetters; + + /** + * A buffer used to accumulate the concatenation of all field values that form the partition + * key. + */ + private final StringBuilder keyBuffer = new StringBuilder(); + + /** + * A prefix of fixed values to be used instead of the corresponding + * {@link RowDataFieldsKinesisPartitioner#dynamicFieldGetters} entries. + */ + private String[] fixedPrefix = new String[]{}; + + /** + * The cumulative length of all values currently stored in the + * {@link RowDataFieldsKinesisPartitioner#fixedPrefix}. + */ + private int fixedPrefixLength = 0; + + public RowDataFieldsKinesisPartitioner(CatalogTable table) { + this(table, DEFAULT_DELIMITER); + } + + public RowDataFieldsKinesisPartitioner(CatalogTable table, String delimiter) { + Preconditions.checkNotNull(table, "table"); + Preconditions.checkNotNull(delimiter, "delimiter"); + Preconditions.checkArgument( + table.isPartitioned(), + "Cannot create a RowDataFieldsKinesisPartitioner for a non-partitioned table"); + Preconditions.checkArgument( + table.getPartitionKeys().size() == new HashSet<>(table.getPartitionKeys()).size(), + "The sequence of partition keys cannot contain duplicates"); + + TableSchema schema = table.getSchema(); + List<String> schemaFieldsList = Arrays.asList(schema.getFieldNames()); + + String[] badKeys = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + boolean hasField = schema.getTableColumn(fieldName).isPresent(); Review comment: This logic should not work on `TableSchema` but rather on `TableSchema.toPersistedDataType` because computed columns won't be present. Persisted metadata columns could be used in the partitioner. ########## File path: flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisDynamicTableFactoryTest.java ########## @@ -0,0 +1,358 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.streaming.api.functions.source.SourceFunction; +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer; +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer; +import org.apache.flink.streaming.connectors.kinesis.RandomKinesisPartitioner; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.CatalogTableImpl; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.connector.sink.DynamicTableSink; +import org.apache.flink.table.connector.sink.SinkFunctionProvider; +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.factories.FactoryUtil; +import org.apache.flink.table.factories.TableOptionsBuilder; +import org.apache.flink.table.factories.TestFormatFactory; +import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext; +import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext; +import org.apache.flink.table.types.DataType; +import org.apache.flink.util.TestLogger; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.IntStream; +import java.util.stream.Stream; + +import static org.apache.flink.core.testutils.FlinkMatchers.containsCause; +import static org.apache.flink.streaming.connectors.kinesis.table.RowDataKinesisDeserializationSchema.Metadata; +import static org.hamcrest.CoreMatchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; + +/** + * Test for {@link KinesisDynamicSource} and {@link KinesisDynamicSink} created + * by {@link KinesisDynamicTableFactory}. + */ +public class KinesisDynamicTableFactoryTest extends TestLogger { + + private static final String SOURCE_STREAM = "sourceStream"; + private static final String SINK_STREAM = "targetStream"; + + private static final String SOURCE_TABLE = "sourceTable"; + private static final String SINK_TABLE = "sinkTable"; + + private static final Properties KINESIS_PROPERTIES = new Properties() {{ + setProperty("aws.region", "us-west-2"); + }}; + + @Rule + public ExpectedException thrown = ExpectedException.none(); + + // -------------------------------------------------------------------------------------------- + // Positive tests + // -------------------------------------------------------------------------------------------- + + @Test + public void testGoodTableSource() { + TableSchema sourceSchema = defaultSourceSchema().build(); + Map<String, String> sourceOptions = defaultSourceOptions().build(); + + // Construct actual DynamicTableSource using FactoryUtil + KinesisDynamicSource actualSource = (KinesisDynamicSource) + FactoryUtil.createTableSource( + null, + ObjectIdentifier.of("default", "default", SOURCE_TABLE), + createSourceTable(sourceSchema, sourceOptions, Collections.emptyList()), + new Configuration(), + Thread.currentThread().getContextClassLoader(), + false); + + // Construct expected DynamicTableSink using factory under test + KinesisDynamicSource expectedSource = new KinesisDynamicSource( + sourceSchema.toPhysicalRowDataType(), + SOURCE_STREAM, + KINESIS_PROPERTIES, + new TestFormatFactory.DecodingFormatMock(",", true)); + + // verify that the constructed DynamicTableSink is as expected + assertEquals(expectedSource, actualSource); + + // verify that the copy of the constructed DynamicTableSink is as expected + assertEquals(expectedSource, actualSource.copy()); + + // verify produced sink + ScanTableSource.ScanRuntimeProvider functionProvider = + actualSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE); + SourceFunction<RowData> sourceFunction = + as(functionProvider, SourceFunctionProvider.class).createSourceFunction(); + assertThat(sourceFunction, instanceOf(FlinkKinesisConsumer.class)); + } + + @Test + public void testGoodTableSourceWithMetadataFields() { + TableSchema sourceSchema = defaultSourceSchema().build(); + Map<String, String> sourceOptions = defaultSourceOptions().build(); + + // Construct actual DynamicTableSource using FactoryUtil + KinesisDynamicSource actualSource = (KinesisDynamicSource) + FactoryUtil.createTableSource( + null, + ObjectIdentifier.of("default", "default", SOURCE_TABLE), + createSourceTable(sourceSchema, sourceOptions, Collections.emptyList()), + new Configuration(), + Thread.currentThread().getContextClassLoader(), + false); + + // Construct expected DynamicTableSink using factory under test + KinesisDynamicSource expectedSource = new KinesisDynamicSource( + sourceSchema.toPhysicalRowDataType(), + SOURCE_STREAM, + KINESIS_PROPERTIES, + new TestFormatFactory.DecodingFormatMock(",", true)); + + List<String> metadataKeys = Arrays.asList("ShardId", "ApproximateArrivalTimestamp"); + DataType producedDataType = getProducedType(sourceSchema, Metadata.values()); + + expectedSource.applyReadableMetadata(metadataKeys, producedDataType); + actualSource.applyReadableMetadata(metadataKeys, producedDataType); + + // verify that the constructed DynamicTableSink is as expected Review comment: copy past error `DynamicTableSink` here and below ########## File path: flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisDynamicTableFactoryTest.java ########## @@ -0,0 +1,358 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.streaming.api.functions.source.SourceFunction; +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer; +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer; +import org.apache.flink.streaming.connectors.kinesis.RandomKinesisPartitioner; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.CatalogTableImpl; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.connector.sink.DynamicTableSink; +import org.apache.flink.table.connector.sink.SinkFunctionProvider; +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.factories.FactoryUtil; +import org.apache.flink.table.factories.TableOptionsBuilder; +import org.apache.flink.table.factories.TestFormatFactory; +import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext; +import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext; +import org.apache.flink.table.types.DataType; +import org.apache.flink.util.TestLogger; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.IntStream; +import java.util.stream.Stream; + +import static org.apache.flink.core.testutils.FlinkMatchers.containsCause; +import static org.apache.flink.streaming.connectors.kinesis.table.RowDataKinesisDeserializationSchema.Metadata; +import static org.hamcrest.CoreMatchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; + +/** + * Test for {@link KinesisDynamicSource} and {@link KinesisDynamicSink} created + * by {@link KinesisDynamicTableFactory}. + */ +public class KinesisDynamicTableFactoryTest extends TestLogger { + + private static final String SOURCE_STREAM = "sourceStream"; + private static final String SINK_STREAM = "targetStream"; + + private static final String SOURCE_TABLE = "sourceTable"; + private static final String SINK_TABLE = "sinkTable"; + + private static final Properties KINESIS_PROPERTIES = new Properties() {{ + setProperty("aws.region", "us-west-2"); + }}; + + @Rule + public ExpectedException thrown = ExpectedException.none(); + + // -------------------------------------------------------------------------------------------- + // Positive tests + // -------------------------------------------------------------------------------------------- + + @Test + public void testGoodTableSource() { + TableSchema sourceSchema = defaultSourceSchema().build(); + Map<String, String> sourceOptions = defaultSourceOptions().build(); + + // Construct actual DynamicTableSource using FactoryUtil + KinesisDynamicSource actualSource = (KinesisDynamicSource) + FactoryUtil.createTableSource( + null, + ObjectIdentifier.of("default", "default", SOURCE_TABLE), + createSourceTable(sourceSchema, sourceOptions, Collections.emptyList()), + new Configuration(), + Thread.currentThread().getContextClassLoader(), + false); + + // Construct expected DynamicTableSink using factory under test + KinesisDynamicSource expectedSource = new KinesisDynamicSource( + sourceSchema.toPhysicalRowDataType(), + SOURCE_STREAM, + KINESIS_PROPERTIES, + new TestFormatFactory.DecodingFormatMock(",", true)); + + // verify that the constructed DynamicTableSink is as expected + assertEquals(expectedSource, actualSource); + + // verify that the copy of the constructed DynamicTableSink is as expected + assertEquals(expectedSource, actualSource.copy()); + + // verify produced sink + ScanTableSource.ScanRuntimeProvider functionProvider = + actualSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE); + SourceFunction<RowData> sourceFunction = + as(functionProvider, SourceFunctionProvider.class).createSourceFunction(); + assertThat(sourceFunction, instanceOf(FlinkKinesisConsumer.class)); + } + + @Test + public void testGoodTableSourceWithMetadataFields() { + TableSchema sourceSchema = defaultSourceSchema().build(); + Map<String, String> sourceOptions = defaultSourceOptions().build(); + + // Construct actual DynamicTableSource using FactoryUtil + KinesisDynamicSource actualSource = (KinesisDynamicSource) + FactoryUtil.createTableSource( + null, + ObjectIdentifier.of("default", "default", SOURCE_TABLE), + createSourceTable(sourceSchema, sourceOptions, Collections.emptyList()), + new Configuration(), + Thread.currentThread().getContextClassLoader(), + false); + + // Construct expected DynamicTableSink using factory under test + KinesisDynamicSource expectedSource = new KinesisDynamicSource( + sourceSchema.toPhysicalRowDataType(), + SOURCE_STREAM, + KINESIS_PROPERTIES, + new TestFormatFactory.DecodingFormatMock(",", true)); + + List<String> metadataKeys = Arrays.asList("ShardId", "ApproximateArrivalTimestamp"); + DataType producedDataType = getProducedType(sourceSchema, Metadata.values()); + + expectedSource.applyReadableMetadata(metadataKeys, producedDataType); Review comment: we are not really testing something if both call the same methods with the same parameters, instead we should test the contents of mutable member variables ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitioner.java ########## @@ -0,0 +1,236 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.Preconditions; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * <p>A {@link KinesisPartitioner} of {@link RowData} elements that constructs the partition key + * from a list of field names.</p> + * + * <p>The key is constructed by concatenating the string representations of a list of fields + * projected from an input element. A fixed prefix can be optionally configured in order to speed + * up the key construction process.</p> + * + * <p>Resulting partition key values are trimmed to the maximum length allowed by Kinesis.</p> + */ +@Internal +public class RowDataFieldsKinesisPartitioner extends KinesisPartitioner<RowData> { + + /** + * Allowed maximum length limit of a partition key. + * + * @link https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#API_PutRecord_RequestSyntax + */ + public static final int MAX_PARTITION_KEY_LENGTH = 256; + + /** + * Default delimiter for {@link RowDataFieldsKinesisPartitioner#delimiter}. + */ + public static final String DEFAULT_DELIMITER = String.valueOf('|'); + + /** + * The character used to delimit field values in the concatenated partition key string. + */ + private final String delimiter; + + /** + * A list of field names used to extract the partition key for a record that will be written to + * a Kinesis stream. + */ + private final List<String> fieldNames; + + /** + * A list of getter functions to dynamically extract the field values for all + * {@link RowDataFieldsKinesisPartitioner#fieldNames} from an input record. + */ + private final RowData.FieldGetter[] dynamicFieldGetters; + + /** + * A buffer used to accumulate the concatenation of all field values that form the partition + * key. + */ + private final StringBuilder keyBuffer = new StringBuilder(); + + /** + * A prefix of fixed values to be used instead of the corresponding + * {@link RowDataFieldsKinesisPartitioner#dynamicFieldGetters} entries. + */ + private String[] fixedPrefix = new String[]{}; + + /** + * The cumulative length of all values currently stored in the + * {@link RowDataFieldsKinesisPartitioner#fixedPrefix}. + */ + private int fixedPrefixLength = 0; + + public RowDataFieldsKinesisPartitioner(CatalogTable table) { + this(table, DEFAULT_DELIMITER); + } + + public RowDataFieldsKinesisPartitioner(CatalogTable table, String delimiter) { Review comment: I wouldn't pass the entire `CatalogTable` to the partitioner. Instead the parameters should represent what is really needed. The constructor contains too much logic that should rather be put into the `KinesisOptions` utility methods. ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitioner.java ########## @@ -0,0 +1,236 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.Preconditions; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * <p>A {@link KinesisPartitioner} of {@link RowData} elements that constructs the partition key + * from a list of field names.</p> + * + * <p>The key is constructed by concatenating the string representations of a list of fields + * projected from an input element. A fixed prefix can be optionally configured in order to speed + * up the key construction process.</p> + * + * <p>Resulting partition key values are trimmed to the maximum length allowed by Kinesis.</p> + */ +@Internal +public class RowDataFieldsKinesisPartitioner extends KinesisPartitioner<RowData> { + + /** + * Allowed maximum length limit of a partition key. + * + * @link https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#API_PutRecord_RequestSyntax + */ + public static final int MAX_PARTITION_KEY_LENGTH = 256; + + /** + * Default delimiter for {@link RowDataFieldsKinesisPartitioner#delimiter}. + */ + public static final String DEFAULT_DELIMITER = String.valueOf('|'); + + /** + * The character used to delimit field values in the concatenated partition key string. + */ + private final String delimiter; + + /** + * A list of field names used to extract the partition key for a record that will be written to + * a Kinesis stream. + */ + private final List<String> fieldNames; + + /** + * A list of getter functions to dynamically extract the field values for all + * {@link RowDataFieldsKinesisPartitioner#fieldNames} from an input record. + */ + private final RowData.FieldGetter[] dynamicFieldGetters; + + /** + * A buffer used to accumulate the concatenation of all field values that form the partition + * key. + */ + private final StringBuilder keyBuffer = new StringBuilder(); + + /** + * A prefix of fixed values to be used instead of the corresponding + * {@link RowDataFieldsKinesisPartitioner#dynamicFieldGetters} entries. + */ + private String[] fixedPrefix = new String[]{}; + + /** + * The cumulative length of all values currently stored in the + * {@link RowDataFieldsKinesisPartitioner#fixedPrefix}. + */ + private int fixedPrefixLength = 0; + + public RowDataFieldsKinesisPartitioner(CatalogTable table) { + this(table, DEFAULT_DELIMITER); + } + + public RowDataFieldsKinesisPartitioner(CatalogTable table, String delimiter) { + Preconditions.checkNotNull(table, "table"); + Preconditions.checkNotNull(delimiter, "delimiter"); + Preconditions.checkArgument( + table.isPartitioned(), + "Cannot create a RowDataFieldsKinesisPartitioner for a non-partitioned table"); + Preconditions.checkArgument( + table.getPartitionKeys().size() == new HashSet<>(table.getPartitionKeys()).size(), + "The sequence of partition keys cannot contain duplicates"); + + TableSchema schema = table.getSchema(); + List<String> schemaFieldsList = Arrays.asList(schema.getFieldNames()); + + String[] badKeys = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + boolean hasField = schema.getTableColumn(fieldName).isPresent(); + return hasField ? Stream.empty() : Stream.of(fieldName); + }) + .toArray(String[]::new); + Preconditions.checkArgument( + badKeys.length == 0, + "The following partition keys are not present in the table: %s", + String.join(", ", badKeys)); + + this.delimiter = delimiter; + this.fieldNames = table.getPartitionKeys(); + this.dynamicFieldGetters = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + Optional<TableColumn> tableColumn = schema.getTableColumn(fieldName); + // convert Optional to Stream to unpack the value in the enclosing flatMap + return tableColumn.map(Stream::of).orElseGet(Stream::empty); + }) + .map(column -> { + LogicalType fieldType = column.getType().getLogicalType(); + int fieldPos = schemaFieldsList.indexOf(column.getName()); + return RowData.createFieldGetter(fieldType, fieldPos); + }) + .toArray(RowData.FieldGetter[]::new); + } + + @Override + public String getPartitionId(RowData element) { + keyBuffer.setLength(fixedPrefixLength); // reset the dynamic part of the key buffer + for (int i = fixedPrefix.length; i < dynamicFieldGetters.length; i++) { + RowData.FieldGetter getter = dynamicFieldGetters[i]; + Object fieldValue = getter.getFieldOrNull(element); + + keyBuffer.append(fieldValue); + keyBuffer.append(delimiter); + + if (keyBuffer.length() >= MAX_PARTITION_KEY_LENGTH) { + break; // stop when the buffer length exceeds the allowed partition key size + } + } + + // return the accumulated concatenated string trimmed to the max allowed partition key size + int length = Math.min( + keyBuffer.length() - delimiter.length(), Review comment: why `- delimiter.length()`? ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataKinesisDeserializationSchema.java ########## @@ -0,0 +1,122 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.streaming.connectors.kinesis.serialization.KinesisDeserializationSchema; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.JoinedRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.DataType; +import org.apache.flink.util.Preconditions; + +import java.io.IOException; +import java.util.List; + +/** + * A {@link KinesisDeserializationSchema} adaptor for {@link RowData} records that delegates + * physical data deserialization to an inner {@link DeserializationSchema} and appends requested + * metadata to the end of the deserialized {@link RowData} record. + */ +@Internal +public class RowDataKinesisDeserializationSchema implements KinesisDeserializationSchema<RowData> { + + /** Internal type for enumerating available metadata. */ + protected enum Metadata { + ApproximateArrivalTimestamp(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3).notNull()), + SequenceNumber(DataTypes.VARCHAR(128).notNull()), + ShardId(DataTypes.VARCHAR(128).notNull()); + + private final DataType dataType; + + Metadata(DataType dataType) { + this.dataType = dataType; + } + + public DataType getDataType() { + return this.dataType; + } + } + + private static final long serialVersionUID = 5551095193778230749L; + + /** A {@link DeserializationSchema} to deserialize the physical part of the row. */ + private final DeserializationSchema<RowData> physicalDeserializer; + + /** The type of the produced {@link RowData} records (physical data with appended metadata]. */ + private final TypeInformation<RowData> producedTypeInfo; + + /** Metadata fields to be appended to the physical {@link RowData} in the produced records. */ + private final List<Metadata> requestedMetadataFields; + + public RowDataKinesisDeserializationSchema( + DeserializationSchema<RowData> physicalDeserializer, + TypeInformation<RowData> producedTypeInfo, + List<Metadata> requestedMetadataFields) { + this.physicalDeserializer = Preconditions.checkNotNull(physicalDeserializer); + this.producedTypeInfo = Preconditions.checkNotNull(producedTypeInfo); + this.requestedMetadataFields = Preconditions.checkNotNull(requestedMetadataFields); + } + + @Override + public void open(DeserializationSchema.InitializationContext context) throws Exception { + physicalDeserializer.open(context); + } + + @Override + public RowData deserialize( + byte[] recordValue, + String partitionKey, + String seqNum, + long approxArrivalTimestamp, + String stream, + String shardId) throws IOException { + + RowData physicalRow = physicalDeserializer.deserialize(recordValue); + GenericRowData metadataRow = new GenericRowData(requestedMetadataFields.size()); + + for (int i = 0; i < metadataRow.getArity(); i++) { + Metadata metadataField = requestedMetadataFields.get(i); + if (metadataField == Metadata.ApproximateArrivalTimestamp) { + metadataRow.setField(i, TimestampData.fromEpochMillis(approxArrivalTimestamp)); + } else if (metadataField == Metadata.SequenceNumber) { + metadataRow.setField(i, StringData.fromString(seqNum)); + } else if (metadataField == Metadata.ShardId) { + metadataRow.setField(i, StringData.fromString(shardId)); + } else { + String msg = String.format("Unsupported metadata key %s", metadataField); + throw new RuntimeException(msg); // should never happen + } + } + + JoinedRowData joinedRowData = new JoinedRowData(physicalRow, metadataRow); Review comment: I was also thinking about moving this into `table-common`. We can do it but I would rather move it to `org.apache.flink.table.data.utils` because it is not a "core data structure" as the others in the package. In any case we should do this change in a separate commit. The downside of your implementation is that you are creating 2 more objects (row+inner array) but of course this can be neglected. ########## File path: flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitionerTest.java ########## @@ -0,0 +1,241 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.CatalogTableImpl; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.factories.TableOptionsBuilder; +import org.apache.flink.table.factories.TestFormatFactory; +import org.apache.flink.util.TestLogger; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.temporal.ChronoUnit; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; + +import static org.apache.flink.core.testutils.FlinkMatchers.containsCause; +import static org.apache.flink.streaming.connectors.kinesis.table.RowDataFieldsKinesisPartitioner.MAX_PARTITION_KEY_LENGTH; +import static org.junit.Assert.assertEquals; + +/** + * Test for {@link RowDataFieldsKinesisPartitioner}. + */ +public class RowDataFieldsKinesisPartitionerTest extends TestLogger { + + /** + * Table name to use for the tests. + */ + private static final String TABLE_NAME = "click_stream"; + + /** + * Table schema to use for the tests. + */ + private static final TableSchema TABLE_SCHEMA = TableSchema.builder() + .field("time", DataTypes.TIMESTAMP(3)) + .field("date", DataTypes.DATE(), "DATE(`time`)") + .field("ip", DataTypes.VARCHAR(16)) + .field("route", DataTypes.STRING()) + .build(); + + /** + * A {@code PARTITION BY(date, ip)} clause to use for the positive tests. + */ + private static final List<String> PARTITION_BY_DATE_AND_IP = Arrays.asList("date", "ip"); + + /** + * A list of field delimiters to use in the tests. + */ + private static final List<String> FIELD_DELIMITERS = Arrays.asList("", "|", ",", "--"); + + /** + * A {@code PARTITION BY(route)} clause to use for the positive tests. + */ + private static final List<String> PARTITION_BY_ROUTE = Collections.singletonList("route"); + + /** + * Some not-so-random {@link LocalDateTime} instances to use for sample {@link RowData} + * elements in the tests. + */ + private static final List<LocalDateTime> DATE_TIMES = Arrays.asList( + LocalDateTime.of(2014, 10, 22, 14, 0), + LocalDateTime.of(2015, 12, 14, 14, 0)); + + /** + * A default IP to use for sample {@link RowData} elements in the tests. + */ + private static final String IP = "255.255.255.255"; + + @Rule + public ExpectedException thrown = ExpectedException.none(); + + // -------------------------------------------------------------------------------------------- + // Positive tests + // -------------------------------------------------------------------------------------------- + + @Test + public void testGoodPartitionerWithDynamicPrefix() { + CatalogTable table = createTable(defaultTableOptions(), PARTITION_BY_DATE_AND_IP); + + for (String delimiter : FIELD_DELIMITERS) { + RowDataFieldsKinesisPartitioner partitioner = + new RowDataFieldsKinesisPartitioner(table, delimiter); + + for (LocalDateTime time : DATE_TIMES) { + String expectedKey = String.join(delimiter, String.valueOf(days(time)), IP); + String actualKey = partitioner.getPartitionId(createElement(time, IP)); + + assertEquals(expectedKey, actualKey); + } + } + } + + @Test + public void testGoodPartitionerWithDynamicPrefixExceedingMaxLength() { + CatalogTable table = createTable(defaultTableOptions(), PARTITION_BY_ROUTE); + RowDataFieldsKinesisPartitioner partitioner = new RowDataFieldsKinesisPartitioner(table); + + String ip = "255.255.255.255"; + String route = "http://www.very-"; + repeat("long-", 50) + "address.com/home"; + String expectedKey = route.substring(0, MAX_PARTITION_KEY_LENGTH); + + for (LocalDateTime time : DATE_TIMES) { + String actualKey = partitioner.getPartitionId(createElement(time, ip, route)); + assertEquals(expectedKey, actualKey); + } + } + + @Test + public void testGoodPartitionerWithStaticPrefix() { + CatalogTable table = createTable(defaultTableOptions(), PARTITION_BY_DATE_AND_IP); + + // fixed prefix + int date = days(DATE_TIMES.get(0)); + + for (String delimiter : FIELD_DELIMITERS) { + RowDataFieldsKinesisPartitioner partitioner = + new RowDataFieldsKinesisPartitioner(table, delimiter); + + partitioner.setFixedPrefix(new HashMap<String, String>() {{ + put("date", String.valueOf(date)); + }}); + + for (LocalDateTime time : DATE_TIMES) { + String expectedKey = String.join(delimiter, String.valueOf(date), IP); + String actualKey = partitioner.getPartitionId(createElement(time, IP)); + + assertEquals(expectedKey, actualKey); + } + } + } + + // -------------------------------------------------------------------------------------------- + // Negative tests + // -------------------------------------------------------------------------------------------- + + @Test + public void testBadPartitionerWithEmptyPrefix() { + thrown.expect(IllegalArgumentException.class); + thrown.expect(containsCause(new IllegalArgumentException( + "Cannot create a RowDataFieldsKinesisPartitioner for a non-partitioned table"))); + + CatalogTable table = createTable(defaultTableOptions(), Collections.emptyList()); + new RowDataFieldsKinesisPartitioner(table); + } + + @Test + public void testBadPartitionerWithDuplicateFieldNamesInPrefix() { + thrown.expect(IllegalArgumentException.class); + thrown.expect(containsCause(new IllegalArgumentException( + "The sequence of partition keys cannot contain duplicates"))); + + CatalogTable table = createTable(defaultTableOptions(), Arrays.asList("ip", "ip")); + new RowDataFieldsKinesisPartitioner(table); + } + + @Test + public void testBadPartitionerWithBadFieldFieldNamesInPrefix() { + thrown.expect(IllegalArgumentException.class); + thrown.expect(containsCause(new IllegalArgumentException( + "The following partition keys are not present in the table: abc"))); + + CatalogTable table = createTable(defaultTableOptions(), Arrays.asList("ip", "abc")); + new RowDataFieldsKinesisPartitioner(table); + } + + // -------------------------------------------------------------------------------------------- + // Utilities + // -------------------------------------------------------------------------------------------- + + private RowData createElement(LocalDateTime time, String ip) { + return createElement(time, ip, "https://flink.apache.org/home";); + } + + private RowData createElement(LocalDateTime time, String ip, String route) { + GenericRowData element = new GenericRowData(TABLE_SCHEMA.getFieldCount()); + element.setField(0, TimestampData.fromLocalDateTime(time)); + element.setField(1, days(time)); + element.setField(2, StringData.fromString(ip)); + element.setField(3, StringData.fromString(route)); + return element; + } + + private int days(LocalDateTime time) { + return (int) ChronoUnit.DAYS.between(LocalDate.ofEpochDay(0), time); + } + + private CatalogTable createTable(TableOptionsBuilder options, List<String> partitionKeys) { + return new CatalogTableImpl(TABLE_SCHEMA, partitionKeys, options.build(), TABLE_NAME); + } + + private TableOptionsBuilder defaultTableOptions() { + String connector = KinesisDynamicTableFactory.IDENTIFIER; + String format = TestFormatFactory.IDENTIFIER; + return new TableOptionsBuilder(connector, format) + // default table options + .withTableOption(KinesisOptions.STREAM, TABLE_NAME) + .withTableOption("properties.aws.region", "us-west-2") + // default format options + .withFormatOption(TestFormatFactory.DELIMITER, ","); + } + + /** + * Repeat {@code n} times the string {@code s}. + * + * @param s The string to be repeated. + * @param n The number of times to repeat the string. + * + * @return The repeated string. + */ + private static String repeat(String s, int n) { Review comment: you can use `org.apache.flink.table.utils.EncodingUtils#repeat(java.lang.String, int)` ########## File path: flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/table/KinesisDynamicTableFactoryTest.java ########## @@ -0,0 +1,358 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.streaming.api.functions.source.SourceFunction; +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer; +import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer; +import org.apache.flink.streaming.connectors.kinesis.RandomKinesisPartitioner; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.CatalogTableImpl; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.connector.sink.DynamicTableSink; +import org.apache.flink.table.connector.sink.SinkFunctionProvider; +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.factories.FactoryUtil; +import org.apache.flink.table.factories.TableOptionsBuilder; +import org.apache.flink.table.factories.TestFormatFactory; +import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext; +import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext; +import org.apache.flink.table.types.DataType; +import org.apache.flink.util.TestLogger; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.IntStream; +import java.util.stream.Stream; + +import static org.apache.flink.core.testutils.FlinkMatchers.containsCause; +import static org.apache.flink.streaming.connectors.kinesis.table.RowDataKinesisDeserializationSchema.Metadata; +import static org.hamcrest.CoreMatchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; + +/** + * Test for {@link KinesisDynamicSource} and {@link KinesisDynamicSink} created + * by {@link KinesisDynamicTableFactory}. + */ +public class KinesisDynamicTableFactoryTest extends TestLogger { + + private static final String SOURCE_STREAM = "sourceStream"; + private static final String SINK_STREAM = "targetStream"; + + private static final String SOURCE_TABLE = "sourceTable"; + private static final String SINK_TABLE = "sinkTable"; + + private static final Properties KINESIS_PROPERTIES = new Properties() {{ + setProperty("aws.region", "us-west-2"); + }}; + + @Rule + public ExpectedException thrown = ExpectedException.none(); + + // -------------------------------------------------------------------------------------------- + // Positive tests + // -------------------------------------------------------------------------------------------- + + @Test + public void testGoodTableSource() { + TableSchema sourceSchema = defaultSourceSchema().build(); + Map<String, String> sourceOptions = defaultSourceOptions().build(); + + // Construct actual DynamicTableSource using FactoryUtil + KinesisDynamicSource actualSource = (KinesisDynamicSource) + FactoryUtil.createTableSource( + null, + ObjectIdentifier.of("default", "default", SOURCE_TABLE), + createSourceTable(sourceSchema, sourceOptions, Collections.emptyList()), + new Configuration(), + Thread.currentThread().getContextClassLoader(), + false); + + // Construct expected DynamicTableSink using factory under test + KinesisDynamicSource expectedSource = new KinesisDynamicSource( + sourceSchema.toPhysicalRowDataType(), + SOURCE_STREAM, + KINESIS_PROPERTIES, + new TestFormatFactory.DecodingFormatMock(",", true)); + + // verify that the constructed DynamicTableSink is as expected + assertEquals(expectedSource, actualSource); + + // verify that the copy of the constructed DynamicTableSink is as expected + assertEquals(expectedSource, actualSource.copy()); + + // verify produced sink + ScanTableSource.ScanRuntimeProvider functionProvider = + actualSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE); + SourceFunction<RowData> sourceFunction = + as(functionProvider, SourceFunctionProvider.class).createSourceFunction(); + assertThat(sourceFunction, instanceOf(FlinkKinesisConsumer.class)); + } + + @Test + public void testGoodTableSourceWithMetadataFields() { + TableSchema sourceSchema = defaultSourceSchema().build(); + Map<String, String> sourceOptions = defaultSourceOptions().build(); + + // Construct actual DynamicTableSource using FactoryUtil + KinesisDynamicSource actualSource = (KinesisDynamicSource) + FactoryUtil.createTableSource( Review comment: nit: move this call into a helper method also for the sink ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitioner.java ########## @@ -0,0 +1,236 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.Preconditions; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * <p>A {@link KinesisPartitioner} of {@link RowData} elements that constructs the partition key + * from a list of field names.</p> + * + * <p>The key is constructed by concatenating the string representations of a list of fields + * projected from an input element. A fixed prefix can be optionally configured in order to speed + * up the key construction process.</p> + * + * <p>Resulting partition key values are trimmed to the maximum length allowed by Kinesis.</p> + */ +@Internal +public class RowDataFieldsKinesisPartitioner extends KinesisPartitioner<RowData> { + + /** + * Allowed maximum length limit of a partition key. + * + * @link https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#API_PutRecord_RequestSyntax + */ + public static final int MAX_PARTITION_KEY_LENGTH = 256; + + /** + * Default delimiter for {@link RowDataFieldsKinesisPartitioner#delimiter}. + */ + public static final String DEFAULT_DELIMITER = String.valueOf('|'); + + /** + * The character used to delimit field values in the concatenated partition key string. + */ + private final String delimiter; + + /** + * A list of field names used to extract the partition key for a record that will be written to + * a Kinesis stream. + */ + private final List<String> fieldNames; + + /** + * A list of getter functions to dynamically extract the field values for all + * {@link RowDataFieldsKinesisPartitioner#fieldNames} from an input record. + */ + private final RowData.FieldGetter[] dynamicFieldGetters; + + /** + * A buffer used to accumulate the concatenation of all field values that form the partition + * key. + */ + private final StringBuilder keyBuffer = new StringBuilder(); + + /** + * A prefix of fixed values to be used instead of the corresponding + * {@link RowDataFieldsKinesisPartitioner#dynamicFieldGetters} entries. + */ + private String[] fixedPrefix = new String[]{}; + + /** + * The cumulative length of all values currently stored in the + * {@link RowDataFieldsKinesisPartitioner#fixedPrefix}. + */ + private int fixedPrefixLength = 0; + + public RowDataFieldsKinesisPartitioner(CatalogTable table) { + this(table, DEFAULT_DELIMITER); + } + + public RowDataFieldsKinesisPartitioner(CatalogTable table, String delimiter) { + Preconditions.checkNotNull(table, "table"); + Preconditions.checkNotNull(delimiter, "delimiter"); + Preconditions.checkArgument( + table.isPartitioned(), + "Cannot create a RowDataFieldsKinesisPartitioner for a non-partitioned table"); + Preconditions.checkArgument( + table.getPartitionKeys().size() == new HashSet<>(table.getPartitionKeys()).size(), + "The sequence of partition keys cannot contain duplicates"); + + TableSchema schema = table.getSchema(); + List<String> schemaFieldsList = Arrays.asList(schema.getFieldNames()); + + String[] badKeys = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + boolean hasField = schema.getTableColumn(fieldName).isPresent(); + return hasField ? Stream.empty() : Stream.of(fieldName); + }) + .toArray(String[]::new); + Preconditions.checkArgument( + badKeys.length == 0, + "The following partition keys are not present in the table: %s", + String.join(", ", badKeys)); + + this.delimiter = delimiter; + this.fieldNames = table.getPartitionKeys(); + this.dynamicFieldGetters = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + Optional<TableColumn> tableColumn = schema.getTableColumn(fieldName); + // convert Optional to Stream to unpack the value in the enclosing flatMap + return tableColumn.map(Stream::of).orElseGet(Stream::empty); + }) + .map(column -> { + LogicalType fieldType = column.getType().getLogicalType(); + int fieldPos = schemaFieldsList.indexOf(column.getName()); + return RowData.createFieldGetter(fieldType, fieldPos); + }) + .toArray(RowData.FieldGetter[]::new); + } + + @Override + public String getPartitionId(RowData element) { + keyBuffer.setLength(fixedPrefixLength); // reset the dynamic part of the key buffer + for (int i = fixedPrefix.length; i < dynamicFieldGetters.length; i++) { + RowData.FieldGetter getter = dynamicFieldGetters[i]; + Object fieldValue = getter.getFieldOrNull(element); + + keyBuffer.append(fieldValue); + keyBuffer.append(delimiter); + + if (keyBuffer.length() >= MAX_PARTITION_KEY_LENGTH) { + break; // stop when the buffer length exceeds the allowed partition key size + } + } + + // return the accumulated concatenated string trimmed to the max allowed partition key size + int length = Math.min( + keyBuffer.length() - delimiter.length(), + MAX_PARTITION_KEY_LENGTH); + return keyBuffer.substring(0, length); + } + + /** + * Update the fixed partition key prefix. + * + * @param staticFields An association of (field name, field value) pairs to be used as static + * partition key prefix. + */ + public void setFixedPrefix(Map<String, String> staticFields) { + Preconditions.checkArgument( + isPartitionPrefix(staticFields.keySet()), + String.format( + "Not all static field names (%s) are defined as part " + + "of the partition key (%s).", + String.join(", ", staticFields.keySet()), + String.join(", ", fieldNames) + )); + + // update the fixed prefix and pre-compute its cumulative length + fixedPrefix = staticFields.values().toArray(new String[0]); Review comment: don't we assume some implicit order here? the order of partition keys is defined by the `CatalogTable`, static fields must not necessarily be a prefix but can. ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitioner.java ########## @@ -0,0 +1,236 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.Preconditions; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * <p>A {@link KinesisPartitioner} of {@link RowData} elements that constructs the partition key + * from a list of field names.</p> + * + * <p>The key is constructed by concatenating the string representations of a list of fields + * projected from an input element. A fixed prefix can be optionally configured in order to speed + * up the key construction process.</p> + * + * <p>Resulting partition key values are trimmed to the maximum length allowed by Kinesis.</p> + */ +@Internal +public class RowDataFieldsKinesisPartitioner extends KinesisPartitioner<RowData> { + + /** + * Allowed maximum length limit of a partition key. + * + * @link https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#API_PutRecord_RequestSyntax + */ + public static final int MAX_PARTITION_KEY_LENGTH = 256; + + /** + * Default delimiter for {@link RowDataFieldsKinesisPartitioner#delimiter}. + */ + public static final String DEFAULT_DELIMITER = String.valueOf('|'); + + /** + * The character used to delimit field values in the concatenated partition key string. + */ + private final String delimiter; + + /** + * A list of field names used to extract the partition key for a record that will be written to + * a Kinesis stream. + */ + private final List<String> fieldNames; + + /** + * A list of getter functions to dynamically extract the field values for all + * {@link RowDataFieldsKinesisPartitioner#fieldNames} from an input record. + */ + private final RowData.FieldGetter[] dynamicFieldGetters; + + /** + * A buffer used to accumulate the concatenation of all field values that form the partition + * key. + */ + private final StringBuilder keyBuffer = new StringBuilder(); + + /** + * A prefix of fixed values to be used instead of the corresponding + * {@link RowDataFieldsKinesisPartitioner#dynamicFieldGetters} entries. + */ + private String[] fixedPrefix = new String[]{}; + + /** + * The cumulative length of all values currently stored in the + * {@link RowDataFieldsKinesisPartitioner#fixedPrefix}. + */ + private int fixedPrefixLength = 0; + + public RowDataFieldsKinesisPartitioner(CatalogTable table) { + this(table, DEFAULT_DELIMITER); + } + + public RowDataFieldsKinesisPartitioner(CatalogTable table, String delimiter) { + Preconditions.checkNotNull(table, "table"); + Preconditions.checkNotNull(delimiter, "delimiter"); + Preconditions.checkArgument( + table.isPartitioned(), + "Cannot create a RowDataFieldsKinesisPartitioner for a non-partitioned table"); + Preconditions.checkArgument( + table.getPartitionKeys().size() == new HashSet<>(table.getPartitionKeys()).size(), + "The sequence of partition keys cannot contain duplicates"); + + TableSchema schema = table.getSchema(); + List<String> schemaFieldsList = Arrays.asList(schema.getFieldNames()); + + String[] badKeys = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + boolean hasField = schema.getTableColumn(fieldName).isPresent(); + return hasField ? Stream.empty() : Stream.of(fieldName); + }) + .toArray(String[]::new); + Preconditions.checkArgument( + badKeys.length == 0, + "The following partition keys are not present in the table: %s", + String.join(", ", badKeys)); + + this.delimiter = delimiter; + this.fieldNames = table.getPartitionKeys(); + this.dynamicFieldGetters = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + Optional<TableColumn> tableColumn = schema.getTableColumn(fieldName); + // convert Optional to Stream to unpack the value in the enclosing flatMap + return tableColumn.map(Stream::of).orElseGet(Stream::empty); + }) + .map(column -> { + LogicalType fieldType = column.getType().getLogicalType(); + int fieldPos = schemaFieldsList.indexOf(column.getName()); + return RowData.createFieldGetter(fieldType, fieldPos); + }) + .toArray(RowData.FieldGetter[]::new); + } + + @Override + public String getPartitionId(RowData element) { + keyBuffer.setLength(fixedPrefixLength); // reset the dynamic part of the key buffer + for (int i = fixedPrefix.length; i < dynamicFieldGetters.length; i++) { + RowData.FieldGetter getter = dynamicFieldGetters[i]; + Object fieldValue = getter.getFieldOrNull(element); + + keyBuffer.append(fieldValue); + keyBuffer.append(delimiter); + + if (keyBuffer.length() >= MAX_PARTITION_KEY_LENGTH) { + break; // stop when the buffer length exceeds the allowed partition key size + } + } + + // return the accumulated concatenated string trimmed to the max allowed partition key size + int length = Math.min( + keyBuffer.length() - delimiter.length(), + MAX_PARTITION_KEY_LENGTH); + return keyBuffer.substring(0, length); + } + + /** + * Update the fixed partition key prefix. + * + * @param staticFields An association of (field name, field value) pairs to be used as static + * partition key prefix. + */ + public void setFixedPrefix(Map<String, String> staticFields) { + Preconditions.checkArgument( + isPartitionPrefix(staticFields.keySet()), + String.format( + "Not all static field names (%s) are defined as part " + + "of the partition key (%s).", + String.join(", ", staticFields.keySet()), + String.join(", ", fieldNames) + )); + + // update the fixed prefix and pre-compute its cumulative length + fixedPrefix = staticFields.values().toArray(new String[0]); + String fixedPrefixString = Arrays.stream(fixedPrefix) + .map(s -> s + delimiter) + .collect(Collectors.joining()); + fixedPrefixLength = fixedPrefixString.length(); + + // set a concatenated string of all fixed values as a prefix in the partition key buffer + keyBuffer.replace(0, fixedPrefixLength, fixedPrefixString); + keyBuffer.setLength(fixedPrefixLength); + } + + /** + * Check whether the set of field names in {@code candidatePrefix} forms a valid prefix of the + * list of field names defined in {@link RowDataFieldsKinesisPartitioner#fieldNames} (assuming + * the latter does not contain duplicates). + * + * @param candidatePrefix A set of field names forming a prefix of + * {@link RowDataFieldsKinesisPartitioner#fieldNames}. + * + * @return true if and only if the {@code candidatePrefix} is actually a prefix. + */ + private boolean isPartitionPrefix(Set<String> candidatePrefix) { + for (String name : fieldNames.subList(0, candidatePrefix.size())) { + if (!candidatePrefix.contains(name)) { + return false; // name at current position is not in the candidatePrefix + } + } + return true; + } + + // -------------------------------------------------------------------------------------------- + // Value semantics for equals and hashCode + // -------------------------------------------------------------------------------------------- + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + final RowDataFieldsKinesisPartitioner that = (RowDataFieldsKinesisPartitioner) o; + return Objects.equals(this.fieldNames, that.fieldNames) && + Arrays.equals(this.fixedPrefix, that.fixedPrefix); + } + + @Override + public int hashCode() { + return Objects.hash( + fieldNames, + Arrays.hashCode(fixedPrefix)); Review comment: remove `Arrays.hashCode`, shall we add the other fields as well for testing? ########## File path: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/table/RowDataFieldsKinesisPartitioner.java ########## @@ -0,0 +1,236 @@ +/* + * 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.streaming.connectors.kinesis.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.streaming.connectors.kinesis.KinesisPartitioner; +import org.apache.flink.table.api.TableColumn; +import org.apache.flink.table.api.TableSchema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.util.Preconditions; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * <p>A {@link KinesisPartitioner} of {@link RowData} elements that constructs the partition key + * from a list of field names.</p> + * + * <p>The key is constructed by concatenating the string representations of a list of fields + * projected from an input element. A fixed prefix can be optionally configured in order to speed + * up the key construction process.</p> + * + * <p>Resulting partition key values are trimmed to the maximum length allowed by Kinesis.</p> + */ +@Internal +public class RowDataFieldsKinesisPartitioner extends KinesisPartitioner<RowData> { + + /** + * Allowed maximum length limit of a partition key. + * + * @link https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html#API_PutRecord_RequestSyntax + */ + public static final int MAX_PARTITION_KEY_LENGTH = 256; + + /** + * Default delimiter for {@link RowDataFieldsKinesisPartitioner#delimiter}. + */ + public static final String DEFAULT_DELIMITER = String.valueOf('|'); + + /** + * The character used to delimit field values in the concatenated partition key string. + */ + private final String delimiter; + + /** + * A list of field names used to extract the partition key for a record that will be written to + * a Kinesis stream. + */ + private final List<String> fieldNames; + + /** + * A list of getter functions to dynamically extract the field values for all + * {@link RowDataFieldsKinesisPartitioner#fieldNames} from an input record. + */ + private final RowData.FieldGetter[] dynamicFieldGetters; + + /** + * A buffer used to accumulate the concatenation of all field values that form the partition + * key. + */ + private final StringBuilder keyBuffer = new StringBuilder(); + + /** + * A prefix of fixed values to be used instead of the corresponding + * {@link RowDataFieldsKinesisPartitioner#dynamicFieldGetters} entries. + */ + private String[] fixedPrefix = new String[]{}; + + /** + * The cumulative length of all values currently stored in the + * {@link RowDataFieldsKinesisPartitioner#fixedPrefix}. + */ + private int fixedPrefixLength = 0; + + public RowDataFieldsKinesisPartitioner(CatalogTable table) { + this(table, DEFAULT_DELIMITER); + } + + public RowDataFieldsKinesisPartitioner(CatalogTable table, String delimiter) { + Preconditions.checkNotNull(table, "table"); + Preconditions.checkNotNull(delimiter, "delimiter"); + Preconditions.checkArgument( + table.isPartitioned(), + "Cannot create a RowDataFieldsKinesisPartitioner for a non-partitioned table"); + Preconditions.checkArgument( + table.getPartitionKeys().size() == new HashSet<>(table.getPartitionKeys()).size(), + "The sequence of partition keys cannot contain duplicates"); + + TableSchema schema = table.getSchema(); + List<String> schemaFieldsList = Arrays.asList(schema.getFieldNames()); + + String[] badKeys = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + boolean hasField = schema.getTableColumn(fieldName).isPresent(); + return hasField ? Stream.empty() : Stream.of(fieldName); + }) + .toArray(String[]::new); + Preconditions.checkArgument( + badKeys.length == 0, + "The following partition keys are not present in the table: %s", + String.join(", ", badKeys)); + + this.delimiter = delimiter; + this.fieldNames = table.getPartitionKeys(); + this.dynamicFieldGetters = table.getPartitionKeys().stream() + .flatMap(fieldName -> { + Optional<TableColumn> tableColumn = schema.getTableColumn(fieldName); + // convert Optional to Stream to unpack the value in the enclosing flatMap + return tableColumn.map(Stream::of).orElseGet(Stream::empty); + }) + .map(column -> { + LogicalType fieldType = column.getType().getLogicalType(); + int fieldPos = schemaFieldsList.indexOf(column.getName()); + return RowData.createFieldGetter(fieldType, fieldPos); + }) + .toArray(RowData.FieldGetter[]::new); + } + + @Override + public String getPartitionId(RowData element) { + keyBuffer.setLength(fixedPrefixLength); // reset the dynamic part of the key buffer + for (int i = fixedPrefix.length; i < dynamicFieldGetters.length; i++) { + RowData.FieldGetter getter = dynamicFieldGetters[i]; + Object fieldValue = getter.getFieldOrNull(element); + + keyBuffer.append(fieldValue); + keyBuffer.append(delimiter); + + if (keyBuffer.length() >= MAX_PARTITION_KEY_LENGTH) { + break; // stop when the buffer length exceeds the allowed partition key size + } + } + + // return the accumulated concatenated string trimmed to the max allowed partition key size + int length = Math.min( + keyBuffer.length() - delimiter.length(), + MAX_PARTITION_KEY_LENGTH); + return keyBuffer.substring(0, length); + } + + /** + * Update the fixed partition key prefix. + * + * @param staticFields An association of (field name, field value) pairs to be used as static + * partition key prefix. + */ + public void setFixedPrefix(Map<String, String> staticFields) { + Preconditions.checkArgument( + isPartitionPrefix(staticFields.keySet()), + String.format( + "Not all static field names (%s) are defined as part " + + "of the partition key (%s).", + String.join(", ", staticFields.keySet()), + String.join(", ", fieldNames) + )); + + // update the fixed prefix and pre-compute its cumulative length + fixedPrefix = staticFields.values().toArray(new String[0]); Review comment: the exception above could occur quite often, instead we should make static partitions possible even if they are not a prefix ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: [email protected]
