Fluss Pipeline Connector

The Fluss Pipeline connector can be used as a Data Source or Data Sink of the pipeline. It reads from or writes data to Fluss. This document describes how to set up both roles.

What can the connector do?

• Create table automatically if not exist
• Data synchronization
• Schema change synchronization (lenient mode)
• Dynamic source table subscriptions

Fluss Source

The following is the minimal configuration for reading dynamically discovered Fluss tables:

source:
  type: fluss
  bootstrap.servers: localhost:9123
  table.discoverer.type: fluss-default
  table.discoverer.pattern: 'inventory\..*'
  scan.discovery.interval: 10 s
  scan.startup.mode: earliest

table.discoverer.type selects the source table discoverer. fluss-default matches fully qualified table names with table.discoverer.pattern; configure another discoverer's required table.discoverer.* options when selecting it instead.

How to create Pipeline

The pipeline for reading data from MySQL and sink to Fluss can be defined as follows:

source:
   type: mysql
   name: MySQL Source
   hostname: 127.0.0.1
   port: 3306
   username: admin
   password: pass
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404

sink:
  type: fluss
  name: Fluss Sink
  bootstrap.servers: localhost:9123
  # Security-related properties for the Fluss client
  properties.client.security.protocol: sasl
  properties.client.security.sasl.mechanism: PLAIN
  properties.client.security.sasl.username: developer
  properties.client.security.sasl.password: developer-pass

pipeline:
  name: MySQL to Fluss Pipeline
  parallelism: 2
  schema.change.behavior: LENIENT

Pipeline Connector Options

Option	Required	Default	Type	Description
type	required	(none)	String	Specify what connector to use, here should be 'fluss'.
name	optional	(none)	String	The name of the sink.
bootstrap.servers	required	(none)	String	The bootstrap servers for the Fluss sink connection.
sink.partitioning.strategy	optional	DEFAULT	String	The partitioning strategy for DataChangeEvent routing. Available values are DEFAULT and FORWARD. DEFAULT hashes primary key tables by primary keys and randomly distributes log table events across downstream subtasks for balanced load. FORWARD routes data events to downstream subtasks with the same indices as upstream and is intended for Fluss-to-Fluss data synchronization. The upstream data distribution must match the Fluss table bucket distribution; otherwise, data correctness issues may occur.
bucket.key	optional	(none)	String	Specific the distribution policy of each Fluss table.Tables are separated by ';', and bucket keys are separated by ','. Format: database1.table1:key1,key2;database1.table2:key3. Data will be distributed to each bucket according to the hash value of bucket-key (It must be a subset of the primary keys excluding partition keys of the primary key table). If the table has a primary key and a bucket key is not specified, the bucket key will be used as primary key(excluding the partition key). If the table has no primary key and the bucket key is not specified, the data will be distributed to each bucket randomly.
bucket.num	optional	(none)	String	The number of buckets of each Fluss table.Tables are separated by ';'.Format: database1.table1:4;database1.table2:8.
properties.table.*	optional	(none)	String	Pass options of Fluss table to pipeline. See Fluss table options.
properties.client.*	optional	(none)	String	Pass options of Fluss client to pipeline. See Fluss client options.

Table options in transform rules

Use table-options in a transform rule to configure Fluss table properties for the matched table. Use Fluss property names such as table.replication.factor, without the properties. prefix used in the sink configuration. These options override matching sink.properties.table.* options; other sink table properties still apply.

transform:
  - source-table: inventory.products
    projection: "*"
    table-options: table.replication.factor=2,table.log.ttl=7d

sink:
  type: fluss
  bootstrap.servers: localhost:9123
  properties.table.replication.factor: 3

In this example, the Fluss table created for inventory.products has a replication factor of 2 and a log retention time of 7 days. Other tables use a replication factor of 3. These options apply when creating tables and do not update the properties of existing tables. Continue to configure bucket keys and bucket counts with the sink options bucket.key and bucket.num.

Usage Notes

• Support Fluss primary key table and log table.

Dynamic source subscriptions

When Fluss is used as a source with a table discoverer, each successful discovery result is the authoritative complete subscription set. Set a positive scan.discovery.interval to enable periodic updates. An empty result unsubscribes every discovered table; a discovery failure does not change the current subscription and fails the job.

Unsubscribing a table only stops and cleans up its source-side readers. It does not delete the Fluss table or change sink behavior. A restored reader waits for a fresh subscription snapshot before it opens restored splits, so a table that remains unsubscribed cannot emit from restored splits after recovery. If a table is subscribed again, it is treated as a new table and uses the configured scan.startup.mode.

Removal is coordinated with checkpoint state: a failure restores source splits and pending removal tombstones from the latest completed checkpoint; subscription is refreshed by discovery. A removal and re-addition that both occur between the same two completed checkpoints may be rolled back as though the removal had not occurred. Restoring a checkpoint from before a removal tombstone while the table is currently re-subscribed does not promise a fresh table lifecycle. For primary key tables, snapshot leases are not released early during removal; their existing expiry and close handling remain in effect.

• For creating table automatically

  • There is no partition key
  • The number of buckets is controlled by bucket.num
  • The distribution keys are controlled by option bucket.key. For primary key table and a bucket key is not specified, the bucket key will be used as primary key(excluding the partition key). For log table has no primary key and the bucket key is not specified, the data will be distributed to each bucket randomly.

• Supports schema change synchronization in lenient mode with schema.change.behavior: lenient. The following schema change events are supported:

  • Add column — new columns are appended to the Fluss table.
  • Drop column — the column is not physically removed in lenient mode. The drop operation is ignored, and subsequent writes will set the column value to null.
  • Rename column — in lenient mode, this is translated into an add-new-column + alter-old-column-type-to-nullable sequence.
  • Alter column type — not supported.

  To enable schema change synchronization, configure the pipeline with schema.change.behavior: lenient. If you want to ignore all schema changes, use schema.change.behavior: IGNORE.

• For data synchronization, the pipeline connector uses Fluss Java Client to write data to Fluss.

Data Type Mapping

Flink CDC type	Fluss type	Note
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(p, s)	DECIMAL(p, s)
BOOLEAN	BOOLEAN
DATE	DATE
TIME	TIME
TIMESTAMP	TIMESTAMP
TIMESTAMP_LTZ	TIMESTAMP_LTZ
CHAR(n)	CHAR(n)
VARCHAR(n)	STRING
BINARY(n)	BINARY(n)
VARBINARY(N)	BYTES
ARRAY	ARRAY	Element type is mapped recursively.
MAP	MAP	Key and value types are mapped recursively.
ROW	ROW	Field types are mapped recursively.