Connector parameters - Realtime Compute for Apache Flink

This topic describes the parameters for the Fluss connector.

How to configure parameters

Configure parameters during table creation

Fluss persists parameters defined during table creation as table metadata. These parameters apply to all read and write operations.

# Set table.log.ttl to 7 days and disable CRC checks during reads
CREATE TABLE `my-catalog`.`my_db`.`my_log_table` (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) WITH (
  'table.log.ttl' = '7d',
  'client.scanner.log.check-crc' = 'false'
);

Important

For a table that is already persisted in the Fluss metadata, that is, a registered table created by a CREATE TABLE DDL statement, you cannot modify the registered connector parameters, such as client.writer.batch-size and client.writer.buffer.memory-size. Such changes do not take effect even if you click Edit in the console.

Dynamic configuration with SQL hints

Fluss also lets you dynamically configure parameters using SQL Hints. These parameters are not persisted in the table metadata and affect only the current query.

Important

Dynamically configured parameters have higher priority over parameters persisted in table metadata.
Storage parameters do not support dynamic configuration using SQL Hints.

-- Set SQL hints for a read operation
SELECT * FROM `my-catalog`.`my_db`.`my_log_table` /*+ OPTIONS('client.scanner.log.check-crc' = 'false') */;

-- Set SQL hints for a write operation
INSERT INTO `my-catalog`.`my_db`.`my_pk_table_2` /*+ OPTIONS('sink.ignore-delete'='true') */ select * from `my-catalog`.`my_db`.`my_pk_table_1`;

Storage parameters

