Deploy and Manage Table API Programs on Confluent Cloud for Apache Flink

A Table API program on Confluent Cloud for Apache Flink® is a regular Java application that you build, test, and deploy with your own tools, like Maven, JUnit, and your CI/CD system. The Confluent plugin submits the statements that your program defines to Confluent Cloud and provides lifecycle actions for managing them, so the same JAR that deploys your pipeline also stops, resumes, deletes, and inspects it.

This page shows how to authenticate your application, control statement names, manage statement lifecycle from code and from the command line, and integrate a Table API program with a CI/CD pipeline.

For an introduction to the Table API and the full list of configuration options, see the Table API reference. For runnable examples, including CI/CD workflow templates, see the Java Examples for Table API on Confluent Cloud repository.

Note

This page applies to the Table API for Java, which is generally available. The Table API for Python is available for preview and doesn’t support the lifecycle actions described on this page.

Authenticate your application

The recommended way to authenticate a Table API application with Confluent Cloud is a global API key, which works for both Flink access and artifact creation, so a single key and secret pair covers statement submission and UDF uploads. For production deployments, create the key for a service account. Only the OrganizationAdmin role, or ResourceOwner on the service account, can create global API keys for a service account. For the creation procedure, see Add an API key.

Set the global key and secret with the client.global-api-key and client.global-api-secret options, the --global-api-key and --global-api-secret command-line arguments, or the GLOBAL_API_KEY and GLOBAL_API_SECRET environment variables.

export CLOUD_PROVIDER="aws"
export CLOUD_REGION="us-east-1"
export GLOBAL_API_KEY="<your-global-api-key>"
export GLOBAL_API_SECRET="<your-global-api-secret>"
export ORG_ID="<your-organization-id>"
export ENV_ID="<your-environment-id>"
export COMPUTE_POOL_ID="<your-compute-pool-id>"

java -jar my-table-program.jar

As an alternative, you can configure service-specific keys: a Flink API key (client.flink-api-key and client.flink-api-secret) for statement submission and, if your program uploads UDF artifacts, an artifact API key (client.artifact-api-key and client.artifact-api-secret). If both global and service-specific keys are configured, the plugin uses the service-specific keys.

For workloads that authenticate with an identity provider instead of API keys, the plugin supports OAuth 2.0 client-credentials and static-token modes, including cloud-native flows like Azure Managed Identity and AWS IAM workload identity. For more information, see Authentication in the Table API reference.

Name applications and statements

By default, statement names are generated with a UUID, which means that a redeployed program creates new statements instead of addressing the ones it created before. For any program that you deploy repeatedly, configure an application name and explicit statement names, so names are deterministic across runs.

The application name (client.application-name, --application-name, or APPLICATION_NAME) acts as a namespace: it’s prefixed automatically to every statement name the program submits. For example, application name marketplace-analytics and statement name vendors-per-brand produce the statement marketplace-analytics-vendors-per-brand.
For a single-statement program, set the statement name with client.statement-name, --statement-name, or STATEMENT_NAME.
For a program that submits multiple statements, set a new name before each submission with ConfluentTools.setStatementName:
```
ConfluentTools.setStatementName(env, "vendors-per-brand");
env.executeSql("...");

ConfluentTools.setStatementName(env, "orders-per-region");
env.executeSql("...");
```
The name that you set with ConfluentTools.setStatementName applies only to the next statement submission and is reset automatically afterward, so set a new name before every submission.

Statement names must contain only lowercase alphanumeric characters and hyphens, must start and end with an alphanumeric character, and are limited to 100 characters, including the application name prefix. The plugin logs a warning when a background statement is submitted without an explicit name.

Manage statement lifecycle in your program

Within a running program, the ConfluentTools.getStatementHandle method returns a StatementHandle that can stop, resume, and delete a statement and retrieve its warnings. For more information, see ConfluentTools.getStatementHandle in the Table API reference.

TableResult tableResult = env.executeSql("SELECT * FROM examples.marketplace.customers");
StatementHandle handle = ConfluentTools.getStatementHandle(tableResult);

handle.stop();
handle.resume();
handle.delete();

Manage statement lifecycle from the built JAR

The plugin supports lifecycle actions that you invoke by running your application JAR with an action as the first argument. This enables managing statements from deployment scripts without writing SQL, installing the Confluent CLI, or calling the REST API directly.

java -jar my-table-program.jar <action> [options]

The action is one of the following.

list: lists statements belonging to the application, or a specific statement.
describe: shows the full JSON representation of a specific statement, as returned by the Statements endpoint.
resume: resumes a stopped statement.
stop: stops a running statement.
delete: deletes a statement entirely from the system.

Two informational flags are also available and don’t require any other configuration:

--help (or -h): prints all supported command-line arguments, with their descriptions, and the available lifecycle actions, then exits.
--version: prints the plugin version, then exits.

To use lifecycle actions, your application’s main() method must parse command-line arguments with ConfluentSettings.fromArgs(args) or ConfluentSettings.newBuilderFromArgs(args). To layer command-line arguments on top of a base configuration, such as a properties file, call applyArgs(args) on an existing builder instead. When an action is detected, the plugin executes it and terminates the JVM, so the rest of your application logic doesn’t run. Running the JAR without an action deploys the program normally.

