Reading and Writing Fluss

Applies toSelf-Managed v3

9 min read

On this page

This guide provides configuration, deployment, and troubleshooting instructions for operators and engineers connecting Apache Flink SQL or Java SDK clients to a Fluss cluster. It details how to set up cluster dependencies, manage secure access, and perform read or write operations against real-time Fluss data tables, including specific configurations for Ververica Platform 3.

Overview

A Fluss client requires 3 pieces of configuration to connect, whether you use a Flink job with the Fluss Flink connector or a JVM application using the Fluss Java client directly:

The cluster's bootstrap address.
The required client-side dependencies (Fluss connector or client JARs, plus optional filesystem and lake plugins).
SASL credentials, when the cluster has authentication enabled.

This manual covers client connections and reads or writes of fresh data. This refers to the real-time rows resident in Fluss, as opposed to historical data that tiering has moved into the lakehouse. For more information on this architecture, see the Fluss streaming-lakehouse overview for the real-time and historical model.

If you are running Flink jobs that touch the historical (Iceberg lake) tier—such as the tiering service, union reads on lake-tiered tables, and submission runners including Ververica Platform 3—see Running Lakehouse (Iceberg) Jobs against Fluss. For Flink SQL syntax, table DDL, and the Fluss Java client API (creating tables, writing rows, and scanning), refer to the Apache Fluss documentation. Running fresh-data jobs on Ververica Platform 3, including console SQL deployments and Debug reads, is covered in Running Jobs on Ververica Platform 3 below.

Bootstrap Address

The bootstrap address points at the coordinator's headless service. For a cluster deployed in namespace <NAMESPACE>, use the following address format:

TEXT

1coordinator-server-0.coordinator-server-hs.<NAMESPACE>.svc.cluster.local:9124

For external listener-exposure options, see Deploying Fluss on Kubernetes. The coordinator forwards client connections to the appropriate tablet server; clients only need the coordinator address.

Install Clients Dependencies

Client-side JARs are delivered by the fluss-setup installer's client subcommand. For the inventory of available setups and the install commands (such as --flink-home and --output-dir semantics, init-container patterns, or pre-baking into a custom Flink image), see Installing Fluss .docx> Setups with a client target.

The specific setups a client needs depend on whether it reads data, writes data, or performs both operations.

For writing

Writers send rows to the Fluss server over RPC, and the server handles all object-storage I/O. Clients therefore need only the Fluss Flink connector. You can install the flink-connector setup, or use any one of the fs-* or lake-iceberg-* setups, which all bundle the connector, if the same client also reads data.

Ensure that your Ververica Platform configuration includes these dependencies before launching your writer jobs.

For reading

Read clients additionally need the filesystem setup matching the Fluss server's remote-storage backend. For example, use fs-s3 for an S3- or NooBaa-backed cluster, or fs-azure for an ABS-backed cluster. This requirement exists because the Fluss server hands out object-storage paths, such as KV snapshots and log segments, and clients fetch the bytes themselves. This behavior applies to every read operation, not only union reads on tiered tables.

Because fs-s3 and fs-azure already bundle the Fluss Flink connector, you do not need a separate flink-connector installation when you install one of these filesystem plugins. Ensure that Ververica Platform has the correct plugin path configured so your read clients can resolve these dependencies at runtime.

Configuring the Flink Catalog

SQL

1CREATE CATALOG fluss WITH (
2    'type'              = 'fluss',
3    'bootstrap.servers' = '<COORDINATOR_BOOTSTRAP>'
4);

When the cluster has SASL enabled, the client presents its credentials through the catalog's client.security.* options:

SQL

1CREATE CATALOG fluss WITH (
2    'type'                           = 'fluss',
3    'bootstrap.servers'              = '<COORDINATOR_BOOTSTRAP>',
4    'client.security.protocol'       = 'sasl',
5    'client.security.sasl.mechanism' = 'PLAIN',
6    'client.security.sasl.username'  = '<FLUSS_USERNAME>',
7    'client.security.sasl.password'  = '<FLUSS_PASSWORD>'
8);

Enabling SASL on the server side is covered in Securing Fluss.docx. The client.security.* keys shown above represent the client-side configuration. For the full set of catalog options regarding security and client tuning, refer to the upstream Fluss Flink connector documentation.

For Flink jobs that read or write Iceberg-tiered tables through union reads or the tiering service, see Running Lakehouse (Iceberg) Jobs against Fluss. Ensure that Ververica Platform has access to these configuration options when establishing authenticated connections.

Configuring the Java SDK

A JVM application creates a Fluss connection by handing a Configuration to ConnectionFactory.createConnection:

TEXT

1Configuration conf = new Configuration();
2conf.setString("bootstrap.servers", "<COORDINATOR_BOOTSTRAP>");
3Connection connection = ConnectionFactory.createConnection(conf);

When the cluster has SASL enabled, set the same client.security.* keys on the Configuration:

TEXT

