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.

For a list of Table API functions supported by Confluent Cloud for Apache Flink, see Table API functions.

For a list of Table API limitations in Confluent Cloud for Apache Flink, see Migrate from Open Preview.

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.

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.

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.

<!-- 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>

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

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

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.
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.
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`.

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.

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

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.

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.

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.

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.

// 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

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

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.

# 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

// 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

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:

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

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

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.

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

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

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.

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.

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

ConfluentSettings settings = ConfluentSettings.fromGlobalVariables();

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

// 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

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

// 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

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)

`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

// 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

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

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

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

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

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)

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-3.
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.
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.

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.

Table API on Confluent Cloud for Apache Flink

Add the Table API to an existing Java project

Configure the plugin

Authentication

API keys (default)

OAuth

Endpoint configuration

client.endpoint-template

client.artifact-endpoint-template

client.rest-endpoint (Deprecated)

Relationship and default behavior

ConfluentSettings class

Properties file

Command-line arguments

Code

Environment variables

Confluent utilities

ConfluentTools.collectChangelog and ConfluentTools.printChangelog

ConfluentTools.collect_materialized and ConfluentTools.print_materialized

ConfluentTools.getStatementHandle

ConfluentTools.deleteArtifact

Confluent table descriptor

Migrate from Open Preview

Known limitations

Unsupported by Table API plugin

Statement management

Issues in Apache Flink

Next steps

Related content

`ConfluentSettings` class

`ConfluentTools.collectChangelog` and `ConfluentTools.printChangelog`

`ConfluentTools.collect_materialized` and `ConfluentTools.print_materialized`

`ConfluentTools.getStatementHandle`

`ConfluentTools.deleteArtifact`