Parameter	Type	Default	Description
bucket.num	int	1	The number of buckets in the Fluss cluster.
bucket.key	String	None	Data will be distributed among the buckets based on the hash value of `bucket.key`. The `bucket.key` must be a subset of the primary key and must not include the partition key. If you specify multiple fields as the bucket.key, separate the fields with a comma `,` . If a table has a primary key and the `bucket.key` is not explicitly specified, the primary key (excluding the partition key) is used as the `bucket.key` by default. If a primary key is not defined for the table and `bucket.key` is not specified, the data will be randomly distributed among the buckets.
table.log.ttl	Duration	7d	The time-to-live (TTL) for log data, which controls the maximum retention period. After this period, Fluss deletes old log data to free up space. A value of -1 disables expiration, so logs are never deleted.
table.auto-partition.enabled	Boolean	false	If enabled, Fluss automatically creates partitions for the table.
table.auto-partition.time-unit	ENUM	DAY	The time granularity for automatic partition creation. Valid values are `YEAR`, `QUARTER`, `MONTH`, `DAY`, or `HOUR`. For example: If the value is `HOUR`, the format for automatically created partitions is `yyyyMMddHH`. If the value is `DAY`, the format is `yyyyMMdd`.
table.auto-partition.num-precreate	Integer	2	The number of future partitions to pre-create. For example, if the current date is 2024-11-11 and this is set to 3, Fluss pre-creates partitions for `20241111`, `20241112`, and `20241113`, skipping any that already exist. The default is 2. With a `time-unit` of `DAY`, this default creates partitions for the current and next day.
table.auto-partition.num-retention	Integer	7	The number of historical partitions to retain. For example, if the current date is 2024-11-11, the time unit is `DAY`, and this value is 3, Fluss retains partitions for `20241108`, `20241109`, and `20241110`. Fluss deletes partitions older than `20241108`. The default is 7, meaning Fluss retains the 7 most recent partitions.
table.auto-partition.time-zone	String	System time zone	The time zone for auto-partitioning. By default, it uses the system time zone.
table.replication.factor	Integer	(None)	The replication factor for log tables. If not set, Fluss uses the cluster's default replication factor, `default.replication.factor`. The value must be a positive integer and cannot be greater than the number of TabletServers in the Fluss cluster. If the value exceeds the number of TabletServers, table creation fails.
table.log.format	Enum	ARROW	The storage format for logs. Supported formats are `ARROW` (default) and `INDEXED`.
table.log.arrow.compression.type	Enum	ZSTD	If the log format is set to ARROW, this parameter specifies the compression type for log data. Supported types are NONE, LZ4_FRAME, and ZSTD. The default value is ZSTD.
table.log.arrow.compression.zstd.level	Integer	3	If the log format is set to ARROW and the compression type is ZSTD, this parameter specifies the compression level. The valid range is from 1 to 22. The default value is 3.
table.statistics.columns	String	(None)	Specifies a comma-separated list of column names for which to collect statistics, such as `'temperature,location'`. Set this parameter to `*` to collect statistics for all supported column types. Column statistics are used for predicate pushdown optimization and are applicable only to log tables that use the Arrow log format. Supported data types include: BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, FLOAT, DOUBLE, DECIMAL, STRING, CHAR, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ. Unsupported types (BYTES, BINARY, ARRAY, MAP, ROW) are automatically excluded. For more information, see Predicate Pushdown.
table.kv.format	Enum	COMPACTED	The storage format for KV data. Supported formats are `COMPACTED` (default) and `INDEXED`.
table.log.tiered.local-segments	Integer	2	When tiered log storage is enabled, this parameter specifies the number of log segments to retain locally for each table. The value must be greater than 0. The default is 2.
table.datalake.enabled	Boolean	false	Specifies whether to enable lakehouse storage for the table. This is disabled by default. If enabled and the lakehouse tiering service is active, Fluss automatically tiers and compresses table data into lakehouse storage.
table.datalake.format	Enum	(None)	Specifies the data lake format that the table uses for layered lakehouse storage. Currently, only `paimon` is supported. The core functions of this parameter are: Ensures consistency in key encoding and bucketing strategy between Fluss and the external lakehouse system. Enables unified queries (Union Read) across Fluss and data lake tables. Configuration Behavior Highest priority: If `table.datalake.format` is explicitly set during table creation, that value is used. Default value: If not set, the cluster-level `datalake.format` configuration in Fluss is used. Dynamic activation: You can pre-configure `table.datalake.format` while `table.datalake.enabled` is `false`. You can then activate the data lake functionality by enabling `table.datalake.enabled` without recreating the table.
table.merge-engine	Enum	(None)	Defines the merge engine for primary key tables. The default strategy is `last_row`, which keeps the most recent row. Other supported strategies are: `first_row`: Keeps the first row for a given primary key. `versioned`: Keeps the row with the highest version number for a given primary key.
table.merge-engine.versioned.ver-column	String	(None)	If the merge engine is set to `versioned`, this parameter is required and must specify the name of the version column.
table.log.local-ttl	Duration	(None)	The time to live for local log segments. This option controls the maximum time local log segments are retained before they become eligible for deletion. When remote log tiering is enabled, an expired local segment is deleted only after it has been copied to remote storage. Set the value to `0ms` to disable TTL-based deletion. If not configured, the value inherits `table.log.ttl`. When both values are positive, this value must be less than or equal to `table.log.ttl`.
table.kv.format-version	Integer	(None)	The version of the KV format. If not configured, the coordinator sets it automatically during table creation. The datalake encoding and bucketing strategies described below take effect only when `datalake.format` is configured at the cluster level. Version 1: Tables created before `table.kv.format-version` was introduced are treated as version 1. Both the primary key and the bucket key use the datalake encoder, such as Paimon or Iceberg. Some datalake encoders do not guarantee that the encoded bucket key bytes are a prefix of the encoded primary key bytes, so prefix lookup may not work properly. Version 2 (current): For new tables whose bucket key differs from the primary key, the primary key uses the Fluss default encoder, which ensures proper prefix lookup support. When the bucket key equals the primary key, the datalake encoder is still used to optimize lake writes. The bucket key encoding always uses the datalake encoder to align with the datalake bucket calculation.
table.changelog.image	Enum	FULL	The changelog image mode for primary key tables. The semantics follow similar settings in database systems such as `binlog_row_image` in MySQL and `replica identity` in PostgreSQL. This option applies only to primary key tables. `FULL` (default): Produces both UPDATE_BEFORE and UPDATE_AFTER records for update operations. This captures complete information about updates and allows you to track previous values. `WAL`: Does not produce UPDATE_BEFORE records. Only INSERT, UPDATE_AFTER, and DELETE records (when allowed) are emitted. When WAL mode is enabled, the default merge engine is used, updates are full row updates rather than partial updates, and there is no auto-increment column, an optimization skips looking up old values and converts INSERT operations to UPDATE_AFTER events. This mode reduces storage and transmission costs but loses the ability to track previous values.
table.delete.behavior	Enum	ALLOW	Controls the behavior of delete operations on primary key tables. This option provides system-level guarantees for downstream pipelines, such as Flink Delta Join, that must not receive delete events in the table changelog. It applies only to primary key tables. `ALLOW` (default for the default merge engine): Allows normal delete operations. `IGNORE`: Silently ignores delete requests without errors. `DISABLE`: Rejects delete requests and throws explicit errors. For tables with the `first_row`, `versioned`, or `aggregation` merge engine, this option is automatically set to `IGNORE` and cannot be overridden. For the `aggregation` merge engine, when set to `ALLOW`, delete operations remove the entire record.
table.auto-inc.batch-size	Long	100000	The batch size of auto-increment IDs fetched from the distributed counter each time. This value determines the length of the locally cached ID segment. A larger batch size may cause significant auto-increment ID gaps, especially when unused cached ID segments are discarded due to TabletServer restarts or abnormal terminations. A smaller batch size increases the frequency of ID fetch requests to the distributed counter, which introduces extra network overhead and reduces write throughput.
table.datalake.auto-compaction	Boolean	false	Whether compaction is triggered automatically when the tiering service writes to the datalake. Disabled by default.
table.datalake.auto-expire-snapshot	Boolean	false	Whether snapshot expiration is triggered automatically when the tiering service commits to the datalake. Disabled by default.
table.datalake.historical-partition.enabled	Boolean	false	Whether to enable historical partition lookup for the table. When enabled, the coordinator creates and retains a system partition for routing lookups of expired partitions to lake storage. Currently, this option only supports auto-partitioned Paimon primary key tables with a single partition key. After you change this option, restart the existing lookup jobs that need to look up historical partition data so that their clients load the updated table configuration.