1conf.setString("client.security.protocol",       "sasl");
2conf.setString("client.security.sasl.mechanism",  "PLAIN");
3conf.setString("client.security.sasl.username",   "<FLUSS_USERNAME>");
4conf.setString("client.security.sasl.password",   "<FLUSS_PASSWORD>");

Enabling SASL on the server side is covered in Securing Fluss. For the full set of client.* keys and the Java client API, refer to the upstream Fluss documentation and configuration reference.

Reading from a Fluss Cluster

The Java SDK reads KV snapshots and log segments from object storage through the Fluss filesystem plugin (S3FileSystemPlugin for an S3- or NooBaa-backed cluster, or AzureFileSystemPlugin for an ABS-backed cluster). Every SDK read operation therefore needs the matching fs-* setup on its classpath; see Reading and Writing Fluss.docx. This requirement is independent of lake access, meaning the same code path runs whether or not the source table has tiering enabled.

On AWS S3 and ABS, no further client-side configuration is required. The Fluss server fetches cloud-vended session credentials (STS on AWS, or the equivalent Azure mechanism) on behalf of the client and forwards them through the delegation-token RPC. The SDK uses those tokens for the KV-snapshot fetch and never needs static keys.

On NooBaa or another STS-less S3-compatible store

S3-compatible stores without an STS endpoint (such as NooBaa, MinIO, Ceph RGW, or ODF) cannot use the delegation-token flow, because the server has no STS endpoint to call. The supported configuration on these backends is to pass static credentials directly on the Fluss configuration:

TEXT

1Configuration conf = new Configuration();
2conf.setString("bootstrap.servers",              "<COORDINATOR_BOOTSTRAP>");
3conf.setString("client.fs.s3.access.key",        "<ACCESS_KEY_ID>");
4conf.setString("client.fs.s3.secret.key",        "<SECRET_ACCESS_KEY>");
5conf.setString("client.fs.s3.endpoint",          "<S3_ENDPOINT>");
6conf.setString("client.fs.s3.region",            "<S3_REGION>");
7conf.setString("client.fs.s3.path-style-access", "true");
8Connection connection = ConnectionFactory.createConnection(conf);

The FlussConnection class strips the client.fs. prefix and forwards the keys to FileSystem.initialize.

The Flink SQL equivalent passes the same keys as a per-query OPTIONS hint. The CREATE CATALOG fluss WITH (...) statement cannot carry them, because it forwards only client.security.* properties. Therefore, the per-query hint is the supported entry point:

SQL

1SELECT * FROM <table>
2  /*+ OPTIONS('client.fs.s3.access.key'        = '<ACCESS_KEY_ID>',
3              'client.fs.s3.secret.key'        = '<SECRET_ACCESS_KEY>',
4              'client.fs.s3.endpoint'          = '<S3_ENDPOINT>',
5              'client.fs.s3.region'            = '<S3_REGION>',
6              'client.fs.s3.path-style-access' = 'true') */;

Writing to a Fluss cluster

Writers (including Flink INSERT INTO statements and SDK UpsertWriter or AppendWriter instances) send rows to the Fluss server over RPC. The server handles all object-storage I/O, so writers do not require a filesystem setup on the client or static credentials. This rule applies regardless of whether the server's remote-storage backend is AWS S3, ABS, or a NooBaa-class service.

The bootstrap address and, when enabled, SASL credentials are the only client-side configurations you need to provide. Ververica Platform manages the underlying storage communication automatically once the writer initiates the RPC connection.

Running Jobs on Ververica Platform 3 (VVP3)

Ververica Platform 3 is a co-equal runner for Flink jobs against Fluss, alongside the Flink CLI or SQL client, the Apache Flink Kubernetes Operator, and standard session or application clusters. It runs jobs in 2 first-class ways:

Console (UI): Interactive SQL in the SQL Editor, SQL and JAR deployments, and Debug queries on a session cluster.
Ververica Platform 3 Kubernetes Operator: A VvpDeployment custom resource applied with kubectl. The platform remains the source of truth, as the operator syncs your Kubernetes-declared intent to it. You must change an operator-managed deployment only through the operator itself. Note that this is the Ververica Platform 3 operator, which is a completely different product from the Apache Flink Kubernetes Operator.

This section assumes an existing Ververica Platform 3 installation and workspace. For installation and onboarding details, see the Ververica Platform 3 getting-started guide. For lake jobs, such as the tiering service and reads of $lake tables, see Running Lakehouse (Iceberg) Jobs against Fluss.docx> Running Jobs on Ververica Platform 3.

Dependencies

Ververica Platform 3 deployments declare their JARs as Additional Dependencies rather than installing Fluss setups into $FLINK_HOME/lib.

To configure these dependencies, perform the following steps:

