You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/01/13 12:57:32 UTC
[GitHub] [flink] fapaul commented on a change in pull request #18290: [FLINK-25391][connectors][formats] Update existing table factories/format factories for catalog table options forwarding
fapaul commented on a change in pull request #18290:
URL: https://github.com/apache/flink/pull/18290#discussion_r783923691
##########
File path: docs/content/docs/connectors/table/elasticsearch.md
##########
@@ -67,15 +67,17 @@ Connector Options
<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: 8%">Forwarded</th>
Review comment:
Can we come up with a better term than `forwarded` to make clear that the state becomes incompatible? `forwarded` is derived by the actual implementation but I do not think many users can relate the term.
##########
File path: docs/content/docs/connectors/table/kinesis.md
##########
@@ -136,34 +137,39 @@ Connector Options
<tr>
<td><h5>connector</h5></td>
<td>required</td>
+ <td>no</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>yes</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>no</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>aws.region</h5></td>
<td>optional</td>
+ <td>no</td>
Review comment:
Why no?
##########
File path: docs/content/docs/connectors/table/hbase.md
##########
@@ -103,34 +105,39 @@ Connector Options
<tr>
<td><h5>table-name</h5></td>
<td>required</td>
+ <td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>The name of HBase table to connect. By default, the table is in 'default' namespace. To assign the table a specified namespace you need to use 'namespace:table'.</td>
</tr>
<tr>
<td><h5>zookeeper.quorum</h5></td>
<td>required</td>
+ <td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>The HBase Zookeeper quorum.</td>
</tr>
<tr>
<td><h5>zookeeper.znode.parent</h5></td>
<td>optional</td>
+ <td>yes</td>
<td style="word-wrap: break-word;">/hbase</td>
<td>String</td>
<td>The root dir in Zookeeper for HBase cluster.</td>
</tr>
<tr>
<td><h5>null-string-literal</h5></td>
<td>optional</td>
+ <td>yes</td>
Review comment:
Can this make the state incompatible?
##########
File path: docs/content/docs/connectors/table/filesystem.md
##########
@@ -325,33 +347,43 @@ Time extractors define extracting time from partition values.
<table class="table table-bordered">
<thead>
<tr>
- <th class="text-left" style="width: 20%">Key</th>
- <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-left" style="width: 8%">Required</th>
+ <th class="text-left" style="width: 8%">Forwarded</th>
+ <th class="text-left" style="width: 7%">Default</th>
<th class="text-left" style="width: 10%">Type</th>
- <th class="text-left" style="width: 55%">Description</th>
+ <th class="text-left" style="width: 42%">Description</th>
</tr>
</thead>
<tbody>
<tr>
<td><h5>partition.time-extractor.kind</h5></td>
+ <td>optional</td>
+ <td>yes</td>
<td style="word-wrap: break-word;">default</td>
<td>String</td>
<td>Time extractor to extract time from partition values. Support default and custom. For default, can configure timestamp pattern\formatter. For custom, should configure extractor class.</td>
</tr>
<tr>
<td><h5>partition.time-extractor.class</h5></td>
+ <td>optional</td>
+ <td>yes</td>
Review comment:
All these partition configuratiosn sound like they change the transformation
##########
File path: docs/content/docs/connectors/table/kafka.md
##########
@@ -179,50 +179,57 @@ Connector Options
<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: 8%">Forwarded</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>
+ <th class="text-center" style="width: 42%">Description</th>
</tr>
</thead>
<tbody>
<tr>
<td><h5>connector</h5></td>
<td>required</td>
+ <td>no</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>Specify what connector to use, for Kafka use <code>'kafka'</code>.</td>
</tr>
<tr>
<td><h5>topic</h5></td>
<td>required for sink</td>
+ <td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>Topic name(s) to read data from when the table is used as source. It also supports topic list for source by separating topic by semicolon like <code>'topic-1;topic-2'</code>. Note, only one of "topic-pattern" and "topic" can be specified for sources. When the table is used as sink, the topic name is the topic to write data to. Note topic list is not supported for sinks.</td>
</tr>
<tr>
<td><h5>topic-pattern</h5></td>
<td>optional</td>
+ <td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>The regular expression for a pattern of topic names to read from. All topics with names that match the specified regular expression will be subscribed by the consumer when the job starts running. Note, only one of "topic-pattern" and "topic" can be specified for sources.</td>
</tr>
<tr>
<td><h5>properties.bootstrap.servers</h5></td>
<td>required</td>
+ <td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>Comma separated list of Kafka brokers.</td>
</tr>
<tr>
<td><h5>properties.group.id</h5></td>
<td>optional for source, not applicable for sink</td>
+ <td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>The id of the consumer group for Kafka source. If group ID is not specified, an automatically generated id "KafkaSource-{tableIdentifier}" will be used.</td>
</tr>
<tr>
<td><h5>properties.*</h5></td>
<td>optional</td>
+ <td>no</td>
Review comment:
Why is this no?
##########
File path: docs/content/docs/connectors/table/kinesis.md
##########
@@ -255,349 +271,399 @@ Connector Options
<tr>
<td><h5>scan.stream.initpos</h5></td>
<td>optional</td>
+ <td>no</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>scan.stream.initpos-timestamp</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>The initial timestamp to start reading Kinesis stream from (when <code>scan.stream.initpos</code> is AT_TIMESTAMP). See <a href="#start-reading-position">Start Reading Position</a> for details.</td>
</tr>
<tr>
<td><h5>scan.stream.initpos-timestamp-format</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">yyyy-MM-dd'T'HH:mm:ss.SSSXXX</td>
<td>String</td>
<td>The date format of initial timestamp to start reading Kinesis stream from (when <code>scan.stream.initpos</code> is AT_TIMESTAMP). See <a href="#start-reading-position">Start Reading Position</a> for details.</td>
</tr>
<tr>
<td><h5>scan.stream.recordpublisher</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">POLLING</td>
<td>String</td>
<td>The <code>RecordPublisher</code> type to use for sources. See <a href="#enhanced-fan-out">Enhanced Fan-Out</a> for details.</td>
</tr>
<tr>
<td><h5>scan.stream.efo.consumername</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>The name of the EFO consumer to register with KDS. See <a href="#enhanced-fan-out">Enhanced Fan-Out</a> for details.</td>
</tr>
<tr>
<td><h5>scan.stream.efo.registration</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">LAZY</td>
<td>String</td>
<td>Determine how and when consumer de-/registration is performed (LAZY|EAGER|NONE). See <a href="#enhanced-fan-out">Enhanced Fan-Out</a> for details.</td>
</tr>
<tr>
<td><h5>scan.stream.efo.consumerarn</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>The prefix of consumer ARN for a given stream. See <a href="#enhanced-fan-out">Enhanced Fan-Out</a> for details.</td>
</tr>
<tr>
<td><h5>scan.stream.efo.http-client.max-concurrency</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">10000</td>
<td>Integer</td>
<td>Maximum number of allowed concurrent requests for the EFO client. See <a href="#enhanced-fan-out">Enhanced Fan-Out</a> for details.</td>
</tr>
<tr>
<td><h5>scan.stream.describe.maxretries</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">50</td>
<td>Integer</td>
<td>The maximum number of <code>describeStream</code> attempts if we get a recoverable exception.</td>
</tr>
<tr>
<td><h5>scan.stream.describe.backoff.base</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">2000</td>
<td>Long</td>
<td>The base backoff time (in milliseconds) between each <code>describeStream</code> attempt (for consuming from DynamoDB streams).</td>
</tr>
<tr>
<td><h5>scan.stream.describe.backoff.max</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">5000</td>
<td>Long</td>
<td>The maximum backoff time (in milliseconds) between each <code>describeStream</code> attempt (for consuming from DynamoDB streams).</td>
</tr>
<tr>
<td><h5>scan.stream.describe.backoff.expconst</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
<td>The power constant for exponential backoff between each <code>describeStream</code> attempt (for consuming from DynamoDB streams).</td>
</tr>
<tr>
<td><h5>scan.list.shards.maxretries</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">10</td>
<td>Integer</td>
<td>The maximum number of <code>listShards</code> attempts if we get a recoverable exception.</td>
</tr>
<tr>
<td><h5>scan.list.shards.backoff.base</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1000</td>
<td>Long</td>
<td>The base backoff time (in milliseconds) between each <code>listShards</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.list.shards.backoff.max</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">5000</td>
<td>Long</td>
<td>The maximum backoff time (in milliseconds) between each <code>listShards</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.list.shards.backoff.expconst</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
<td>The power constant for exponential backoff between each <code>listShards</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.stream.describestreamconsumer.maxretries</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">50</td>
<td>Integer</td>
<td>The maximum number of <code>describeStreamConsumer</code> attempts if we get a recoverable exception.</td>
</tr>
<tr>
<td><h5>scan.stream.describestreamconsumer.backoff.base</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">2000</td>
<td>Long</td>
<td>The base backoff time (in milliseconds) between each <code>describeStreamConsumer</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.stream.describestreamconsumer.backoff.max</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">5000</td>
<td>Long</td>
<td>The maximum backoff time (in milliseconds) between each <code>describeStreamConsumer</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.stream.describestreamconsumer.backoff.expconst</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
<td>The power constant for exponential backoff between each <code>describeStreamConsumer</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.stream.registerstreamconsumer.maxretries</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">10</td>
<td>Integer</td>
<td>The maximum number of <code>registerStream</code> attempts if we get a recoverable exception.</td>
</tr>
<tr>
<td><h5>scan.stream.registerstreamconsumer.timeout</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">60</td>
<td>Integer</td>
<td>The maximum time in seconds to wait for a stream consumer to become active before giving up.</td>
</tr>
<tr>
<td><h5>scan.stream.registerstreamconsumer.backoff.base</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">500</td>
<td>Long</td>
<td>The base backoff time (in milliseconds) between each <code>registerStream</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.stream.registerstreamconsumer.backoff.max</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">2000</td>
<td>Long</td>
<td>The maximum backoff time (in milliseconds) between each <code>registerStream</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.stream.registerstreamconsumer.backoff.expconst</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
<td>The power constant for exponential backoff between each <code>registerStream</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.stream.deregisterstreamconsumer.maxretries</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">10</td>
<td>Integer</td>
<td>The maximum number of <code>deregisterStream</code> attempts if we get a recoverable exception.</td>
</tr>
<tr>
<td><h5>scan.stream.deregisterstreamconsumer.timeout</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">60</td>
<td>Integer</td>
<td>The maximum time in seconds to wait for a stream consumer to deregister before giving up.</td>
</tr>
<tr>
<td><h5>scan.stream.deregisterstreamconsumer.backoff.base</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">500</td>
<td>Long</td>
<td>The base backoff time (in milliseconds) between each <code>deregisterStream</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.stream.deregisterstreamconsumer.backoff.max</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">2000</td>
<td>Long</td>
<td>The maximum backoff time (in milliseconds) between each <code>deregisterStream</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.stream.deregisterstreamconsumer.backoff.expconst</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
<td>The power constant for exponential backoff between each <code>deregisterStream</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.shard.subscribetoshard.maxretries</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">10</td>
<td>Integer</td>
<td>The maximum number of <code>subscribeToShard</code> attempts if we get a recoverable exception.</td>
</tr>
<tr>
<td><h5>scan.shard.subscribetoshard.backoff.base</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1000</td>
<td>Long</td>
<td>The base backoff time (in milliseconds) between each <code>subscribeToShard</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.shard.subscribetoshard.backoff.max</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">2000</td>
<td>Long</td>
<td>The maximum backoff time (in milliseconds) between each <code>subscribeToShard</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.shard.subscribetoshard.backoff.expconst</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
<td>The power constant for exponential backoff between each <code>subscribeToShard</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.shard.getrecords.maxrecordcount</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">10000</td>
<td>Integer</td>
<td>The maximum number of records to try to get each time we fetch records from a AWS Kinesis shard.</td>
</tr>
<tr>
<td><h5>scan.shard.getrecords.maxretries</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">3</td>
<td>Integer</td>
<td>The maximum number of <code>getRecords</code> attempts if we get a recoverable exception.</td>
</tr>
<tr>
<td><h5>scan.shard.getrecords.backoff.base</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">300</td>
<td>Long</td>
<td>The base backoff time (in milliseconds) between <code>getRecords</code> attempts if we get a ProvisionedThroughputExceededException.</td>
</tr>
<tr>
<td><h5>scan.shard.getrecords.backoff.max</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1000</td>
<td>Long</td>
<td>The maximum backoff time (in milliseconds) between <code>getRecords</code> attempts if we get a ProvisionedThroughputExceededException.</td>
</tr>
<tr>
<td><h5>scan.shard.getrecords.backoff.expconst</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
<td>The power constant for exponential backoff between each <code>getRecords</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.shard.getrecords.intervalmillis</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">200</td>
<td>Long</td>
<td>The interval (in milliseconds) between each <code>getRecords</code> request to a AWS Kinesis shard in milliseconds.</td>
</tr>
<tr>
<td><h5>scan.shard.getiterator.maxretries</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">3</td>
<td>Integer</td>
<td>The maximum number of <code>getShardIterator</code> attempts if we get ProvisionedThroughputExceededException.</td>
</tr>
<tr>
<td><h5>scan.shard.getiterator.backoff.base</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">300</td>
<td>Long</td>
<td>The base backoff time (in milliseconds) between <code>getShardIterator</code> attempts if we get a ProvisionedThroughputExceededException.</td>
</tr>
<tr>
<td><h5>scan.shard.getiterator.backoff.max</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1000</td>
<td>Long</td>
<td>The maximum backoff time (in milliseconds) between <code>getShardIterator</code> attempts if we get a ProvisionedThroughputExceededException.</td>
</tr>
<tr>
<td><h5>scan.shard.getiterator.backoff.expconst</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">1.5</td>
<td>Double</td>
<td>The power constant for exponential backoff between each <code>getShardIterator</code> attempt.</td>
</tr>
<tr>
<td><h5>scan.shard.discovery.intervalmillis</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">10000</td>
<td>Integer</td>
<td>The interval between each attempt to discover new shards.</td>
</tr>
<tr>
<td><h5>scan.shard.adaptivereads</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>The config to turn on adaptive reads from a shard. See the <code>AdaptivePollingRecordPublisher</code> documentation for details.</td>
</tr>
<tr>
<td><h5>scan.shard.idle.interval</h5></td>
<td>optional</td>
+ <td>no</td>
<td style="word-wrap: break-word;">-1</td>
<td>Long</td>
<td>The interval (in milliseconds) after which to consider a shard idle for purposes of watermark generation. A positive value will allow the watermark to progress even when some shards don't receive new records.</td>
</tr>
<tr>
<td><h5>scan.watermark.sync.interval</h5></td>
<td>optional</td>
+ <td>no</td>
Review comment:
I don't understand most of the nos for the kinesis connector.
##########
File path: docs/content/docs/connectors/table/kinesis.md
##########
@@ -136,34 +137,39 @@ Connector Options
<tr>
<td><h5>connector</h5></td>
<td>required</td>
+ <td>no</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>yes</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>no</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>aws.region</h5></td>
<td>optional</td>
+ <td>no</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>aws.endpoint</code> are required.</td>
</tr>
<tr>
<td><h5>aws.endpoint</h5></td>
<td>optional</td>
+ <td>no</td>
Review comment:
No?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org