Read parameters

Parameter	Type	Default	Description
scan.startup.mode	Enum	full	The connector's startup mode, which specifies where to start consuming data. Supported modes are `full` (default), `earliest`, `latest`, and `timestamp`. For details, see Read Data.
scan.startup.timestamp	Long	(None)	The timestamp to start reading from. This option applies only when `scan.startup.mode` is `timestamp`. Specify the value as either milliseconds since the epoch (e.g., '1678883047356') or a 'yyyy-MM-dd HH:mm:ss' string (e.g., '2023-12-09 23:09:12'). Note This parameter does not support Flink built-in time functions, such as CURRENT_TIMESTAMP, to generate a time value dynamically. Connector parameters in the WITH clause accept only fixed literals and cannot resolve SQL function expressions. To pass a dynamic time, inject a fixed timestamp or millisecond value from outside the job by using a variable.
scan.partition.discovery.interval	Duration	10s	The interval for the Fluss source to check for new partitions. The source dynamically adds new partitions to the read plan and unsubscribes from deleted ones. A negative value disables automatic partition discovery.
client.scanner.log.check-crc	Boolean	true	Specifies whether to perform a CRC32 check on messages that are read. This check verifies that messages are not corrupted in transit or on disk, improving data reliability. Although this adds minor computational overhead, we recommend enabling it in production to ensure data integrity.
client.scanner.log.max-poll-records	Integer	500	The maximum number of records returned by a single `poll()` call from the LogScanner. Note that this setting does not affect the underlying fetch behavior. The LogScanner caches records from each fetch request and returns them gradually with each poll.
client.scanner.log.fetch.max-bytes	MemorySize	16mb	The maximum data size per fetch request. If the first record batch in the first non-empty bucket exceeds this limit, the client still returns it to ensure progress. This value is therefore a soft limit, not an absolute one.
client.scanner.log.fetch.max-bytes-for-bucket	MemorySize	1mb	The maximum amount of data returned per table bucket in a single fetch request. Records are fetched in batches, and the maximum byte size is configured by this option.
client.scanner.log.fetch.min-bytes	MemorySize	1b	The minimum amount of data in bytes the client expects from the server in response to each log fetch request. If the data size is insufficient, the client waits for a maximum of `client.scanner.log.fetch-wait-max-time` before returning.
client.scanner.log.fetch.wait-max-time	Duration	500ms	The maximum time the client waits for a response from the server for each log fetch request.
client.scanner.io.tmpdir	String	System.getProperty("java.io.tmpdir") + "/fluss"	The local directory where the client temporarily stores data files, such as KV snapshots and log segment files, for reading.
client.scanner.remote-log.prefetch-num	Integer	4	The number of remote log segments downloaded from remote storage by the LogScanner. These log segments are saved in local temporary files. The default value is 4.
client.remote-file.download-thread-num	Integer	3	The number of threads the client uses to download remote files.
scan.kv.snapshot.lease.id	String	UUID	The lease ID used to protect acquired KV snapshots from deletion. If specified, the snapshots are retained until the consumer finishes processing all of them or the lease duration expires. If not explicitly provided, this value is a randomly generated UUID string.
scan.kv.snapshot.lease.duration	Duration	1day	The time period to wait before expiring the KV snapshot lease, which prevents snapshots from blocking deletion.

