# 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](flink-rest-api.md#flink-rest-api). Every query runs as a [snapshot query](../concepts/snapshot-queries.md#flink-sql-snapshot-queries), so it returns a finite result set and then exits, which is the behavior JDBC clients expect. Before you start, review the [Prerequisites](#flink-connect-jdbc-prerequisites) and [How JDBC concepts map to Flink](#flink-connect-jdbc-concepts). 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](#flink-connect-jdbc-get-driver) - [Step 2: Collect connection coordinates](#flink-connect-jdbc-coordinates) - [Step 3: Configure the connection](#flink-connect-jdbc-configure) - [Step 4: Run a query with sqlline](#flink-connect-jdbc-sqlline) After the walkthrough, you can learn more about: - [Connect from a BI tool or SQL client](#flink-connect-jdbc-bi-tools) - [Connect from a Java application](#flink-connect-jdbc-java) - [How queries run](#flink-connect-jdbc-query-behavior) - [Connection property reference](#flink-connect-jdbc-properties) - [Limitations](#flink-connect-jdbc-limitations) ## Prerequisites You need the following prerequisites to complete this walkthrough: - [Access to Confluent Cloud](https://confluent.cloud/) - A [Flink compute pool](../concepts/compute-pools.md#flink-sql-compute-pools) in your Confluent Cloud environment - A [Flink API key](generate-api-key-for-flink.md#flink-generate-api-key) with [role-based access control (RBAC)](flink-rbac.md#flink-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](../reference/example-data.md#flink-sql-example-data). ## 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](https://packages.confluent.io/maven/io/confluent/jdbc/confluent-jdbc/1.0.0/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: ```xml confluent https://packages.confluent.io/maven/ io.confluent.jdbc confluent-jdbc 1.0.0 ``` 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. 1. Get your organization ID: ```bash confluent organization list ``` 2. Get your environment ID and name: ```bash confluent environment list ``` 3. Get your compute pool ID, cloud provider, and region: ```bash confluent flink compute-pool list --environment ``` 4. Get the name of the Kafka cluster to use as your default database: ```bash confluent kafka cluster list --environment ``` 5. If you don’t already have a Flink API key, create one: ```bash confluent api-key create \ --resource flink \ --cloud \ --region \ --environment ``` 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](generate-api-key-for-flink.md#flink-generate-api-key). ## 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 1. Create a file named `flink-jdbc.properties` with the following content, replacing the placeholders with the values that you collected in Step 2: ```properties client.organization-id= client.environment-id= client.compute-pool-id= client.cloud= client.region= client.flink-api-key= client.flink-api-secret= sql.current-catalog= sql.current-database= ``` For example, set `client.cloud=aws` and `client.region=us-east-1`. 2. Restrict access to the file, because it contains your API secret: ```bash chmod 600 flink-jdbc.properties ``` Don’t commit this file to source control. 3. Reference the file in the JDBC URL with the `propertiesFile` parameter: ```text 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: ```text jdbc:confluent:flink://? ``` Build it from the following parts: | Part | Value | |-------------------|----------------------------------------------------------------------------------------------------------| | `` | `flink...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](#flink-connect-jdbc-java). 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](#flink-connect-jdbc-properties). 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: ```text 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. 1. Download `sqlline-1.12.0-jar-with-dependencies.jar` from [Maven Central](https://repo1.maven.org/maven2/sqlline/sqlline/1.12.0/). This JAR file bundles `sqlline` with everything it needs to run. 2. Start `sqlline` with both JAR files on the classpath: ```bash 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. 3. List the tables in your default database: ```text !tables ``` 4. Query the `examples` catalog: ```sql 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: ```text +--------------------------------------+-------------+-------+ | 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. 5. Try a join across two example tables: ```sql 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: ```text +-----------------------+-------------+ | 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: 1. 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`. 2. Create a connection that uses the new driver. Enter a JDBC URL from [Step 3: Configure the connection](#flink-connect-jdbc-configure). 3. If you used URL query parameters, enter your Flink API key as the user name and your Flink API secret as the password. 4. 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: ```java 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=" + "&environment-id=" + "&compute-pool-id=" + "&catalog=" + "&database="; 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](../concepts/snapshot-queries.md#flink-sql-snapshot-queries) 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](../concepts/flink-private-networking.md#flink-sql-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](#flink-connect-jdbc-time-zones). | 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](#flink-connect-jdbc-time-zones). | 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](#flink-connect-jdbc-query-behavior). 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](flink-rest-api.md#flink-rest-api). ## Related content - [Snapshot Queries in Confluent Cloud for Apache Flink](../concepts/snapshot-queries.md#flink-sql-snapshot-queries) - [Generate an API Key for Access in Confluent Cloud for Apache Flink](generate-api-key-for-flink.md#flink-generate-api-key) - [Example Data Streams in Confluent Cloud for Apache Flink](../reference/example-data.md#flink-sql-example-data) - [Flink SQL REST API for Confluent Cloud for Apache Flink](flink-rest-api.md#flink-rest-api) - [Grant Role-Based Access in Confluent Cloud for Apache Flink](flink-rbac.md#flink-rbac) - [Compute Pools in Confluent Cloud for Apache Flink](../concepts/compute-pools.md#flink-sql-compute-pools) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).