public static void main(String[] args) {
  // Automatically detects and executes lifecycle actions, if present.
  // If an action is specified, the program exits here.
  EnvironmentSettings settings = ConfluentSettings.fromArgs(args);
  TableEnvironment env = TableEnvironment.create(settings);

  // Your application logic here (runs only if no action was specified).
}

The describe, resume, stop, and delete actions require a statement name, provided with --statement-name, the STATEMENT_NAME environment variable, or client.statement-name in a properties file. If an application name is configured, it’s prefixed automatically to the statement name.

The list action doesn’t require a statement name. With an application name configured, it lists all statements matching that application prefix. Without an application name, it lists all statements in the environment, which can be expensive in shared environments, so configuring an application name is recommended for CI/CD usage.

The following example stops a statement.

java -jar marketplace-analytics.jar stop \
  --statement-name vendors-per-brand \
  --application-name marketplace-analytics \
  --cloud aws \
  --region us-east-1 \
  --organization-id b0b21724-4586-4a07-b787-d0bb5aacbf87 \
  --environment-id env-z3y2x1 \
  --compute-pool-id lfcp-8m03rm \
  --global-api-key <key> \
  --global-api-secret <secret>

Wait for completion

By default, resume, stop, and delete return as soon as the Confluent Cloud API has accepted the request, while the statement might still be transitioning in the background. For pipelines where the next step depends on the new phase being reached, pass --action.await to block until the action has fully taken effect.

resume waits until the statement reaches the RUNNING phase.
stop waits until the statement reaches the STOPPED phase.
delete waits until the statement doesn’t exist.

The default timeout is 15 minutes. Pass a duration to override it, for example, --action.await 10min. If the target phase isn’t reached before the timeout elapses, the action fails with exit code 1. The list and describe actions ignore --action.await.

Actions exit with code 0 on success and code 1 on failure, for example, when the statement isn’t found, permission is denied, configuration is missing, or the timeout elapses, so failures propagate naturally in CI/CD pipelines.

Handle conflicting deployments

When a statement with the same name already exists with a different specification, the plugin’s default behavior (client.on-conflict=fail) is to fail the submission with an exception. To redeploy over the existing statement instead, set client.on-conflict=replace or pass --on-conflict replace: the conflicting statement is deleted, and the submission is retried once. The replace setting requires an application name, so statement names are stable across runs and a redeploy targets the same name.

Re-running an unchanged program is idempotent: statements whose specification hasn’t changed remain untouched.

Important

--on-conflict replace deletes the existing statement and submits a new one. The new statement starts from its configured source offsets and doesn’t resume the previous statement’s state. For stateless pipelines, like filters, projections, and routing, this has no effect on results. For stateful pipelines, like aggregations, joins, and deduplication, the new statement rebuilds its state by reprocessing from the configured start position, so choose the redeploy timing and the source startup mode accordingly.

Integrate with a CI/CD pipeline

Because deployment is running main() and lifecycle management is running the same JAR with an action argument, a Table API program fits any CI/CD system without extra tooling. A typical pipeline performs the following steps.

Build and unit-test the project, with no Confluent Cloud credentials required.
Run integration tests against a development or staging environment.
Deploy by running the JAR with --statement-name, --application-name, and --on-conflict replace, provided as deployment configuration by the pipeline rather than hardcoded in the program.

Statement and application names passed by the pipeline make deployments idempotent and addressable: the same names are used for deployment and for later stop, resume, describe, or delete operations. Promotion from staging to production is a matter of running the same deploy job against a different environment with its own credentials and configuration.

The examples repository contains GitHub Actions workflow templates that implement this pattern, including a deploy workflow and a lifecycle management workflow. For more information, see CI/CD Integration in the examples repository.

Diagnose deployments with plugin logs

The plugin logs its operations, such as statement submission and artifact upload, through SLF4J on the io.confluent.flink.plugin logger. If a lifecycle action or deployment is slow or appears to hang, set this logger to DEBUG to surface operation breadcrumbs and status-polling detail. The plugin doesn’t ship a logging binding, so your application must provide one. The Java quick start includes the required log4j dependencies.

Test your application

The plugin executes all statements on Confluent Cloud and doesn’t include a local Flink runtime. Table programs can still be tested locally in tiers, from fastest feedback to highest fidelity.

Unit tests on plain logic. UDFs and other business logic are plain Java classes and can be tested with JUnit alone, with no Flink or Confluent Cloud connectivity required.
Local pipeline tests on |af-oss|. Pipeline logic that is structured as a function from input Table objects to an output Table can run locally on the open source Flink planner and runtime, added as test-scoped dependencies, with mock data from fromValues(). Keep pipeline logic free of io.confluent.flink.plugin imports, so unit tests can execute it locally. Local runs don’t have Confluent Cloud system columns, like $rowtime, catalogs, or Confluent-specific SQL syntax.
Integration tests against Confluent Cloud. The same pipeline logic runs on the real service with exact Confluent Cloud semantics, for example, as Maven failsafe tests that require connection credentials.

To unit-test a process table function (PTF) locally, without a Confluent Cloud connection, use the ProcessTableFunctionTestHarness, which ships with the plugin. For more information, see Testing Process Table Functions.

The examples repository shows the three test tiers and the required Maven configuration, including how to keep the plugin and the local Flink planner from conflicting on the test classpath. For more information, see Testing Table Programs in the examples repository.