Connect to Confluent Cloud for Apache Flink with JDBC

The Confluent Java Database Connectivity (JDBC) driver connects business intelligence (BI) tools, SQL clients, and Java applications to Confluent Cloud for Apache Flink®, so you can run Flink SQL queries from the tools you already use.

JDBC clients, such as DBeaver, DataGrip, and the sqlline command-line shell, can browse catalogs, databases, tables, and columns, and they can run queries against your tables. The driver translates standard java.sql calls into requests to the Flink SQL Statements REST API. Every query runs as a snapshot query, so it returns a finite result set and then exits, which is the behavior JDBC clients expect.

Before you start, review the Prerequisites and How JDBC concepts map to Flink. The following walkthrough gets the driver, configures a connection, and runs a first query with sqlline, a command-line JDBC client. sqlline is only an example: Steps 1 through 3 apply to any JDBC client, so if you use a different tool, you can skip Step 4.

Step 1: Get the driver
Step 2: Collect connection coordinates
Step 3: Configure the connection
Step 4: Run a query with sqlline

After the walkthrough, you can learn more about:

Connect from a BI tool or SQL client
Connect from a Java application
How queries run
Connection property reference
Limitations

Prerequisites

You need the following prerequisites to complete this walkthrough:

Access to Confluent Cloud
A Flink compute pool in your Confluent Cloud environment
A Flink API key with role-based access control (RBAC) permissions to run statements in the compute pool and read the tables you want to query
The Confluent CLI, logged in with confluent login
Java 17 or later

The driver doesn’t require a compute pool. If you don’t set one, statements run in the default compute pool for your environment and region. This walkthrough sets a compute pool explicitly.

How JDBC concepts map to Flink

JDBC clients organize data as catalogs, schemas, and tables. Flink uses the same three-part hierarchy, but the names correspond to different Confluent Cloud entities. Use the following table to find what to enter in a connection dialog or what to expect in a schema browser:

JDBC concept	Flink SQL concept	Confluent Cloud entity
Catalog	Catalog	Environment
Schema	Database	Apache Kafka® cluster
Table	Table	Kafka topic

The catalog and database connection settings take the names of the environment and Kafka cluster, not their IDs. For example, use my-environment and my-cluster, not env-a1b2c3 and lkc-x1y2z3. The environment-id setting is different: it takes the environment ID.

Every environment also includes the read-only examples catalog, which contains the marketplace database. Its orders, customers, clicks, and products tables generate data continuously, so you can test your connection without creating any tables. For more information, see Example Data Streams in Confluent Cloud for Apache Flink.

Step 1: Get the driver

The driver ships as a single JAR file, confluent-jdbc-1.0.0.jar. The JAR file bundles all of the driver’s dependencies under a renamed package, so you can add it to the classpath of any JDBC client without conflicting with that client’s own libraries.

Download confluent-jdbc-1.0.0.jar and save it to a stable location, such as a ~/drivers directory. You reference this path when you configure a client.

To add the driver to a Maven project instead, add the Confluent Maven repository and the driver dependency to your pom.xml file:

<repositories>
  <repository>
    <id>confluent</id>
    <url>https://packages.confluent.io/maven/</url>
  </repository>
</repositories>

<dependencies>
  <dependency>
    <groupId>io.confluent.jdbc</groupId>
    <artifactId>confluent-jdbc</artifactId>
    <version>1.0.0</version>
  </dependency>
</dependencies>

Use the following values when a client asks for driver details:

Setting	Value
Driver class	`io.confluent.jdbc.avatica.ConfluentJdbcDriver`
URL prefix	`jdbc:confluent:flink://`

Step 2: Collect connection coordinates

The driver needs to know which organization, environment, and compute pool to run statements in, and where to find your tables.

Get your organization ID:
```
confluent organization list
```
Get your environment ID and name:
```
confluent environment list
```

Get your compute pool ID, cloud provider, and region:

confluent flink compute-pool list --environment <env_id>

Get the name of the Kafka cluster to use as your default database:
```
confluent kafka cluster list --environment <env_id>
```
If you don’t already have a Flink API key, create one:
```
confluent api-key create \
  --resource flink \
  --cloud <cloud_provider> \
  --region <region> \
  --environment <env_id>
```
Save the API key and secret. You can’t retrieve the secret later. For more options, including how to create a key owned by a service account, see Generate an API Key for Access in Confluent Cloud for Apache Flink.