Write parameters

Parameter	Type	Default	Description
sink.ignore-delete	Boolean	false	When `true`, the sink ignores `DELETE` and `UPDATE_BEFORE` changelog events.
sink.bucket-shuffle	Boolean	true	Specifies whether to shuffle data by bucket ID before writing to the sink. Shuffling to group records with the same bucket ID into the same task improves client-side efficiency and reduces resource usage. For log tables, this requires a `bucket.key`. This option is enabled by default for primary key tables.
client.writer.buffer.memory-size	MemorySize	64mb	The total number of bytes that the writer can use to buffer internal rows.
client.writer.batch-size	MemorySize	2mb	The size of a batch for a single bucket. The writer groups records for the same bucket into batches, which improves both client and server performance.
client.writer.buffer.wait-timeout	Duration	2^(63)-1ns	Defines how long the writer blocks when waiting for a segment to become available.
client.writer.batch-timeout	Duration	100ms	To improve throughput, the writer can add a small artificial delay to wait for more records to arrive before sending a batch. This is similar to Nagle's algorithm in TCP and sets an upper bound on send latency. A batch is sent immediately when it reaches `client.writer.batch-size`, but if it's smaller, the writer waits up to this timeout for more records.
client.writer.bucket.no-key-assigner	Enum	STICKY	Specifies the bucket assignment strategy for keyless tables. For tables with a bucket key or primary key, the system selects a bucket based on the key's hash value. For keyless tables, you can select the bucket assigner with this parameter. The following strategies are supported: `STICKY`: Assigns a new bucket ID only when the record batch changes. Otherwise, it uses the same bucket ID as the previous record. `ROUND_ROBIN`: Assigns bucket IDs to input rows in a round-robin fashion.
client.writer.acks	String	all	Valid values: `-1` (or `all`): The leader waits for all in-sync replicas (ISRs) to acknowledge the record. This guarantees no data loss as long as at least one in-sync replica remains alive. `1`: The leader acknowledges immediately after writing the record to the local log, without waiting for replicas. If the leader fails before replication completes, data may be lost. `0`: The writer does not wait for any acknowledgment from the server. The writer cannot guarantee the server received the record. Network errors or server failures may cause data loss. This is suitable for high-throughput scenarios where some data loss is acceptable. Note In production environments, use `all` to ensure high availability and prevent data loss.
client.writer.request-max-size	MemorySize	10mb	The maximum size of a request in bytes. This setting limits the number of record batches the writer can send in a single request to avoid sending overly large requests.
client.writer.retries	Integer	Integer.MAX_VALUE	Setting a value greater than zero causes the client to resend any record that fails with a potentially transient error.
client.writer.enable-idempotence	Boolean	true	Specifies whether to enable idempotence, which is enabled by default. When enabled, the writer ensures exactly-once semantics, which prevents duplicates and preserves ordering. If disabled, client retries due to server failures can result in duplicate messages.
client.writer.max-inflight-requests-per-bucket	Integer	5	The maximum number of unacknowledged requests per bucket. This setting applies only when `client.writer.enable-idempotence` is set to `true`. If the number of in-flight requests for a bucket reaches this limit, the writer blocks until a pending request is acknowledged before sending another.
client.writer.buffer.page-size	MemorySize	128kb	The size of every page in the memory buffers specified by `client.writer.buffer.memory-size`.
client.writer.buffer.per-request-memory-size	MemorySize	16mb	The minimum number of bytes allocated by the writer, rounded down to the closest multiple of `client.writer.buffer.page-size`. The value must be greater than or equal to `client.writer.buffer.page-size`. Allocating memory in batches improves CPU cache friendliness because the segments are contiguous.
client.writer.dynamic-batch-size.enabled	Boolean	true	Controls whether the client writer dynamically adjusts the batch size based on the actual write throughput. Enabled by default. With dynamic batch sizing enabled, the writer adapts memory allocation per batch according to historical write sizes for the target table or partition, which improves memory utilization and performance under varying throughput conditions. The dynamic batch size is bounded: it does not exceed `client.writer.batch-size` or fall below `client.writer.buffer.page-size`. When disabled, the writer uses the fixed batch size `client.writer.batch-size` for all batches. This may lead to frequent memory waits and suboptimal write performance if the incoming data rate is inconsistent across partitions.
sink.distribution-mode	Enum	AUTO	The distribution mode for shuffling data to the sink. `AUTO` (default): Automatically chooses the best mode based on the table type. It uses BUCKET for primary key tables and for log tables with bucket keys to maximize throughput, and uses NONE for log tables without bucket keys. `NONE`: Uses the default Flink shuffle strategy. It typically uses FORWARD when the sink parallelism matches the upstream parallelism, and ROUND_ROBIN when the parallelisms differ. `BUCKET`: Shuffles data by bucket ID before writing to the sink, so data with the same bucket ID is processed by the same task. This improves client processing efficiency and reduces resource consumption. This mode is particularly recommended for primary key tables because it can significantly improve throughput. For log tables, bucket shuffle takes effect only when `bucket.key` is defined. When the sink parallelism exceeds the number of buckets, some sink tasks may remain idle. `PARTITION_DYNAMIC`: Dynamically adjusts the shuffle strategy based on partition key traffic patterns. This mode monitors data distribution and adjusts shuffle behavior to balance the load. It is supported only for partitioned log tables, not for primary key tables. Use this mode when data is highly skewed across partitions or when there are many partitions. This mode has overhead costs, including data statistics collection and additional shuffle operations.

