Confluent Cloud for Apache Flink®️ is currently available for Preview. A Preview feature is a Confluent Cloud component that is being introduced to gain early feedback from developers. Preview features can be used for evaluation and non-production testing purposes or to provide feedback to Confluent. The warranty, SLA, and Support Services provisions of your agreement with Confluent do not apply to Preview features. Confluent may discontinue providing Preview releases of the Preview features at any time in Confluent’s sole discretion. Check out Getting Help for questions, feedback and requests.
For Flink SQL features and limitations in the preview program, see Notable Limitations in Public Preview.
Confluent Cloud for Apache Flink®️ enables setting Flink SQL shell properties to different values.
SET 'key' = 'value';
Modify or list the Flink SQL shell configuration.
If no key and value are specified,
SET prints all of the properties that
you have assigned for the session.
To reset a session property to its default value, use the RESET Statement.
The following examples show how to run a
SET statement in the Flink SQL
SET 'table.local-time-zone' = 'America/Los_Angeles';
Your output should resemble:
Statement successfully submitted. Statement phase is COMPLETED. configuration updated successfully.
To list the current session settings, run the
SET command with no
Your output should resemble:
Statement successfully submitted. Statement phase is COMPLETED. +-----------------------+--------------------------+ | Key | Value | +-----------------------+--------------------------+ | catalog | default (default) | | default_database | <your_cluster> (default) | | table.local-time-zone | America/Los_Angeles | +-----------------------+--------------------------+
Available SET Statements¶
These are the available SET statements in Confluent Cloud for Flink SQL.
|client.results-timeout||600000||Long||Total amount of time in milliseconds to wait before timing out the request waiting for results to be ready.|
|client.statement-name||(None)||String||Give your Flink statement a meaningful name that can help you identify it more easily. Instead of 123e4567-e89b-12d3, this sets the statement name to the given value. The name resets itself after successful submission to avoid naming conflicts.|
|sql.current-catalog||(None)||String||Default catalog to use. Is also set by USE CATALOG [catalog_name]|
|sql.current-database||(None)||String||Default database to use. Is also set by USE [database_id]|
|sql.local-time-zone||Local device time zone||String||Specify the local time zone offset for Flink SQL. When converting to data types that don’t include a time zone (e.g. TIMESTAMP, TIME, or simply STRING), this time zone is used. The input of option is either a full name such as “America/Los_Angeles”, or a custom timezone id such as “GMT-08:00”.|
|sql.state-ttl||0 ms||Duration||Specifies a minimum time interval for how long idle state (i.e. state which was not updated), will be retained. State will never be cleared until it was idle for less than the minimum time, and will be cleared at some time after it was idle.|
This specifies the bounded mode for Kafka consumer. The valid enumerations are:
group-offsets: bounded by committed offsets of a specific consumer group. This is evaluated at the start of consumption from a given partition. latest-offset: bounded by latest offsets. This is evaluated at the start of consumption from a given partition. timestamp: bounded by a user-supplied timestamp.
If this config option is not set, the default is an unbounded table.
If timestamp is specified, config option sql.tables.scan.bounded.timestamp-millis is required to specify a specific bounded timestamp in milliseconds since January 1, 1970 00:00:00.000 GMT.
|sql.tables.scan.bounded.timestamp-millis||(None)||Long||End at the specified epoch timestamp (milliseconds) used in case of ‘timestamp’ bounded mode.|
|sql.tables.scan.idle-timeout||0 ms||Duration||When a source do not receive any elements for the timeout time, it will be marked as temporarily idle. This allows downstream tasks to advance their watermarks without the need to wait for watermarks from this source while it is idle. Default value is 0 ms, which means detecting source idleness is not enabled.|
This specifies the startup mode for Kafka consumer for this specific statement. The valid enumerations are:
group-offsets: start from comitted offsets of a specific consumer group. earliest-offset: start from the earliest offset possible. latest-offset: start from the latest offset. timestamp: start from user-supplied timestamp for each partition. If this config option is not set, the default is earliest-offset for this statement. This value can also be set on a table configuration instead of per statement.
If timestamp is specified, config option sql.tables.scan.startup.timestamp-millis is required to specify a specific startup timestamp in milliseconds since January 1, 1970 00:00:00.000 GMT.
|sql.tables.scan.startup.timestamp-millis||(None)||String||Start from the specified epoch timestamp (milliseconds) used in case of ‘timestamp’ startup mode.|