Step 3: Configure the connection

You can give the driver its settings in two ways: in a properties file or as query parameters in the JDBC URL. A properties file uses the full property names, such as client.organization-id and sql.current-catalog. In the URL, you can also use short names for five of them, such as organization-id and catalog. A properties file ignores the short names and logs a warning, so don’t copy URL parameters into the file.

Use a properties file when you can. BI tools often log connection URLs, and your shell saves URLs that you type in a terminal, so a URL can leak your API secret.

Use a properties file

Create a file named flink-jdbc.properties with the following content, replacing the placeholders with the values that you collected in Step 2:

client.organization-id=<org_id>
client.environment-id=<env_id>
client.compute-pool-id=<compute_pool_id>
client.cloud=<cloud_provider>
client.region=<region>
client.flink-api-key=<flink_api_key>
client.flink-api-secret=<flink_api_secret>
sql.current-catalog=<environment_name>
sql.current-database=<kafka_cluster_name>

For example, set client.cloud=aws and client.region=us-east-1.

Restrict access to the file, because it contains your API secret:
```
chmod 600 flink-jdbc.properties
```
Don’t commit this file to source control.

Reference the file in the JDBC URL with the propertiesFile parameter:

jdbc:confluent:flink://?propertiesFile=/path/to/flink-jdbc.properties

Use URL query parameters

If your client can’t read a local file, encode the cloud provider and region in the host name and pass the remaining coordinates as query parameters. The URL has the following form:

jdbc:confluent:flink://<flink_host>?<query_parameters>

Build it from the following parts:

Part	Value
`<flink_host>`	`flink.<region>.<cloud_provider>.confluent.cloud`, for example `flink.us-east-1.aws.confluent.cloud`
`organization-id`	Your organization ID
`environment-id`	Your environment ID
`compute-pool-id`	Your compute pool ID. Optional.
`catalog`	Your environment name
`database`	Your Kafka cluster name

Separate the query parameters with &. For a complete URL, see the example in Connect from a Java application.

The cloud provider and region have no short names. If you pass cloud or region as a query parameter, the driver ignores it and logs a warning. Encode them in the host name instead, or use the full client.cloud and client.region names.

Provide the Flink API key and secret as the JDBC user and password, which most clients show as separate fields in their connection dialog. Don’t put credentials in the URL.

Precedence of connection settings

If you set the same property in more than one place, the driver resolves it in the following order. Credentials follow a different order from all other properties.

Source	Most properties	`user` and `password`
URL query parameters	1 (highest)	2
Client-supplied properties, such as the user and password fields in a connection dialog	2	1 (highest)
The `propertiesFile`	3	Not read

The driver accepts user and password as URL query parameters, but it logs a warning when you connect with credentials in the URL. For the credential properties to use in a properties file, see Connection property reference.

URL query parameters don’t always override a properties file. If you name a propertiesFile in the URL and the file contains your API key or secret, the URL can’t change where the driver sends them. The following settings must be absent from the URL or match the values in the file:

The URL host
client.flink-endpoint, client.rest-endpoint, client.cloud, and client.region
hostnameVerification, trustStorePath, and trustStoreType
proxyHost and proxyPort, if the file sets hostnameVerification=NONE or contains proxyPassword

Otherwise, the connection fails before the driver sends any request. For example, the following URL fails if the file sets client.region=us-west-2, because the host names a different region:

jdbc:confluent:flink://flink.us-east-1.aws.confluent.cloud?propertiesFile=/path/to/flink-jdbc.properties

You can add parameters that don’t affect the destination, such as catalog and database.

Step 4: Run a query with `sqlline`

The sqlline shell is a command-line client for JDBC databases, and it’s a quick way to verify your connection before you configure a BI tool.

