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

# 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](../reference/table-api.md#flink-table-api). For runnable
examples, including CI/CD workflow templates, see the
[Java Examples for Table API on Confluent Cloud](https://github.com/confluentinc/flink-table-api-java-examples)
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](../../security/authenticate/workload-identities/service-accounts/api-keys/overview.md#cloud-global-api-keys), 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](../../security/authenticate/workload-identities/service-accounts/overview.md#service-accounts). 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](../../security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.md#create-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.

```bash
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](../reference/table-api.md#flink-table-api-authentication) in the Table API
reference.

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

## 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`:
  ```java
  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](../reference/table-api.md#flink-table-api-getstatementhandle)
in the Table API reference.

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

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

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

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

```bash
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](/cloud/current/api.html#tag/Statements-(sqlv1)).
- `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.

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

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

1. Build and unit-test the project, with no Confluent Cloud credentials required.
2. Run integration tests against a development or staging environment.
3. 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](https://github.com/confluentinc/flink-table-api-java-examples#cicd-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](../get-started/quick-start-java-table-api.md#flink-java-table-api-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.

1. **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.
2. **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.
3. **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)](../how-to-guides/create-ptf.md#flink-ptfs-quickstart)
locally, without a Confluent Cloud connection, use the
`ProcessTableFunctionTestHarness`, which ships with the plugin. For more
information, see [Testing Process Table Functions](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/ptfs/#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](https://github.com/confluentinc/flink-table-api-java-examples#testing-table-programs)
in the examples repository.

## 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)
- [Table API reference](../reference/table-api.md#flink-table-api)
- [Flink SQL Development Lifecycle in Confluent Cloud for Apache Flink](development-lifecycle.md#flink-development-lifecycle)

## Related content

- [Java Examples for Table API on Confluent Cloud](https://github.com/confluentinc/flink-table-api-java-examples)
- [Table API functions](../reference/functions/table-api-functions.md#flink-table-api-functions)
- [Flink SQL REST API for Confluent Cloud for Apache Flink](flink-rest-api.md#flink-rest-api)

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