Produce the JARs your job needs with the fluss-setup installer. You need the Fluss Flink connector for any deployment, plus the filesystem and lake plugins for lake jobs. For details, see Reading and Writing Fluss.docxand Running Lakehouse (Iceberg) Jobs against Fluss.docx > Flink-Classpath Plugins.
Upload each generated JAR file to your workspace Artifacts.
Reference the uploaded artifacts from your deployment. Use the artifact as the JAR URI for a JAR deployment, or add it as an Additional Dependency for any other deployment type.

Note

The Fluss release artifacts are published to the Ververica Maven repository (resolved automatically by the fluss-setup installer). See Obtaining Registry Access.

Registering the Fluss Catalog

In the SQL Editor, register the Fluss catalog once per workspace. Subsequent deployments and editor sessions can reference it without re-creating it:

SQL

1CREATE CATALOG fluss WITH (
2    'type'                           = 'fluss',
3    'bootstrap.servers'              = '<COORDINATOR_BOOTSTRAP>',
4    'client.security.protocol'       = 'sasl',
5    'client.security.sasl.mechanism' = 'PLAIN',
6    'client.security.sasl.username'  = '<FLUSS_SASL_USERNAME>',
7    'client.security.sasl.password'  = '${secret_values.fluss_sasl_password}'
8);

For the correct bootstrap address format, see Bootstrap Address. You can omit the client.security.* lines when the cluster has no SASL authentication enabled. For more details on authentication configurations, see Securing Fluss.

Important

Do not inline credentials in SQL. Store the SASL password as a VVP secret value and reference it as ${secret_values.<name>}, as shown above. The same substitution works in deployment SQL and in JAR main arguments. See Manage Secret Values..

Writing - SQL Deployment

A continuous writer runs as a SQL Deployment (navigate to Deployments > Create Deployment > SQL Deployment).

Each deployment must contain exactly one INSERT INTO statement. Ververica Platform 3 runs jobs in Application Mode, which permits only a single sink per deployment. If you need to write data to one target table from several different streams, you must join or union those multiple sources first within your query logic before sending the output to the sink.

SQL

1CREATE TEMPORARY TABLE source WITH (
2    'connector' = 'faker'
3    -- field expressions ...
4) LIKE fluss.app.pk_table (EXCLUDING ALL);
5INSERT INTO fluss.app.pk_table SELECT * FROM source;

Click Save → Deploy, then Start from the Deployments page. The deployment moves through STARTING → RUNNING and writes continuously until you stop it.

Add them to the catalog WITH

Reading - Debug on a Session Cluster

Bounded reads run from the SQL Editor with Debug, which executes on a Session Cluster (create one first under Session Clusters). Batch reads require a LIMIT (capped at 2048 in the current engine version):

SQL

1-- Bounded full-snapshot read: 'scan.startup.mode' = 'full' returns both
2-- fresh in-Fluss rows and tiered remote segments.
3SELECT * FROM fluss.app.pk_table /*+ OPTIONS('scan.startup.mode' = 'full') */
4LIMIT 2048;

Results appear in the editor's Results panel.

Troubleshooting

Connection refused or DNS-resolution failures

If your client cannot reach the coordinator's headless service, check the following configuration and network details to troubleshoot the connection:

Bootstrap Host Check: Verify that your bootstrap host accurately targets coordinator-server-0.coordinator-server-hs.<NAMESPACE>.svc.cluster.local.
Network Resolution: Ensure that the client application is running on a network capable of resolving internal Kubernetes services inside the <NAMESPACE> namespace (this usually requires the client to be inside the same Kubernetes cluster).
Pod Health: Confirm that the Fluss coordinator pod is actively Running and marked as Ready (1/1).
Firewalls & Network Policies: Double-check that your Kubernetes NetworkPolicies or cloud security groups are not blocking port 9124 between your client's network space and the Fluss namespace.

ClassNotFoundException for FlussCatalogFactory or filesystem plugin classes

A required JAR is missing from your runtime classpath. Depending on which dependency is absent, you will encounter specific errors:

Missing the Fluss Flink connector: This causes a FlussCatalogFactory not found error inside Flink when attempting to instantiate or register the catalog.
Missing fs-s3 or fs-azure: This causes a "filesystem plugin classes not found" error. This issue will surface specifically when a read client attempts to fetch tiered segments from object storage.

Re-check Reading and Writing Fluss.

Authentication Failure (SASL)

The cluster requires SASL but the client did not declare the client.security.* keys. Add them to the catalog WITH (...) clause (see Reading and Writing Fluss) or to the Java Configuration (see Reading and Writing Fluss).

Reads return zero rows on an S3-compatible cluster without STS

For Java SDK clients, the S3FileSystemPlugin will fail to find any credentials if there is no core-site.xml file present on the application classpath. To resolve this, you must inject the client.fs.s3.* configuration keys directly into your client's Java Configuration builder. For setup examples, see On NooBaa or another STS-less S3-compatible store.

For Flink read jobs, which handle credentialing natively through a traditional core-site.xml configuration, refer instead to the dedicated debugging steps in Running Lakehouse (Iceberg) Jobs against Fluss.docx > Troubleshooting.