Download sqlline-1.12.0-jar-with-dependencies.jar from Maven Central. This JAR file bundles sqlline with everything it needs to run.
Start sqlline with both JAR files on the classpath:
```
CP=sqlline-1.12.0-jar-with-dependencies.jar
CP=$CP:/path/to/confluent-jdbc-1.0.0.jar
java -cp "$CP" sqlline.SqlLine \
  -d io.confluent.jdbc.avatica.ConfluentJdbcDriver \
  -u "jdbc:confluent:flink://?propertiesFile=$HOME/flink-jdbc.properties" \
  -n "" -p "" \
  --showWarnings=false
```
The empty -n and -p options stop sqlline from prompting for a user name and password, so the driver reads your API key and secret from the properties file. The driver reports a warning for each empty value, and sqlline prints each warning with a stack trace. The --showWarnings=false option hides these warnings.

On Windows, separate classpath entries with ; instead of :.

When it connects, sqlline prints a message that the TRANSACTION_REPEATABLE_READ isolation level isn’t supported. sqlline requests that level by default, and the driver supports only TRANSACTION_NONE. The message appears on every Java version, and you can ignore it.

On Java 24 and later, sqlline also prints warnings about restricted native access when it starts. You can ignore these warnings too.
List the tables in your default database:
```
!tables
```

Query the examples catalog:

SELECT `order_id`, `customer_id`, `price`
FROM `examples`.`marketplace`.`orders`
LIMIT 5;

The query runs as a snapshot query in your compute pool and returns five rows. Your output should resemble:

+--------------------------------------+-------------+-------+
|               order_id               | customer_id | price |
+--------------------------------------+-------------+-------+
| b4462f7e-9fea-40fb-b12c-feaa655aaede | 3051        | 30.2  |
| 73604908-9f01-4396-87eb-67a3a7e58785 | 3226        | 16.14 |
| a453378a-fddb-4b38-87a7-2a158cb909a9 | 3237        | 69.09 |
| 4e3bde87-79fe-4be9-8cc3-3273dc36efe5 | 3117        | 53.27 |
| a00e537d-d458-4d96-b7e2-9db8110ff7c8 | 3247        | 72.64 |
+--------------------------------------+-------------+-------+
5 rows selected (3.8 seconds)

The examples tables generate new data continuously, so your values differ.

Try a join across two example tables:

SELECT c.`name`, COUNT(*) AS order_count
FROM `examples`.`marketplace`.`orders` AS o
JOIN `examples`.`marketplace`.`customers` AS c
  ON o.`customer_id` = c.`customer_id`
GROUP BY c.`name`
LIMIT 10;

Your output should resemble:

+-----------------------+-------------+
|         name          | order_count |
+-----------------------+-------------+
| Erin Luettgen         | 7           |
| Milton Heidenreich    | 2           |
| Dorine Denesik        | 8           |
| Sammie Hodkiewicz Sr. | 7           |
| Elton Huels           | 5           |
| Denyse Yundt          | 1           |
| Tena Blanda           | 3           |
| Arcelia Dickens       | 1           |
| Renata Bernhard       | 4           |
| Robt Kutch            | 3           |
+-----------------------+-------------+
10 rows selected (4.957 seconds)

To exit sqlline, enter !quit.

Connect from a BI tool or SQL client

Most JDBC-compatible tools follow the same pattern for adding a custom driver:

Register a new driver in the tool’s driver manager. Add the driver JAR file and set the driver class to io.confluent.jdbc.avatica.ConfluentJdbcDriver.
Create a connection that uses the new driver. Enter a JDBC URL from Step 3: Configure the connection.
If you used URL query parameters, enter your Flink API key as the user name and your Flink API secret as the password.
Test the connection, then browse the examples catalog to confirm that catalogs, databases, tables, and columns appear.

Connect from a Java application

Add the driver JAR file to your application’s classpath and open a connection with DriverManager. Pass credentials in a Properties object instead of the URL:

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Properties;

public class FlinkJdbcExample {
  public static void main(String[] args) throws Exception {
    String url = "jdbc:confluent:flink://flink.us-east-1.aws.confluent.cloud"
        + "?organization-id=<org_id>"
        + "&environment-id=<env_id>"
        + "&compute-pool-id=<compute_pool_id>"
        + "&catalog=<environment_name>"
        + "&database=<kafka_cluster_name>";

    Properties props = new Properties();
    props.setProperty("user", System.getenv("FLINK_API_KEY"));
    props.setProperty("password", System.getenv("FLINK_API_SECRET"));

    try (Connection conn = DriverManager.getConnection(url, props);
         Statement stmt = conn.createStatement();
         ResultSet rs = stmt.executeQuery(
             "SELECT `order_id`, `price` "
                 + "FROM `examples`.`marketplace`.`orders` LIMIT 5")) {
      while (rs.next()) {
        System.out.println(
            rs.getString("order_id") + " " + rs.getDouble("price"));
      }
    }
  }
}

