<a id="flink-table-api"></a>

# Table API on Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® supports programming applications with the Table API in Java and
Python. Confluent provides a plugin for running applications that use the
Table API on Confluent Cloud.

The Table API enables a programmatic way of developing, testing, and submitting
Flink pipelines for processing data streams. Streams can be finite or infinite,
with insert-only or changelog data. Changelog data enables handling Change Data
Capture (CDC) events.

To use the Table API, you work with tables that change over time, a concept
inspired by relational databases. A Table program is a declarative and
structured graph of transformations. The Table API is inspired by SQL and
complements it with additional tools for manipulating real-time data. You can
use both Flink SQL and the Table API in your applications.

A table program has these characteristics:

- Runs in a regular `main()` method (Java)
- Uses Flink APIs
- Communicates with Confluent Cloud by using REST requests, for example,
  [Statements endpoint](/cloud/current/api.html#tag/Statements-(sqlv1)/operation/createSqlv1Statement).

For a list of Table API functions supported by Confluent Cloud for Apache Flink, see
[Table API functions](functions/table-api-functions.md#flink-table-api-functions).

For a list of Table API limitations in Confluent Cloud for Apache Flink, see
[Migrate from Open Preview](#flink-table-api-limitations).

Use the Confluent for VS Code extension to generate a new Flink Table API
project that interacts with your Confluent Cloud resources. This option is ideal if
you’re learning about the Table API.

For more information see [Confluent for VS Code for Confluent Cloud](../../client-apps/vs-code-extension.md#cc-vscode-extension).

#### NOTE
The Flink Table API for Java is generally available. The Table API for
Python is available for preview.


A Preview feature is a Confluent Cloud component that is being introduced to gain
early feedback from developers. Preview features can be used for evaluation
and non-production testing purposes or to provide feedback to Confluent.
The warranty, SLA, and Support Services provisions of your agreement with
Confluent do not apply to Preview features. Confluent may discontinue
providing preview releases of the Preview features at any time in
Confluent’s’ sole discretion.

The preceding preview notice applies only to the Table API for Python.

Comments, questions, and suggestions related to the Table API are
encouraged and can be submitted through the [established channels](../get-help.md#ccloud-flink-help).

<a id="flink-java-table-api-add-to-existing-project"></a>

## Add the Table API to an existing Java project

To add the Table API to an existing project, include the following dependencies
in the `<dependencies>` section of your pom.xml file.

```xml
<!-- Apache Flink dependencies -->
<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-table-api-java</artifactId>
   <version>${flink.version}</version>
</dependency>

<!-- Confluent Flink Table API Java plugin -->
<dependency>
   <groupId>io.confluent.flink</groupId>
   <artifactId>confluent-flink-table-api-java-plugin</artifactId>
   <version>${confluent-plugin.version}</version>
</dependency>
```

<a id="flink-java-table-api-configure-plugin"></a>

## Configure the plugin

The plugin requires a set of configuration options for establishing a connection
to Confluent Cloud. The following configuration options are required.

| Property key             | Command-line argument   | Environment variable   | Notes                                                                                                                                                                                                                 |
|--------------------------|-------------------------|------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| client.cloud             | –cloud                  | CLOUD_PROVIDER         | Confluent identifier for a cloud provider. Valid values are `aws`, `azure`, and `gcp`.                                                                                                                                |
| client.compute-pool-id   | –compute-pool-id        | COMPUTE_POOL_ID        | (optional) ID of the compute pool, for example, `lfcp-1a2b3c`. If not specified, the Table API uses the default compute pool.                                                                                         |
| client.environment-id    | –environment-id         | ENV_ID                 | ID of the environment, for example, `env-z3y2x1`.                                                                                                                                                                     |
| client.flink-api-key     | –flink-api-key          | FLINK_API_KEY          | API key for Flink access. Required unless a global API key and secret are configured. For more information, see [Generate an API Key](../operate-and-deploy/generate-api-key-for-flink.md#flink-generate-api-key).    |
| client.flink-api-secret  | –flink-api-secret       | FLINK_API_SECRET       | API secret for Flink access. Required unless a global API key and secret are configured. For more information, see [Generate an API Key](../operate-and-deploy/generate-api-key-for-flink.md#flink-generate-api-key). |
| client.global-api-key    | –global-api-key         | GLOBAL_API_KEY         | (optional) API key for both Flink access and artifact creation. Recommended for most applications. For more information, see [Authentication](#flink-table-api-authentication).                                       |
| client.global-api-secret | –global-api-secret      | GLOBAL_API_SECRET      | (optional) API secret for both Flink access and artifact creation. Recommended for most applications. For more information, see [Authentication](#flink-table-api-authentication).                                    |
| client.organization-id   | –organization-id        | ORG_ID                 | ID of the organization, for example, `b0b21724-4586-4a07-b787-d0bb5aacbf87`.                                                                                                                                          |
| client.region            | –region                 | CLOUD_REGION           | Confluent identifier for a cloud provider’s region, for example, `us-east-1`. For available regions, see [Supported Regions](../overview.md#ccloud-flink-overview-everywhere) or run `confluent flink region list`.   |

The following configuration options are required for supporting UDF uploads
when no global API key and secret are configured. If you use a global API key,
it covers artifact creation, and you don’t need a separate artifact key. For
more information, see
[Upload the artifact to Confluent Cloud](../how-to-guides/create-udf.md#flink-sql-create-udf-upload-artifact).

#### NOTE
Create a Confluent Cloud API key artifact key and secret in Confluent Cloud Console under
**API keys** > **Cloud resource management**. For more information, see
[Manage API Keys in |ccloud|](../../security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.md#manage-api-keys).

| Property key               | Command-line argument   | Environment variable   | Notes                            |
|----------------------------|-------------------------|------------------------|----------------------------------|
| client.artifact-api-key    | –artifact-api-key       | ARTIFACT_API_KEY       | API key for artifact creation    |
| client.artifact-api-secret | –artifact-api-secret    | ARTIFACT_API_SECRET    | API secret for artifact creation |

The following configuration options are optional.

| Property key                      | Command-line argument       | Environment variable       | Notes                                                                                                                                                                                                                                                                                                                                            |
|-----------------------------------|-----------------------------|----------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| client.action.await               | –action.await [<duration>]  |                            | When set, the `resume`, `stop`, and `delete` lifecycle actions block until the target phase is reached or `client.timeout` elapses. An optional duration overrides the timeout. The default is `false`.                                                                                                                                          |
| client.action.kind                |                             |                            | Lifecycle action for CI/CD integration. One of `list`, `describe`, `resume`, `stop`, or `delete`. Usually passed as the first argument when running the application JAR. For more information, see [Deploy and Manage Table API Programs on Confluent Cloud for Apache Flink](../operate-and-deploy/table-api-deploy.md#flink-table-api-deploy). |
| client.action.skip-exit           |                             |                            | Skip `System.exit()` after a lifecycle action runs. The default is `false`.                                                                                                                                                                                                                                                                      |
| client.application-name           | –application-name           | APPLICATION_NAME           | A name for this Table API application. Serves as a namespace prefix for all statement names. Lowercase alphanumeric characters and hyphens only. The combined statement name, including the application name prefix, is limited to 100 characters. For example, `my-table-program`.                                                              |
| client.artifact-endpoint-template | –artifact-endpoint-template | ARTIFACT_ENDPOINT_TEMPLATE | A template for the artifact endpoint URL, for example, `https://api.{region}.{cloud}.confluent.cloud`.                                                                                                                                                                                                                                           |
| client.catalog-cache              |                             |                            | Expiration time for catalog objects, for example, `'5 min'`. The default is `'1 min'`. `'0'` disables caching.                                                                                                                                                                                                                                   |
| client.endpoint-template          | –endpoint-template          | ENDPOINT_TEMPLATE          | A template for the endpoint URL, for example, `https://flinkpls-dom123.{region}.{cloud}.confluent.cloud`.                                                                                                                                                                                                                                        |
| client.on-conflict                | –on-conflict                | ON_CONFLICT                | Behavior when a statement with the same name already exists with a different spec: `fail` (default) or `replace`. Requires `client.application-name`. For more information, see [Deploy and Manage Table API Programs on Confluent Cloud for Apache Flink](../operate-and-deploy/table-api-deploy.md#flink-table-api-deploy).                    |
| client.principal-id               | –principal-id               | PRINCIPAL_ID               | Principal that runs submitted statements, for example, `sa-23kgz4` for a service account.                                                                                                                                                                                                                                                        |
| client.rest-endpoint              | –rest-endpoint              | REST_ENDPOINT              | URL to the REST endpoint, for example, `proxyto.confluent.cloud`.                                                                                                                                                                                                                                                                                |
| client.statement-name             | –statement-name             | STATEMENT_NAME             | Unique name for statement submission. If an application name is set, it’s prefixed to this name. By default, generated using a UUID.                                                                                                                                                                                                             |
| client.timeout                    |                             |                            | Maximum time to wait when `client.action.await` is set, or for the blocking `ConfluentTools` and `StatementHandle` lifecycle methods, for example, `5min` or `900s`. The default is `15min`.                                                                                                                                                     |
| client.tmp-dir                    | –tmp-dir                    |                            | Directory for temporary files created by the plugin, like UDF jars, for example, `/tmp`. The default is `java.io.tmpdir`.                                                                                                                                                                                                                        |

<a id="flink-table-api-authentication"></a>

### Authentication

The plugin authenticates with the Confluent Cloud REST APIs by using the mode that
you select with the `client.auth-mode` option. If `client.auth-mode` is
not set, the default is `api-key`.

| Property key     | Command-line argument   | Environment variable   | Notes                                                                                                               |
|------------------|-------------------------|------------------------|---------------------------------------------------------------------------------------------------------------------|
| client.auth-mode | –auth-mode              | AUTH_MODE              | One of `api-key`, `oauth-client-credentials`, or `oauth-static-token` (case-insensitive). The default is `api-key`. |

#### API keys (default)

In the default `api-key` mode, the plugin first tries the global API key
and secret (`client.global-api-key` and `client.global-api-secret`),
which work for both Flink access and artifact creation. Global API keys are
the recommended way to authenticate your Table API applications with
Confluent Cloud. For supported resources, limitations, and how to create a global
API key, see [Global API keys](../../security/authenticate/workload-identities/service-accounts/api-keys/overview.md#cloud-global-api-keys).

If no global key and secret are set, the plugin falls back to the dedicated
Flink API key and secret (`client.flink-api-key` and
`client.flink-api-secret`) and, only when you upload UDF artifacts, a
separate artifact API key and secret (`client.artifact-api-key` and
`client.artifact-api-secret`).

#### OAuth

The plugin supports two OAuth-based authentication modes.

- `oauth-client-credentials`: the plugin fetches and refreshes access
  tokens from an external identity provider by using the OAuth 2.0 client
  credentials flow.
- `oauth-static-token`: you supply a pre-issued bearer token as a string
  by using the `client.oauth.external-access-token` option, or
  programmatically with the Java builder method
  `setOAuthTokenProvider(OAuthTokenProvider)` for cloud-native flows, for
  example, Azure Managed Identity or AWS IAM workload identity.

In both modes, a Confluent Cloud [identity pool](../../security/authenticate/workload-identities/identity-providers/oauth/identity-pools.md#add-oauth-identity-pools)
with the correct permission assignments must exist for the intended
workload. When the plugin runs in an OAuth mode, UDF artifact uploads work
without further configuration. For detailed setup instructions, see
[OAuth for Confluent Cloud](../../security/authenticate/workload-identities/identity-providers/oauth/overview.md#oauth-overview).

The following configuration options apply to the OAuth modes.

| Property key                        | Command-line argument         | Environment variable         | Notes                                                                                                                                               |
|-------------------------------------|-------------------------------|------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------|
| client.oauth.external-access-token  | –oauth.external-access-token  | OAUTH_EXTERNAL_ACCESS_TOKEN  | Pre-issued bearer token, provided as a string. No refreshes are performed. Required for `oauth-static-token` when no token provider is set in code. |
| client.oauth.external-client-id     | –oauth.external-client-id     | OAUTH_EXTERNAL_CLIENT_ID     | Client ID registered with the identity provider. Required for `oauth-client-credentials`.                                                           |
| client.oauth.external-client-secret | –oauth.external-client-secret | OAUTH_EXTERNAL_CLIENT_SECRET | Client secret for the configured client ID. Required for `oauth-client-credentials`.                                                                |
| client.oauth.external-token-scope   | –oauth.external-token-scope   | OAUTH_EXTERNAL_TOKEN_SCOPE   | (optional) Additional scopes attached during the client credentials flow.                                                                           |
| client.oauth.external-token-url     | –oauth.external-token-url     | OAUTH_EXTERNAL_TOKEN_URL     | URL of the identity provider’s OAuth 2.0 token endpoint. Required for `oauth-client-credentials`.                                                   |
| client.oauth.identity-pool-id       | –oauth.identity-pool-id       | OAUTH_IDENTITY_POOL_ID       | Confluent Cloud identity pool ID, for example, `pool-xxxxx`. Required for any OAuth mode.                                                           |

The following example shows the client credentials flow configured with
environment variables.

```bash
export AUTH_MODE="oauth-client-credentials"
export OAUTH_EXTERNAL_TOKEN_URL="https://mycompany.okta.com/oauth2/abc123/v1/token"
export OAUTH_EXTERNAL_CLIENT_ID="cid"
export OAUTH_EXTERNAL_CLIENT_SECRET="csec"
export OAUTH_IDENTITY_POOL_ID="pool-xxxxx"
export OAUTH_EXTERNAL_TOKEN_SCOPE="write:service"
```

The following example shows a static token configured with environment
variables.

```bash
export AUTH_MODE="oauth-static-token"
export OAUTH_EXTERNAL_ACCESS_TOKEN="eyJ-..."
export OAUTH_IDENTITY_POOL_ID="pool-xxxxx"
```

For cloud-native flows, provide a token programmatically with
`setOAuthTokenProvider`. The following example wraps Azure Managed
Identity. The same `OAuthTokenProvider` interface can wrap AWS STS
`AssumeRoleWithWebIdentity` or any other provider.

```java
TokenCredential credential = new DefaultAzureCredentialBuilder().build();
TokenRequestContext request =
    new TokenRequestContext().addScopes("api://<client_id>/.default");

ConfluentSettings settings = ConfluentSettings.newBuilder()
    .setAuthMode(AuthMode.OAUTH_STATIC_TOKEN)
    .setOAuthIdentityPoolId("pool-xxxxx")
    .setOAuthTokenProvider(
        () -> {
          AccessToken at = credential.getToken(request).block();
          return new OAuthToken(at.getToken(), at.getExpiresAt().toInstant());
        })
    .build();
```

### Endpoint configuration

The Confluent Flink plugin provides options to configure endpoints for
connecting to Confluent Cloud services. The template-based approach is the
preferred method.

#### client.endpoint-template

This option provides a template for constructing the Flink statement API
endpoint URL.

- Default value: `https://flink.{region}.{cloud}.confluent.cloud`
- Example: `https://flinkpls-dom123.{region}.{cloud}.confluent.cloud`
- Usage: The template supports placeholders {region} and {cloud} that are replaced with the configured region and cloud provider values.
- Environment Variable: `ENDPOINT_TEMPLATE`

#### client.artifact-endpoint-template

This option provides a template for constructing the URL used for uploading
artifacts, like UDF JARs.

- Default value: `https://api.confluent.cloud`
- Example: `https://api.{region}.{cloud}.confluent.cloud`
- Usage: Similar to the endpoint template, this supports placeholders
  `{region}` and `{cloud}`.
- Environment Variable: `ARTIFACT_ENDPOINT_TEMPLATE`

#### client.rest-endpoint (Deprecated)

This option specifies the base domain for REST API calls to Confluent Cloud. While
still supported, using the template-based configuration is preferred.

- Default value: No default value
- Example: `proxy.confluent.cloud`
- Usage: When specified, the plugin constructs the full Flink statement API
  endpoint URL as `https://flink.{region}.{cloud}.{rest-endpoint}` where
  `{region}` and `{cloud}` are replaced with the configured region and
  cloud provider values.
- Environment Variable: `REST_ENDPOINT`

#### IMPORTANT
`client.endpoint-template` and `client.rest-endpoint` are mutually
exclusive. If you set both, Flink throws an exception.

#### Relationship and default behavior

The following rules control the relationship between the configuration options.

- The `client.endpoint-template` and `client.rest-endpoint` configuration
  options can’t be set simultaneously
- The `client.artifact-endpoint-template` and `client.rest-endpoint`
  configuration options can’t be set simultaneously.

The following rules control the default behavior.

- If neither `client.rest-endpoint` nor `client.endpoint-template` is
  configured, the default template, `https://flink.{region}.{cloud}.confluent.cloud`
  is used for statement API
- If neither `client.rest-endpoint` nor `client.artifact-endpoint-template`
  is specified, the default artifact endpoint, `https://api.confluent.cloud`
  is used
- If endpoint templates are used, each endpoint is constructed independently
  with the provided templates.

The following simple example shows different ways to configure endpoints.

```java
// Option 1 (RECOMMENDED): Using endpoint templates
// Resolved endpoints:
// - Statement API: https://flinkpls-dom123.us-east-1.aws.confluent.cloud
ConfluentSettings settings1 = ConfluentSettings.newBuilder()
      .setRegion("us-east-1")
      .setCloud("aws")
      .setEndpointTemplate("https://flinkpls-dom123.{region}.{cloud}.confluent.cloud")
      .setArtifactEndpointTemplate("https://artifacts.{region}.{cloud}.custom-domain.com")
      // Other required settings...
      .build();

// Option 2: Using properties file with endpoint templates
// cloud.properties:
// client.region=us-east-1
// client.cloud=aws
// client.endpoint-template=https://flinkpls-dom123.{region}.{cloud}.confluent.cloud
// Resolved endpoints:
// - Statement API: https://flinkpls-dom123.us-east-1.aws.confluent.cloud
// - Artifact API: https://api.confluent.cloud (default)
ConfluentSettings settings2 = ConfluentSettings.fromResource("/cloud.properties");

// Option 3 (DISCOURAGED): Using rest-endpoint (both statement endpoint will be derived from this)
// Resolved endpoints:
// - Statement API: https://flink.us-east-1.aws.proxy.confluent.cloud
// - Artifact API: https://api.proxy.confluent.cloud
ConfluentSettings settings3 = ConfluentSettings.newBuilder()
   .setRegion("us-east-1")
   .setCloud("aws")
   .setRestEndpoint("proxy.confluent.cloud")
   // Other required settings...
   .build();
```

### `ConfluentSettings` class

The `ConfluentSettings` class provides configuration options from various
sources, so you can combine external input, code, and environment variables
to set up your applications.

The following precedence order applies to configuration sources, from highest
to lowest:

- CLI arguments or properties file
- Code
- Environment variables

The following code example shows a `TableEnvironment` that’s configured by a
combination of command-line arguments and code.

### Java

```java
public static void main(String[] args) {
  // Args might set cloud, region, org, env, and compute pool.
  // Environment variables might pass key and secret.

  // Code sets the application name and SQL-specific options.
  ConfluentSettings settings = ConfluentSettings.newBuilderFromArgs(args)
   .setApplicationName("my-table-program")
   .setOption("sql.local-time-zone", "UTC")
   .build();

  TableEnvironment env = TableEnvironment.create(settings);
}
```

### Python

```python
from pyflink.table.confluent import ConfluentSettings
from pyflink.table import TableEnvironment

def run():
  # Properties file might set cloud, region, org, env, and compute pool.
  # Environment variables might pass key and secret.

  # Code sets the session name and SQL-specific options.
  settings = ConfluentSettings.new_builder_from_file(...) \
   .set_context_name("MyTableProgram") \
   .set_option("sql.local-time-zone", "UTC") \
   .build()

  env = TableEnvironment.create(settings)
```

### Properties file

You can store options in a `cloud.properties` file and reference the file in
code.

```properties
# Cloud region
client.cloud=aws
client.region=eu-west-1

# Access & compute resources
client.flink-api-key=XXXXXXXXXXXXXXXX
client.flink-api-secret=XxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXx
client.organization-id=00000000-0000-0000-0000-000000000000
client.environment-id=env-xxxxx
client.compute-pool-id=lfcp-xxxxxxxxxx
```

Reference the `cloud.properties` file in code:

### Java

```java
// Arbitrary file location in file system
ConfluentSettings settings = ConfluentSettings.fromPropertiesFile("/path/to/cloud.properties");

// Part of the JAR package (in src/main/resources)
ConfluentSettings settings = ConfluentSettings.fromPropertiesResource("/cloud.properties");
```

### Python

```python
from pyflink.table.confluent import ConfluentSettings

# Arbitrary file location in file system
settings = ConfluentSettings.from_file("/path/to/cloud.properties")
```

### Command-line arguments

You can pass the configuration settings as command-line options when you run
your application’s JAR:

```bash
java -jar my-table-program.jar \
  --cloud aws \
  --region us-east-1 \
  --flink-api-key key \
  --flink-api-secret secret \
  --organization-id b0b21724-4586-4a07-b787-d0bb5aacbf87 \
  --environment-id env-z3y2x1 \
  --compute-pool-id lfcp-1a2b3c
```

Access the configuration settings from the command-line arguments by using the
`ConfluentSettings.fromArgs` method:

### Java

```java
public static void main(String[] args) {
  ConfluentSettings settings = ConfluentSettings.fromArgs(args);
}
```

### Python

```python
from pyflink.table.confluent import ConfluentSettings

settings = ConfluentSettings.from_global_variables()
```

You can also layer command-line arguments on top of an existing builder, so a
single program combines a base configuration, for example, from a properties
file, with per-invocation arguments. Use the `applyArgs` method. The base
configuration provides the defaults, and the command-line arguments take
precedence over it.

```java
public static void main(String[] args) {
  ConfluentSettings settings = ConfluentSettings.newBuilder() // base config from the FLINK_PROPERTIES file, if set
    .setApplicationName("my-table-program")                   // code default, overridable by --application-name
    .applyArgs(args)                                          // per-invocation overrides
    .build();
}
```

### Code

You can assign the configuration settings in code by using the builder
provided with the `ConfluentSettings` class:

### Java

```java
ConfluentSettings settings = ConfluentSettings.newBuilder()
  .setCloud("aws")
  .setRegion("us-east-1")
  .setFlinkApiKey("key")
  .setFlinkApiSecret("secret")
  .setOrganizationId("b0b21724-4586-4a07-b787-d0bb5aacbf87")
  .setEnvironmentId("env-z3y2x1")
  .setComputePoolId("lfcp-1a2b3c")
  .build();
```

### Python

```python
from pyflink.table.confluent import ConfluentSettings

settings = ConfluentSettings.new_builder() \
  .set_cloud("aws") \
  .set_region("us-east-1") \
  .set_flink_api_key("key") \
  .set_flink_api_secret("secret") \
  .set_organization_id("b0b21724-4586-4a07-b787-d0bb5aacbf87") \
  .set_environment_id("env-z3y2x1") \
  .set_compute_pool_id("lfcp-1a2b3c") \
  .build()
```

#### NOTE
The `compute_pool_id` setting is optional. If not specified, the Table API
uses the default compute pool for your environment and region.

<a id="flink-table-api-environment-variables"></a>

### Environment variables

Set the following environment variables to provide configuration settings.
Setting COMPUTE_POOL_ID is optional. If not specified, the Table API uses the
default compute pool.

```bash
export CLOUD_PROVIDER="aws"
export CLOUD_REGION="us-east-1"
export FLINK_API_KEY="key"
export FLINK_API_SECRET="secret"
export ORG_ID="b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="env-z3y2x1"
export COMPUTE_POOL_ID="lfcp-1a2b3c"

java -jar my-table-program.jar
```

In code, call:

### Java

```java
ConfluentSettings settings = ConfluentSettings.fromGlobalVariables();
```

### Python

```python
from pyflink.table.confluent import ConfluentSettings

settings = ConfluentSettings.from_global_variables()
```

## Confluent utilities

The `ConfluentTools` class provides more methods that you can use for
developing and testing Table API programs.

### `ConfluentTools.collectChangelog` and `ConfluentTools.printChangelog`

Runs the specified table transformations on Confluent Cloud and returns the results
locally as a list of changelog rows or prints to the console in a table style.

These methods run `table.execute().collect()` and consume a fixed number of
rows from the returned iterator.

These methods can work on both finite and infinite input tables. If the
pipeline is potentially unbounded, they stop fetching after the desired number
of rows has been reached.

### Java

```java
// On a Table object
Table table = env.from("examples.marketplace.customers");
List<Row> rows = ConfluentTools.collectMaterialized(table, 100);
ConfluentTools.printMaterialized(table, 100);

// On a TableResult object
TableResult tableResult = env.executeSql("SELECT * FROM examples.marketplace.customers");
List<Row> rows = ConfluentTools.collectMaterialized(tableResult, 100);
ConfluentTools.printMaterialized(tableResult, 100);

// For finite (i.e. bounded) tables
ConfluentTools.collectMaterialized(table);
ConfluentTools.printMaterialized(table);
```

### Python

```python
from pyflink.table.confluent import ConfluentSettings, ConfluentTools
from pyflink.table import TableEnvironment

settings = ConfluentSettings.from_global_variables()
env = TableEnvironment.create(settings)
# On a Table object
table = env.from_path("examples.marketplace.customers")
rows = ConfluentTools.collect_changelog_limit(table, 100)
ConfluentTools.print_changelog_limit(table, 100)

# On a TableResult object
tableResult = env.execute_sql("SELECT * FROM examples.marketplace.customers")
rows = ConfluentTools.collect_changelog_limit(tableResult, 100)
ConfluentTools.print_changelog_limit(tableResult, 100)

# For finite (i.e. bounded) tables
ConfluentTools.collect_changelog(table)
ConfluentTools.print_changelog(table)
```

### `ConfluentTools.collect_materialized` and `ConfluentTools.print_materialized`

Runs the specified table transformations on Confluent Cloud and returns the results
locally as a materialized changelog. Changes are applied to an in-memory table
and returned as a list of insert-only rows or printed to the console in a table
style.

These methods run `table.execute().collect()` and consume a fixed number of
rows from the returned iterator.

These methods can work on both finite and infinite input tables. If the pipeline
is potentially unbounded, they stop fetching after the desired number of rows
have been reached.

### Java

```java
// On a Table object
Table table = env.from("examples.marketplace.customers");
List<Row> rows = ConfluentTools.collectMaterialized(table, 100);
ConfluentTools.printMaterialized(table, 100);

// On a TableResult object
TableResult tableResult = env.executeSql("SELECT * FROM examples.marketplace.customers");
List<Row> rows = ConfluentTools.collectMaterialized(tableResult, 100);
ConfluentTools.printMaterialized(tableResult, 100);

// For finite (i.e. bounded) tables
ConfluentTools.collectMaterialized(table);
ConfluentTools.printMaterialized(table);
```

### Python

```python
from pyflink.table.confluent import ConfluentSettings, ConfluentTools
from pyflink.table import TableEnvironment

settings = ConfluentSettings.from_global_variables()
env = TableEnvironment.create(settings)
# On Table object
table = env.from_path("examples.marketplace.customers")
rows = ConfluentTools.collect_materialized_limit(table, 100)
ConfluentTools.print_materialized_limit(table, 100)

# On TableResult object
tableResult = env.execute_sql("SELECT * FROM examples.marketplace.customers")
rows = ConfluentTools.collect_materialized_limit(tableResult, 100)
ConfluentTools.print_materialized_limit(tableResult, 100)

# For finite (i.e. bounded) tables
ConfluentTools.collect_materialized(table)
ConfluentTools.print_materialized(table)
```

<a id="flink-table-api-getstatementhandle"></a>

### `ConfluentTools.getStatementHandle`

Lifecycle methods are available for controlling statements on Confluent Cloud after
they have been submitted. In Java, the `ConfluentTools.getStatementHandle`
method returns a `StatementHandle` object that you can use to stop, resume,
or delete a statement, and to retrieve warnings.

By default, the `stop`, `resume`, and `delete` methods block until the
statement reaches its target phase or `client.timeout` elapses (the default
is `15min`). Pass `false` to return as soon as the request has been
accepted, instead of waiting for it to take effect. The same applies to the
`ConfluentTools.stopStatement`, `ConfluentTools.resumeStatement`, and
`ConfluentTools.deleteStatement` methods.

### Java

```java
// From a TableResult object
TableResult tableResult = env.executeSql("SELECT * FROM examples.marketplace.customers");
StatementHandle handle = ConfluentTools.getStatementHandle(tableResult);

// From a statement name
StatementHandle handle2 = ConfluentTools.getStatementHandle(env, "my-table-program-query1");

// Get the statement name
String name = handle.getName();

// Stop the statement execution, blocking until it reaches STOPPED
handle.stop();

// Resume a previously stopped statement, without waiting for it to resume
handle.resume(false);

// Delete the statement entirely from the system
handle.delete();

// Retrieve warnings associated with this statement
List<StatementWarning> warnings = handle.getWarnings();
for (StatementWarning warning : warnings) {
    System.out.println(warning.getSeverity() + ": " + warning.getMessage());
}

// Get the raw OpenAPI SqlV1Statement response for detailed information
SqlV1Statement sqlStatement = handle.getSqlV1Statement();
```

### Python

```python
# On TableResult object
table_result = env.execute_sql("SELECT * FROM examples.marketplace.customers")
statement_name = ConfluentTools.get_statement_name(table_result)
ConfluentTools.stop_statement(table_result)

# Based on statement name
ConfluentTools.stop_statement_by_name(env, "table-api-2024-03-21-150457-36e0dbb2e366-sql")
```

To manage statement lifecycle from a built JAR file or a CI/CD pipeline, see
[Deploy and Manage Table API Programs on Confluent Cloud for Apache Flink](../operate-and-deploy/table-api-deploy.md#flink-table-api-deploy).

### `ConfluentTools.createArtifact`

The `ConfluentTools.createArtifact` method packages one or more user-defined
function classes, and their transitive dependencies, into a single JAR. It
uploads the JAR to Confluent Cloud as an artifact and returns a
`ConfluentArtifact`.

Use `ConfluentArtifact.getReference()` to get a
`confluent-artifact://<id>/<version>` reference for a
`CREATE FUNCTION ... USING JAR` statement. To delete the artifact later,
pass `ConfluentArtifact.getId()` (for example, `cfa-abc123`) to
`ConfluentTools.deleteArtifact`.

This method requires artifact credentials: either a global API key and secret
(`client.global-api-key` and `client.global-api-secret`) or a dedicated
artifact API key and secret (`client.artifact-api-key` and
`client.artifact-api-secret`).

```java
// Package and upload a single UDF class into an artifact
ConfluentArtifact artifact = ConfluentTools.createArtifact(env, "my-udfs", MyScalarFn.class);

// Register a persistent function backed by the uploaded artifact
env.executeSql(
    "CREATE FUNCTION my_fn AS '" + MyScalarFn.class.getName() + "' "
        + "USING JAR '" + artifact.getReference() + "'");

// Clean up when the function is no longer needed
env.dropFunction("my_fn");
ConfluentTools.deleteArtifact(env, artifact.getId());
```

You can bundle UDF classes into a single artifact. The returned
reference backs them all, so shared dependencies are uploaded only once.

```java
ConfluentArtifact artifact =
    ConfluentTools.createArtifact(env, "my-udfs", MyScalarFn.class, MyOtherFn.class);
```

Customize the artifact metadata with `ConfluentArtifactOptions`, which
supports a custom `description`, a `documentationLink`, and the artifact
`defaultPath`. Any unset option falls back to a default: a generated
description, no documentation link, and a default path derived from the
provided classes (the single class’s name, or `default` when you bundle
more than one class).

```java
ConfluentArtifact artifact = ConfluentTools.createArtifact(env, "my-udfs",
    ConfluentArtifactOptions.newBuilder()
        .description("Email helper functions")
        .documentationLink("https://docs.example.com/udfs")
        .defaultPath(MyScalarFn.class.getName())
        .build(),
    MyScalarFn.class);
```

### `ConfluentTools.deleteArtifact`

The `ConfluentTools.deleteArtifact` method deletes a UDF artifact from
Confluent Cloud by its ID, for example, `cfa-abc123`. This is useful for cleaning
up artifacts that were uploaded for inline UDFs but are no longer referenced
by any statement.

This method requires artifact credentials: either a global API key and secret
(`client.global-api-key` and `client.global-api-secret`) or a dedicated
artifact API key and secret (`client.artifact-api-key` and
`client.artifact-api-secret`).

```java
ConfluentTools.deleteArtifact(env, "cfa-abc123");
```

### Confluent table descriptor

A table descriptor for creating tables located in Confluent Cloud programmatically.

Compared to the regular Flink class, the `ConfluentTableDescriptor` class adds
support for Confluent’s system columns and convenience methods for working
with Confluent tables.

The `for_managed()` method corresponds to
`TableDescriptor.for_connector("confluent")`.

### Java

```java
TableDescriptor descriptor = ConfluentTableDescriptor.forManaged()
  .schema(
    Schema.newBuilder()
      .column("i", DataTypes.INT())
      .column("s", DataTypes.INT())
      .watermark("$rowtime", $("$rowtime").minus(lit(5).seconds())) // Access $rowtime system column
      .build())
  .build();

env.createTable("t1", descriptor);
```

### Python

```python
from pyflink.table.confluent import ConfluentTableDescriptor
from pyflink.table import Schema, DataTypes
from pyflink.table.expressions import col, lit

descriptor = ConfluentTableDescriptor.for_managed() \
  .schema(
     Schema.new_builder()
       .column("i", DataTypes.INT())
       .column("s", DataTypes.INT())
       .watermark("$rowtime", col("$rowtime").minus(lit(5).seconds)) # Access $rowtime system column
       .build()) \
  .build()

env.createTable("t1", descriptor)
```

<a id="flink-table-api-limitations"></a>

<a id="flink-table-api-migrate-op"></a>

## Migrate from Open Preview

If your application was built against an Open Preview version of the plugin,
apply the following changes when you upgrade.

- Update your dependencies to the current versions, for example,
  `flink-table-api-java` version `2.3.0` and
  `confluent-flink-table-api-java-plugin` version `2.3-6`.
- The `client.context` configuration option and its `--context`
  command-line argument were renamed to `client.application-name` and
  `--application-name`. In code, use
  `ConfluentSettings.Builder.setApplicationName` instead of
  `setContextName`. The application name now also serves as a namespace
  prefix for all statement names submitted by the application.
- The reference documents statement lifecycle through the `StatementHandle`
  object. The `ConfluentTools.getStatementName(tableResult)` convenience
  method is still available; the same call with a handle is
  `ConfluentTools.getStatementHandle(tableResult).getName()`.

## Known limitations

### Unsupported by Table API plugin

The following features are not supported.

- Temporary catalog objects (including tables, views, functions)
- Custom modules
- Custom catalogs
- Anonymous, inline objects (including data types). Inline functions must be
  serializable by class name, so anonymous inner classes are rejected. Use a
  top-level class or a static nested class instead.
- Inline functions in `createView`
- CompiledPlan features
- Batch mode
- Restrictions from Confluent Cloud
  - custom connectors/formats
  - processing time operations
  - many configuration options
  - limited SQL syntax

### Statement management

- In-place modification of a running statement isn’t supported. To change the
  logic of a statement, delete and re-create it, which loses processing
  state. This behavior is the same as for Flink SQL statements.
- Statement names must contain only lowercase alphanumeric characters and
  hyphens, must start and end with an alphanumeric character, and have a
  maximum length of 100 characters, including the application name prefix.
- Lifecycle actions on a built JAR file require that the `main()` method
  parses command-line arguments with `ConfluentSettings.fromArgs` or
  `ConfluentSettings.newBuilderFromArgs`. For more information, see
  [Deploy and Manage Table API Programs on Confluent Cloud for Apache Flink](../operate-and-deploy/table-api-deploy.md#flink-table-api-deploy).
- The plugin doesn’t include a local Flink runtime, and all statements run on
  Confluent Cloud. For local testing patterns, see [Deploy and Manage Table API Programs on Confluent Cloud for Apache Flink](../operate-and-deploy/table-api-deploy.md#flink-table-api-deploy).

### Issues in Apache Flink

- Both catalog and database must be set, or identifiers must be fully
  qualified. A mixture of setting a current catalog and using two-part
  identifiers can cause errors.
- Selecting `.rowtime` in windows causes errors.
- Using `.limit()` can cause errors.

## Next steps

- [Java Table API Quick Start on Confluent Cloud for Apache Flink](../get-started/quick-start-java-table-api.md#flink-java-table-api-quick-start)
- [Python Table API Quick Start on Confluent Cloud for Apache Flink](../get-started/quick-start-python-table-api.md#flink-python-table-api-quick-start)
- [Deploy and Manage Table API Programs on Confluent Cloud for Apache Flink](../operate-and-deploy/table-api-deploy.md#flink-table-api-deploy)

## Related content

- Course: [Apache Flink® Table API: Processing Data Streams in Java](https://developer.confluent.io/courses/flink-table-api-java/exercise-connecting-to-confluent-cloud/)
- [Table API functions](functions/table-api-functions.md#flink-table-api-functions)
- [Built-in Functions](functions/overview.md#flink-sql-functions-overview)

#### 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).