Dimension table parameters

Parameter	Type	Default	Description
lookup.async	Boolean	true	Specifies whether to use asynchronous lookup. Asynchronous lookup provides better throughput than synchronous lookup.
`lookup.insert-if-not-exists`	Boolean	false	When `true`, automatically inserts a new row if a lookup key is not found. Fluss inserts a new row using the lookup key as the primary key and then returns that row. This feature applies only to primary key lookups and is not supported for prefix lookups.
lookup.cache	Enum	NONE	The caching policy for this lookup table. Supported values are `NONE` and `PARTIAL`.
lookup.max-retries	Integer	3	The maximum number of retries allowed when a lookup operation fails.
lookup.partial-cache.expire-after-access	Duration	(None)	The expiration time for a cache entry after it has been accessed.
lookup.partial-cache.expire-after-write	Duration	(None)	The expiration time for a cache entry after it has been written.
lookup.partial-cache.cache-missing-key	Boolean	true	If `true`, caches the fact that a key is missing (by storing a null value) to avoid repeated lookups for non-existent keys.
lookup.partial-cache.max-rows	Long	(None)	The maximum number of rows to store in the cache.
client.lookup.queue-size	Integer	25600	The maximum number of pending lookup operations.
client.lookup.max-batch-size	Integer	128	The maximum number of lookups to batch into a single request.
client.lookup.max-inflight-requests	Integer	128	The maximum number of unacknowledged lookup requests for a lookup operation.
client.lookup.batch-timeout	Duration	100ms	The maximum time to wait for a lookup batch to fill. If this timeout is reached, the lookup batch is closed and sent.
client.lookup.max-retries	Integer	Integer.MAX_VALUE	The number of times the client resends a lookup request that fails with a potentially transient error. A value greater than zero enables resending.

Other parameters

Parameter	Type	Default	Description
bootstrap.servers	List	(None)	Specify the initial list of nodes used to connect to the Fluss cluster, in the format `host1:port1,host2:port2,...`. The client uses this list to discover the full cluster topology, so it does not need to include all servers. We recommend configuring multiple nodes for high availability to avoid a single point of failure.
client.id	String	""	An ID string to pass to the server with each request. Using a meaningful ID (e.g., `order-service-prod`) helps identify the request source in server-side logs.
client.connect-timeout	Duration	120s	The connection timeout for the Netty client.
client.request-timeout	Duration	30s	The timeout for completing a request. When `client.writer.acks` is set to `all` (-1), this specifies the maximum time the client waits for a write to be fully acknowledged. Default: 30s.
client.filesystem.security.token.renewal.backoff	Duration	1h	The time interval to wait before retrying to fetch a new security token for the file system after a fetch failure.
client.filesystem.security.token.renewal.time-ratio	Double	0.75	A ratio of the token's lifetime that determines when to renew it. For example, a ratio of 0.75 means renewal is attempted after 75% of the token's validity period has passed.
client.metrics.enabled	Boolean	false	Enables client metrics. When enabled, the client collects and reports metrics through a JMX metrics reporter.