To cancel a long-running query, call Statement.cancel() from another thread. The driver stops the query, and you can reuse the Statement afterward.

How queries run

Each query that you run through the driver becomes a Flink SQL statement in your compute pool, and you can see it in the Confluent Cloud Console alongside your other statements. The driver handles the following differences between JDBC and Flink for you:

Snapshot mode

The driver runs every query as a snapshot query that reads your data as of the moment you submit it. A query against a Kafka-backed table returns a finite result instead of streaming new rows indefinitely. You can’t change this setting, because a JDBC result set must end.

DDL and INSERT INTO statements

You can also run INSERT INTO and data definition language (DDL) statements, such as CREATE TABLE. The driver runs them as snapshot statements too, and it waits for each one to finish before it returns, up to the client.result-wait-timeout-millis timeout. For these statements, executeUpdate() returns 0, not the number of rows written.

Autocommit only

Flink SQL has no transactions, so every statement commits as soon as it runs. The driver handles the transaction methods as follows:

Method	Behavior
`setAutoCommit(false)`	Accepted, but the connection stays in autocommit mode. The first call adds an `SQLWarning` with the message `no transactions; every statement commits immediately`.
`getAutoCommit()`	Always returns `true`.
`commit()`	Has no effect.
`rollback()`	Succeeds if nothing has been written on the connection. After a write, raises `SQLFeatureNotSupportedException` with the message `Cannot roll back: a write has already committed on this connection`.

Prepared statements

You can use ? placeholders in a PreparedStatement. The Statements API has no parameter binding, so the driver substitutes each bound value into the SQL text as a literal, and it chooses the literal’s type from the Java type of the value you bind.

Row values

A ROW column reads as a java.sql.Struct. Nested ROW fields are also Struct values, so generic JDBC code can walk nested rows.

Error reporting

When a statement fails, the driver raises a SQLException with the error message from Flink and a standard SQLState code that it chooses from the server’s error. For example, a query on a table that doesn’t exist fails with SQLState 42S02. The original error from the server is attached as the exception’s cause.

Connection property reference

You can set any of the following properties as a URL query parameter, as a client-supplied property, or in a properties file, with the following exceptions:

The driver doesn’t read user and password from a properties file. Use client.flink-api-key and client.flink-api-secret instead.
The driver reads timeZone only from client-supplied properties.

Property	Description	Default
`user`	Flink API key. Takes precedence over `client.flink-api-key`.	None
`password`	Flink API secret. Takes precedence over `client.flink-api-secret`.	None
`propertiesFile`	Path to a properties file. Explicit properties take precedence over values in the file.	None
`client.organization-id`	Confluent Cloud organization ID.	None
`client.environment-id`	Confluent Cloud environment ID, in the form `env-xxxxxx`.	None
`client.compute-pool-id`	Compute pool that runs your statements, in the form `lfcp-xxxxxx`. Optional. If you don’t set it, statements run in the default compute pool for the environment and region.	None
`client.cloud`	Cloud provider of the compute pool. The driver uses it to build the Flink endpoint host name.	None
`client.region`	Region of the compute pool. The driver uses it to build the Flink endpoint host name.	None
`client.rest-endpoint`	Confluent Cloud domain that the driver combines with `client.cloud` and `client.region` to build the Flink endpoint host name.	`confluent.cloud`
`client.flink-endpoint`	Full Flink endpoint host or URL. Replaces the host name that the driver builds from `client.cloud` and `client.region`. Use this to connect to a private networking endpoint.	None
`client.flink-api-key`	Flink API key.	None
`client.flink-api-secret`	Flink API secret.	None
`sql.current-catalog`	Default catalog (environment name) for unqualified table names. In the URL, use `catalog`.	None
`sql.current-database`	Default database (Kafka cluster name) for unqualified table names. In the URL, use `database`.	None
`client.request-timeout-millis`	Timeout for each HTTP request.	60,000
`client.connect-timeout-millis`	Timeout for opening a connection.	10,000
`client.result-wait-timeout-millis`	Longest time to wait for a result schema or a page of results. `Statement.setQueryTimeout()` overrides this for a statement.	3,600,000
`client.poll-interval-millis`	Delay between polls for a statement’s status. The driver treats a value above 60,000 as 60,000.	500
`client.retry-budget-millis`	Total time that the driver spends retrying a request after a transient failure. Set to `0` to turn off retries. The default allows two full-length attempts at the default request timeout, plus the longest wait between attempts. It doesn’t change when you change `client.request-timeout-millis`.	126,250
`timeZone`	Time zone that the driver uses to read `DATE`, `TIME`, and `TIMESTAMP` values. It’s also the session time zone, unless you set `sql.local-time-zone`. For details, see Time zone settings.	The Java virtual machine (JVM) default
`sql.local-time-zone`	Session time zone that the driver sends with every statement. Use a time zone ID, a GMT offset such as `GMT+01:00`, or `UTC`. For details, see Time zone settings.	The connection’s `timeZone`, otherwise the JVM default
`proxyHost`	Host name of an HTTP proxy.	None
`proxyPort`	Port of the HTTP proxy, from 1 to 65535.	None
`proxyUser`	User name for the HTTP proxy.	None
`proxyPassword`	Password for the HTTP proxy.	None
`nonProxyHosts`	Hosts to reach without the proxy, as a pipe-separated list of host patterns. A pattern can include `*`.	None
`trustStorePath`	Path to a trust store with custom certificate authorities for Transport Layer Security (TLS), for example to trust a corporate proxy’s certificate authority.	None
`trustStorePassword`	Password for the trust store.	None
`trustStoreType`	Type of the trust store, such as `PKCS12`.	The JVM default key store type
`keyStorePath`	Path to a key store with a client certificate for mutual TLS (mTLS) authentication.	None
`keyStorePassword`	Password for the key store.	None
`keyPassword`	Password for the private key in the key store.	The value of `keyStorePassword`
`keyStoreType`	Type of the key store, such as `PKCS12`.	The JVM default key store type
`hostnameVerification`	`STRICT` or `NONE`, in any case. Set to `NONE` to skip the check that the server certificate matches the host name. The driver still validates the certificate chain. Use `NONE` only behind a TLS-inspecting proxy, and prefer trusting the proxy’s certificate authority with `trustStorePath`.	`STRICT`

The timeout, interval, and retry properties take values in milliseconds.

The driver fails the connection if a trust store or key store password or type is set without its store path, or if only one of proxyUser and proxyPassword is set. It also fails the connection if hostnameVerification has any value other than STRICT or NONE.

If you set proxyUser and proxyPassword, the driver rejects them unless you remove Basic from the JVM’s jdk.http.auth.tunneling.disabledSchemes system property, which lists Basic by default. Otherwise, use a proxy that doesn’t require authentication.

Time zone settings

The driver accepts timeZone in two other spellings, TIMEZONE and TIME_ZONE, in any case. Set only one spelling. If the value is blank or isn’t a known time zone, the driver uses GMT.

The sql.local-time-zone property determines how CURRENT_TIMESTAMP and TIMESTAMP_LTZ values resolve. It doesn’t change the time zone that the driver uses to read DATE, TIME, and TIMESTAMP results.

Limitations

The Confluent JDBC driver has the following limitations:

No transactions: Connections are autocommit-only. setAutoCommit(false) adds a warning instead of starting a transaction, and rollback() raises SQLFeatureNotSupportedException after a write. For more information, see How queries run.
No batch execution: executeBatch() raises SQLFeatureNotSupportedException with the message Batch execution is not supported.
Forward-only result sets: Scrollable result sets aren’t supported. If your client requests a scrollable result set, the driver doesn’t raise an error, and it returns a forward-only result set. Methods such as previous() and absolute() raise SQLFeatureNotSupportedException.
No streaming results: Every query runs as a snapshot query. To process data continuously, deploy a statement in the Cloud Console, the Confluent CLI, or the REST API.