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

# Flink SQL REST API for Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® provides a REST API for managing your Flink SQL statements,
materialized tables, compute pools, and connections programmatically.

Use the REST API to manage these features:

- [Artifacts (user-defined functions)](#flink-rest-api-manage-artifacts)
- [Compute pools](#flink-rest-api-manage-compute-pools)
- [Connections](#flink-rest-api-manage-connections)
- [List available regions](#flink-rest-api-list-regions)
- [Materialized tables](#flink-rest-api-manage-materialized-tables)
- [Statements](#flink-rest-api-manage-statements)
- [Workspaces](#flink-rest-api-manage-workspaces)

For the complete Flink REST API reference, see:

- [Artifacts (user-defined functions)](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1))
- [Compute Pools](/cloud/current/api.html#tag/Compute-Pools-(fcpmv2))
- [Connections](/cloud/current/api.html#tag/Connections-(sqlv1))
- [Materialized Tables](/cloud/current/api.html#tag/Materialized-Tables-(sqlv1))
- [Regions](/cloud/current/api.html#tag/Regions-(fcpmv2))
- [Statements](/cloud/current/api.html#tag/Statements-(sqlv1))
- [Workspaces](/cloud/current/api.html#tag/Workspaces-(ws/v1))

In addition to the REST API, you can manage Flink resources by using these
Confluent tools:

- [Cloud Console](../get-started/quick-start-cloud-console.md#flink-sql-quick-start-run-sql-statement)
- [Confluent CLI](../reference/flink-sql-cli.md#flink-sql-confluent-cli)
- [SQL shell](../get-started/quick-start-shell.md#flink-sql-quick-start-shell)
- [Confluent Terraform Provider](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs)

## Prerequisites

To manage Flink resources by using the REST API, you need Confluent Cloud account
details, like your organization and environment identifiers, and you must choose
an authentication method.

### Authentication methods

- **OAuth authentication (preferred for production):** Use tokens from your
  identity provider. For more information, see [OAuth authentication](#flink-rest-api-oauth).
- **API key authentication:** Generate a Flink API key and secret. For more
  information, see [Generate a Flink API key](#flink-rest-api-generate-api-key).

### Account details

- **Organization ID:** The identifier of your organization, for example,
  `b0b21724-4586-4a07-b787-d0bb5aacbf87`.
- **Environment ID:** The identifier of the environment where your Flink SQL
  statements run, for example, “env-z3y2x1”.
- **Cloud provider name:** The name of the cloud provider where your cluster
  runs, for example, “AWS”. To see the available providers, run the
  `confluent flink region list` command.
- **Cloud region:** The name of the region where your cluster runs, for
  example, “us-east-1”. To see the available regions, run the
  `confluent flink region list` command.

Depending on the request, you might need these details:

- **OAuth token:** If using OAuth authentication, you need an OAuth token from
  your identity provider, an OAuth identity pool ID such as `pool-W5Qe`,
  and a principal, which can be either the same pool ID or a service account ID.
  For more information, see [OAuth authentication](#flink-rest-api-oauth).
- **Cloud API key:** Some requests require a Confluent Cloud API key and secret, which
  are distinct from a Flink API key and secret. To create a new API key for
  Confluent Cloud, see [Cloud API keys](../../security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.md#cloud-cloud-api-keys). On the
  [https://confluent.cloud/settings/api-keys](https://confluent.cloud/settings/api-keys)
  page, select the **Cloud resource management** tile for the API key’s resource
  scope.
- **Principal ID:** The identifier of your user account or a service account,
  for example, “u-aq1dr2” for a user account or “sa-23kgz4” for a service
  account.
- **Compute pool ID:** The identifier of the compute pool that runs your
  Flink SQL statements, for example, “lfcp-8m03rm”.
- **Statement name:** A unique name for a Flink SQL statement.
- **SQL code:** The code for a Flink SQL statement.

## Rate limits

Requests to the Flink REST API are rate-limited per IP address.

- Concurrent connections: 100
- Requests per minute: 1000
- Requests per second: 50

## Private networking endpoints

If you have enabled Flink [private networking](../concepts/flink-private-networking.md#flink-sql-private-networking),
the REST endpoints are different.

```html
<!-- Without private network -->
https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/

<!-- With private network -->
https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.private.confluent.cloud/
```

For example, if you send a request to the `us-east-1` AWS region without
a private network, the host is:

```html
<!-- Without private network -->
https://flink.us-east-1.aws.confluent.cloud
```

With a private network, the host is:

```html
<!-- With private network -->
https://flink.us-east-1.aws.private.confluent.cloud
```

<a id="flink-rest-api-generate-api-key"></a>

## Generate a Flink API key

To access the REST API with basic authentication, you need an API key
specifically for Flink. This key is distinct from the Confluent Cloud API key. If
you’re using OAuth authentication, you can skip this step.

Before you create an API key for Flink access, decide whether you want to
create long-running Flink SQL statements. For long-running statements,
use a service account and create an API key for it. If you want to run
only interactive queries or statements for a short time while developing
queries, you can create an API key for your user account.

- Follow the steps in [Generate an API Key for Access](generate-api-key-for-flink.md#flink-generate-api-key).

Run the following commands to save your API key and secret in environment
variables.

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
```

If you’re using basic authentication, you need to create a base64-encoded
string from your API key and secret. For details, see
[Basic authentication with API keys](#flink-rest-api-basic-auth).

## Authentication methods

The Flink REST API supports two authentication methods: OAuth tokens and API keys
with basic authentication.

<a id="flink-rest-api-oauth"></a>

### OAuth authentication

Confluent Cloud for Apache Flink supports OAuth authentication for Flink REST API endpoints as an
alternative to API keys. This enables you to use tokens from your identity
provider instead of managing API keys.

#### NOTE
This section covers control plane authentication, which is authenticating to
Flink REST APIs to manage statements and compute pools. It does not cover data
plane connectivity, where Flink jobs connect to Kafka topics.

If you build applications with the Java Table API, the plugin implements the
OAuth flows described in this section for you, including token refresh and
the `Confluent-Identity-Pool-Id` header, by using the `client.auth-mode`
configuration option. For more information, see
[Table API authentication](../reference/table-api.md#flink-table-api-authentication).

#### OAuth support by API type

- `sqlv1` (Statement APIs): Full OAuth support using service account delegation.
- `sqlv1` (Materialized Table APIs): Full OAuth support using service account delegation.
- `fcpmv2` (Compute Pool APIs): OAuth support using STS token exchange.
- `artifactv1` (Artifact APIs): OAuth support using STS token exchange.
- `sqlv1` (Connection APIs): OAuth is not supported. Use Flink API keys.

#### OAuth for Statement APIs

Statement APIs (`sqlv1`) support OAuth authentication with two patterns:

- Identity pool as principal: The identity pool authenticates with your identity provider generated `jwt-token` and executes
  the statement with access granted to the identity pool.
- Service account delegation: The identity pool authenticates with your identity provider generated `jwt-token` and the specified service
  account executes the statement with access granted to the service account.

##### Prerequisites

Before using OAuth authentication with Flink:

1. Create an OAuth identity provider. For more information, see
   [Configure OAuth/OIDC identity providers](/cloud/current/security/authenticate/workload-identities/identity-providers/oauth/identity-providers.html).
2. Create an identity pool with appropriate claim filters. For more information,
   see [Use OAuth identity pools](/cloud/current/security/authenticate/workload-identities/identity-providers/oauth/identity-pools.html).

##### Pattern 1: Identity pool as principal

In this pattern, the identity pool both authenticates the request and executes
the statement.

Grant the identity pool `FlinkDeveloper` permissions:

```bash
confluent iam rbac role-binding create \
  --principal "User:pool-W5Qe" \
  --role FlinkDeveloper \
  --environment env-12345
```

Submit a statement using the identity pool:

```bash
# 1. Get OAuth token from your identity provider
export OAUTH_TOKEN=$(curl -X POST https://your-idp.example.com/oauth2/token \
  -d "grant_type=client_credentials" \
  -d "client_id=YOUR_CLIENT_ID" \
  -d "client_secret=YOUR_CLIENT_SECRET" \
  | jq -r '.access_token')

# 2. Submit statement with identity pool as principal
curl --request POST \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements" \
  --header "Authorization: Bearer ${OAUTH_TOKEN}" \
  --header "Confluent-Identity-Pool-Id: pool-W5Qe" \
  --header "content-type: application/json" \
  --data '{
    "name": "oauth-example",
    "spec": {
      "statement": "SELECT * FROM my_table;",
      "compute_pool_id": "lfcp-12345"
    }
  }'
```

The required components are:

- Authorization header: Bearer token from your identity provider.
- Confluent-Identity-Pool-Id header: Your identity pool ID, such as
  `pool-W5Qe`. You can omit this header to use auto pool mapping, or
  specify a comma-separated list of identity pool IDs. For details, see
  [Use OAuth identity pools](/cloud/current/security/authenticate/workload-identities/identity-providers/oauth/identity-pools.html).
- spec.principal: If omitted, identity pool ID in the header is automatically used.

##### Pattern 2: Service account delegation

In this pattern, the identity pool authenticates the request while a service
account executes the statement. This provides separation between authentication
and execution, making it easier to manage permissions if your organization is already using service accounts.

#### IMPORTANT
The identity pool being authenticated must have the **Assigner** role on the service account.
Without this role binding, API requests fail with authorization errors.

Create a service account and grant required RBAC roles:

```bash
# Grant Assigner role (pool → service account)
confluent iam rbac role-binding create \
  --principal "User:pool-W5Qe" \
  --resource "service-account:sa-xyz123" \
  --role Assigner

# Grant Flink permissions (service account → environment)
confluent iam rbac role-binding create \
  --principal "User:sa-xyz123" \
  --role FlinkDeveloper \
  --environment env-12345
```

For more information about Flink RBAC roles, see [Grant Role-Based Access in Confluent Cloud for Apache Flink](flink-rbac.md#flink-rbac).

Submit a statement using service account delegation:

```bash
# 1. Get OAuth token from your identity provider
export OAUTH_TOKEN=$(curl -X POST https://your-idp.example.com/oauth2/token \
  -d "grant_type=client_credentials" \
  -d "client_id=YOUR_CLIENT_ID" \
  -d "client_secret=YOUR_CLIENT_SECRET" \
  | jq -r '.access_token')

# 2. Submit statement with three required auth components
curl --request POST \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements" \
  --header "Authorization: Bearer ${OAUTH_TOKEN}" \
  --header "Confluent-Identity-Pool-Id: pool-W5Qe" \
  --header "content-type: application/json" \
  --data '{
    "name": "oauth-example",
    "spec": {
      "statement": "SELECT * FROM my_table;",
      "principal": "sa-xyz123",
      "compute_pool_id": "lfcp-12345"
    }
  }'
```

The three required components are:

- Authorization header: Bearer token from your identity provider.
- Confluent-Identity-Pool-Id header: Your identity pool ID. You can omit
  this header to use auto pool mapping, or specify a comma-separated list
  of identity pool IDs. For details, see
  [Use OAuth identity pools](/cloud/current/security/authenticate/workload-identities/identity-providers/oauth/identity-pools.html).
- spec.principal: Service account that executes the statement.

##### List and manage statements

List all statements in an environment:

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements" \
  --header "Authorization: Bearer ${OAUTH_TOKEN}" \
  --header "Confluent-Identity-Pool-Id: ${POOL_ID}"
```

Get statement results:

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}/results" \
  --header "Authorization: Bearer ${OAUTH_TOKEN}" \
  --header "Confluent-Identity-Pool-Id: ${POOL_ID}"
```

Get statement exceptions:

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}/exceptions" \
  --header "Authorization: Bearer ${OAUTH_TOKEN}" \
  --header "Confluent-Identity-Pool-Id: ${POOL_ID}"
```

##### Troubleshooting

**401 Unauthorized**
: The OAuth token is invalid or expired. Verify token generation from your
  identity provider.

**403 Forbidden: “You don’t have the required permissions”**
: The identity pool does not have the Assigner role on the service account.
  Verify the role binding:
  <br/>
  ```bash
  confluent iam rbac role-binding list \
    --principal "User:pool-XXXX" \
    --role Assigner
  ```

**400 Bad Request: “cannot issue an OAuth request on behalf of another identity pool”**
: The pool ID in the `Confluent-Identity-Pool-Id` header does not match the
  pool ID in `spec.principal`. When using identity pool as principal, both
  must be the same pool.

**400 Bad Request: “cannot issue an OAuth request with user principal”**
: You cannot use a user ID in `spec.principal` when authenticating with OAuth.
  Use either an identity pool ID or a service account ID.

**Statement fails with “Service account does not have enough permissions”**
: The service account does not have `FlinkDeveloper` or `FlinkAdmin` role. Grant
  the appropriate role at the environment or compute pool level:
  <br/>
  ```bash
  confluent iam rbac role-binding create \
    --principal "User:sa-YYYY" \
    --role FlinkDeveloper \
    --environment env-XXXXX
  ```

#### OAuth for Compute Pool and Artifact APIs

Compute Pool APIs (fcpmv2) and Artifact APIs (artifactv1) support OAuth
authentication but require an additional token exchange step. You must exchange
your identity provider OAuth token for a Confluent token using the STS
(Security Token Service) endpoint.

To authenticate with OAuth for Compute Pool and Artifact APIs:

1. Obtain an OAuth token from your identity provider.
2. Exchange the token for a Confluent token using the
   [STS endpoint](/cloud/current/api.html#tag/OAuth-Tokens-(stsv1)/operation/exchangeStsV1OauthToken).
3. Use the Confluent token as a Bearer token in the `Authorization` header
   in API requests (for example, `Authorization: Bearer <confluent-token>`).

For more information about the STS endpoint and token exchange, see
[OAuth Tokens (stsv1)](/cloud/current/api.html#tag/OAuth-Tokens-(stsv1)).

<a id="flink-rest-api-basic-auth"></a>

### Basic authentication with API keys

As an alternative to OAuth, you can use basic authentication with Flink API keys.
Basic authentication means you provide a base64-encoded string made from your
Flink API key and secret in the `Authorization` header of your requests.

You can use the `base64` command to encode the “key:secret” string. Be
sure to use the `-n` option of the `echo` command to prevent embedding
newlines in the encoded string. If you’re on Linux, be sure to use the
`-w 0` option of the `base64` command to prevent line-wrapping the
string.

For convenience, save the encoded string in an environment variable:

```bash
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
```

Use this environment variable in the `Authorization` header of your API
requests, as shown in the examples throughout this page.

<a id="flink-rest-api-versions"></a>

## API versions

The Flink REST API uses version identifiers in its URL paths, for example,
`sql/v1`, `fcpm/v2`, and `ws/v1`. Most Flink API groups have a single,
generally available version. The Statements API (`sql/*`) is the
exception. It has three versions that reflect its progression from early
access to general availability: `v1`, `v1beta1`, and `v1alpha1`.

- `v1` (**General Availability**): The current, stable version of the
  Statements API, and the version used throughout this page. Use `v1` for
  all new integrations. Requests use the path prefix
  `/sql/v1/organizations/{organization_id}/environments/{environment_id}/`,
  and the API supports the full set of statement operations, including
  submitting, listing, updating, patching, and deleting statements.
- `v1beta1` (**Open Preview**): An earlier, statement-only version of the
  API. It doesn’t support connections, materialized tables, or agents.
  Access to `v1beta1` isn’t self-service; requests without granted access
  return `404 Not Found`. Contact your Confluent account team if an
  existing integration depends on it.
- `v1alpha1` (**Early Access**): The earliest version of the Statements
  API. It predates the `organizations` path segment, so its URLs have the
  form `/sql/v1alpha1/environments/{environment_id}/...` instead. Access
  to `v1alpha1` isn’t self-service, and it isn’t recommended for new
  integrations.

If an existing integration uses `v1beta1` or `v1alpha1`, migrate it to
`v1` when possible. There’s no automatic upgrade path between versions;
you must update the client to use the `v1` URL paths and request access
if you don’t already have it.

For more information about API lifecycle stages and the Confluent Cloud API
compatibility policy, see
[API Lifecycle Policy](/cloud/current/api.html#section/Versioning/API-Lifecycle-Policy).

<a id="flink-rest-api-manage-statements"></a>

## Manage statements

Using requests to the Flink REST API, you can perform these actions:

- [Submit a statement](#flink-rest-api-submit-statement)
- [Get a statement](#flink-rest-api-get-statement)
- [List statements](#flink-rest-api-list-statements)
- [Update metadata for a statement](#flink-rest-api-update-statement)
- [Patch a statement](#flink-rest-api-patch-statement)
- [Delete a statement](#flink-rest-api-delete-statement)

### Flink SQL statement schema

A statement has the following schema:

```yaml
api_version: "sql/v1"
kind: "Statement"
organization_id: "" # Identifier of your Confluent Cloud organization
environment_id: "" # Identifier of your Confluent Cloud environment
name: "" # Primary identifier of the statement, must be unique within the environment, 100 max length, [a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*
metadata:
  created_at: "" # Creation timestamp of this resource
  updated_at: "" # Last updated timestamp of this resource
  resource_version: "" # Generated by the system and updated whenever the statement is updated (including by the system). Opaque and should not be parsed.
  self: "" # An absolute URL to this resource
  uid: "" # uid is unique in time and space (i.e., even if the name is re-used)
spec:
  compute_pool_id: "" # Optional. The ID of the compute pool the statement should run in. If not specified, uses the default compute pool. DNS Subdomain (RFC 1123) – 255 max len, [a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*
  principal: "" # user or service account ID
  properties: map[string]string # Optional. request/client properties
  statement: "SELECT * from Orders;" # The raw SQL text
  stopped: false # Boolean, specifying if the statement should be stopped
  execution_mode: "streaming" # Read-only. The execution mode of the statement (streaming or batch)
status:
  phase: PENDING | RUNNING | COMPLETED | DELETING | FAILING | FAILED
  detail: "" # Optional. Human-readable description of phase.
  scaling_status:
    scaling_state: "" # Scaling state (for example, OK, SCALING_UP, SCALING_DOWN)
    last_updated: "" # Timestamp of last scaling status update
  traits:
    sql_kind: "" # Optional. The kind of SQL statement (for example, SELECT, INSERT, CREATE_TABLE)
    is_bounded: false # Optional. Whether the statement is bounded (batch) or unbounded (streaming)
    is_append_only: false # Optional. Whether the statement is append-only
    schema:
      columns: [] # Optional. JSON object in TableSchema format; describes the data returned by the results serving API
  network_kind: "" # Network kind (for example, PUBLIC, PRIVATE)
```

The statement name has a maximum length of 100 characters and must satisfy
the following regular expression:

```none
[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*
```

The underscore character (`_`) and period character (`.`) are not supported.

<a id="flink-rest-api-submit-statement"></a>

### Submit a statement

You can submit a Flink SQL statement by sending a POST request to the
[Statements endpoint](/cloud/current/api.html#tag/Statements-(sqlv1)/operation/createSqlv1Statement).

Submitting a Flink SQL statement requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export STATEMENT_NAME="<statement-name>" # example: "user-filter"
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export COMPUTE_POOL_ID="<compute-pool-id>" # (optional) example: "lfcp-8m03rm"
export PRINCIPAL_ID="<principal-id>" # (optional) example: "sa-23kgz4" for a service account, or "u-aq1dr2" for a user account
export SQL_CODE="<sql-statement-text>" # example: "SELECT * FROM USERS;"
export JSON_DATA="<payload-string>"
```

The COMPUTE_POOL_ID and PRINCIPAL_ID parameters are optional. If you don’t
specify COMPUTE_POOL_ID, the statement uses the default compute pool.
Confluent Cloud infers the principal from the provided Flink API key if you don’t
specify PRINCIPAL_ID.

The following JSON shows an example payload:

```json
{
  "name": "${STATEMENT_NAME}",
  "organization_id": "${ORG_ID}",
  "environment_id": "${ENV_ID}",
  "spec": {
    "statement": "${SQL_CODE}",
    "properties": {
      "key1": "value1",
      "key2": "value2"
    },
    "compute_pool_id": "${COMPUTE_POOL_ID}",
    "principal": "${PRINCIPAL_ID}",
    "stopped": false
  }
}
```

You must escape quotation mark characters in the JSON string, so the payload
string to send resembles the following:

```bash
export JSON_DATA="{
  \"name\": \"${STATEMENT_NAME}\",
  \"organization_id\": \"${ORG_ID}\",
  \"environment_id\": \"${ENV_ID}\",
  \"spec\": {
    \"statement\": \"${SQL_CODE}\",
    \"properties\": {
      \"key1\": \"value1\",
      \"key2\": \"value2\"
    },
    \"compute_pool_id\": \"${COMPUTE_POOL_ID}\",
    \"principal\": \"${PRINCIPAL_ID}\",
    \"stopped\": false
  }
}"
```

The following command sends a POST request that submits a Flink SQL
statement.

```bash
curl --request POST \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to submit a SQL statement
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "Statement",
  "metadata": {
    "self": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/statements/demo-statement-1",
    "created_at": "2023-12-16T17:12:08.914198Z",
    "updated_at": "2023-12-16T17:12:08.914198Z",
    "uid": "0005dd7b-8a7e-4274-b97e-c21b134d98f0",
    "resource_version": "1"
  },
  "name": "demo-statement-1",
  "organization_id": "b0b21724-4586-4a07-b787-d0bb5aacbf87",
  "environment_id": "env-z3y2x1",
  "spec": {
    "statement": "select 1;",
    "properties": {},
    "compute_pool_id": "lfcp-8m03rm",
    "principal": "u-aq1dr2",
    "stopped": false
  },
  "status": {
    "phase": "PENDING",
    "detail": "Statement is being submitted"
  }
}
```

</details>

<a id="flink-rest-api-get-statement"></a>

### Get a statement

Get the details about a Flink SQL statement by sending a GET request to the
[Statements endpoint](/cloud/current/api.html#tag/Statements-(sqlv1)/operation/getSqlv1Statement).

Getting a Flink SQL statement requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export STATEMENT_NAME="<statement-name>" # example: "user-filter"
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
```

The following command gets a Flink SQL statement’s details by its name.
Attempting to get a deleted statement returns `404`.

```bash
curl --request GET \
 --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}" \
 --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to get a SQL statement
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "Statement",
  "metadata": {
    "self": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/statements/demo-statement-1",
    "created_at": "2023-12-16T16:08:36.650591Z",
    "updated_at": "2023-12-16T16:10:05.353298Z",
    "uid": "5387a4a4-02dd-4375-8db1-80bdd82ede96",
    "resource_version": "13"
  },
  "name": "demo-statement-1",
  "organization_id": "b0b21724-4586-4a07-b787-d0bb5aacbf87",
  "environment_id": "env-z3y2x1",
  "spec": {
    "statement": "select 1;",
    "properties": {},
    "compute_pool_id": "lfcp-8m03rm",
    "principal": "u-aq1dr2",
    "stopped": false,
    "execution_mode": "streaming"
  },
  "status": {
    "phase": "COMPLETED",
    "scaling_status": {
      "scaling_state": "OK",
      "last_updated": "2023-12-16T16:10:05Z"
    },
    "detail": "Statement completed successfully",
    "traits": {
      "sql_kind": "SELECT",
      "is_bounded": true,
      "is_append_only": true,
      "schema": {
        "columns": [
          {
            "name": "EXPR$0",
            "type": {
              "type": "INTEGER",
              "nullable": false
            }
          }
        ]
      }
    },
    "network_kind": "PUBLIC"
  }
}
```

</details>
<br/>

<a id="flink-rest-api-list-statements"></a>

### List statements

List the statements in an environment by sending a GET request to the
[Statements endpoint](/cloud/current/api.html#tag/Statements-(sqlv1)/operation/listSqlv1Statements).

Request query parameters
: - `spec.compute_pool_id` (optional): Fetch only the statements under this
    compute pool ID.
  - `page_token` (optional): Retrieve a page based on a previously received token
    (via the `metadata.next` field of `StatementList`).
  - `page_size` (optional): Maximum number of items to return in a page.
  - `label_selector` (optional): A comma-separated label selector to filter
    the statements by label. For more information, see
    [Filter statements by label](#flink-rest-api-filter-statements-by-label).

Listing all Flink SQL statements requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
```

The following command returns details for all non-deleted Flink SQL
statements under the scope of the environment (one or more compute pools)
where you have permission to do a GET request.

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to list the statements in an environment
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "StatementList",
  "metadata": {
    "self": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/statements",
    "next": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/statements?page_token=UvmDWOB1iwfAIBPj6EYb"
  },
  "data": [
    {
      "api_version": "sql/v1",
      "kind": "Statement",
      "metadata": {
        "self": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/statements/demo-statement-1",
        "created_at": "2023-12-16T16:08:36.650591Z",
        "updated_at": "2023-12-16T16:10:05.353298Z",
        "uid": "5387a4a4-02dd-4375-8db1-80bdd82ede96",
        "resource_version": "13"
      },
      "name": "demo-statement-1",
      "organization_id": "b0b21724-4586-4a07-b787-d0bb5aacbf87",
      "environment_id": "env-z3y2x1",
      "spec": {
        "statement": "select 1;",
        "properties": {},
        "compute_pool_id": "lfcp-8m03rm",
        "principal": "u-aq1dr2",
        "stopped": false,
        "execution_mode": "streaming"
      },
      "status": {
        "phase": "COMPLETED",
        "scaling_status": {
          "scaling_state": "OK",
          "last_updated": "2023-12-16T16:10:05Z"
        },
        "detail": "Statement completed successfully",
        "traits": {
          "sql_kind": "SELECT",
          "is_bounded": true,
          "is_append_only": true,
          "schema": {
            "columns": [
              {
                "name": "EXPR$0",
                "type": {
                  "type": "INTEGER",
                  "nullable": false
                }
              }
            ]
          }
        },
        "network_kind": "PUBLIC"
      }
    }
  ]
}
```

</details>
<br/>

<a id="flink-rest-api-filter-statements-by-label"></a>

#### Filter statements by label

Use the `label_selector` query parameter to filter the statements that a
list request returns. The parameter value is a comma-separated list of
`<label>=<value>` or `<label>!=<value>` expressions. When you provide more
than one expression, Confluent Cloud for Apache Flink combines them with AND logic, so only
statements that match every expression are returned. Each label in a
`label_selector` expression must use the `user.confluent.io/` prefix. For
more information about adding, updating, and removing labels on a statement,
see [Update labels](#flink-rest-api-update-labels).

Cloud Console and other Confluent client tooling label the
`INFORMATION_SCHEMA` statements that they submit internally, for example, to
populate the schema browser or autocomplete, with `user.confluent.io/hidden`
set to `true`. Confluent Cloud for Apache Flink doesn’t apply this label automatically based on a
statement’s SQL text. A statement that queries `INFORMATION_SCHEMA` isn’t
hidden unless the client that submitted it, or you, set the label explicitly.
To exclude statements that carry this label from the response, use the
following `label_selector` value.

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements?label_selector=user.confluent.io/hidden!=true" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

If you don’t specify a `label_selector`, the response includes all
statements, regardless of their labels.

<a id="flink-rest-api-update-statement"></a>

### Update metadata for a statement

Update the metadata for a statement by sending a PUT request to the
[Statements endpoint](/cloud/current/api.html#tag/Statements-(sqlv1)/operation/updateSqlv1Statement).

You can stop and resume a statement by setting `stopped` in the `spec` to
`true` to stop the statement and `false` to resume the statement.

- You can update the statement’s name, compute pool, and security principal.
  To update the compute pool or principal, you must stop the statement, send
  the update request, then restart the statement.
- The statement’s code is immutable.
- You must specify a resource version in the payload metadata.

Updating metadata for an existing Flink SQL statement requires the following
inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export STATEMENT_NAME="<statement-name>" # example: "user-filter"
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export COMPUTE_POOL_ID="<compute-pool-id>" # example: "lfcp-8m03rm"
export PRINCIPAL_ID="<principal-id>" # (optional) example: "sa-23kgz4" for a service account, or "u-aq1dr2" for a user account
export SQL_CODE="<sql-statement-text>" # example: "SELECT * FROM USERS;"
export RESOURCE_VERSION="<version>" # example: "a3e", must be fetched from the latest version of the statement
export JSON_DATA="<payload-string>"
```

The PRINCIPAL_ID parameter is optional. Confluent Cloud infers the principal from the
provided Flink API key.

The following JSON shows an example payload:

```json
{
  "name": "${STATEMENT_NAME}",
  "organization_id": "${ORG_ID}",
  "environment_id": "${ENV_ID}",
  "spec": {
    "statement": "${SQL_CODE}",
    "properties": {
      "key1": "value1",
      "key2": "value2"
    },
    "compute_pool_id": "${COMPUTE_POOL_ID}",
    "principal": "${PRINCIPAL_ID}",
    "stopped": false
  },
  "metadata": {
     "resource_version": "${RESOURCE_VERSION}"
  }
}
```

You must escape quotation mark characters in the JSON string, so the payload
string to send resembles the following:

```bash
export JSON_DATA="{
  \"name\": \"${STATEMENT_NAME}\",
  \"organization_id\": \"${ORG_ID}\",
  \"environment_id\": \"${ENV_ID}\",
  \"spec\": {
    \"statement\": \"${SQL_CODE}\",
    \"properties\": {
      \"key1\": \"value1\",
      \"key2\": \"value2\"
    },
    \"compute_pool_id\": \"${COMPUTE_POOL_ID}\",
    \"principal\": \"${PRINCIPAL_ID}\",
    \"stopped\": false
  },
  \"metadata\": {
    \"resource_version\": \"${RESOURCE_VERSION}\"
  }
}"
```

The following command sends a PUT request that updates metadata for an existing
Flink SQL statement.

```bash
curl --request PUT \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

The PUT request requires a resource version, which changes every time the
system or the user updates the statement. You can’t calculate the resource
version ahead of time, so if the statement has changed since you fetched
it, you must submit a GET request, reapply the modifications, and try the
update again.

This means you must loop and retry on 409 errors. The following pseudocode
shows the loop.

```text
while true:
  statement = getStatement()
  # make modifications to the current statement
  statement.spec.stopped = True
  # send the update
  response = updateStatement(statement)
  # if a conflict, retry
  if response.code == 409:
    continue
  elif response.code == 200:
    return "success"
  else:
    return response.error()
```

<a id="flink-rest-api-move-statement-compute-pool"></a>

#### Move a statement to a different compute pool

To move a running statement to a different compute pool, stop the
statement, then send another update request with the new compute pool ID.
For more information, see
[Move a Statement to a Different Compute Pool](../how-to-guides/move-statement-compute-pool.md#flink-sql-move-statement-compute-pool).

1. Send a GET request to the
   [Statements endpoint](#flink-rest-api-get-statement) to fetch the
   statement’s current `resource_version`.
2. Send a PUT request with `stopped` set to `true` to stop the
   statement.
   ```bash
   export JSON_DATA="{
     \"name\": \"${STATEMENT_NAME}\",
     \"organization_id\": \"${ORG_ID}\",
     \"environment_id\": \"${ENV_ID}\",
     \"spec\": {
       \"statement\": \"${SQL_CODE}\",
       \"compute_pool_id\": \"${COMPUTE_POOL_ID}\",
       \"stopped\": true
     },
     \"metadata\": {
       \"resource_version\": \"${RESOURCE_VERSION}\"
     }
   }"

   curl --request PUT \
     --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}" \
     --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
     --header 'content-type: application/json' \
     --data "${JSON_DATA}"
   ```
3. Poll the
   [Statements endpoint](#flink-rest-api-get-statement) until
   `status.phase` is `STOPPED`, and note the new
   `resource_version` from the response. The resource version changes
   every time the statement’s status changes, so you must fetch it again
   after the statement stops.
4. Send another PUT request with `stopped` set to `false`,
   `compute_pool_id` set to the ID of the new compute pool, and the
   `resource_version` you fetched in the previous step.
   ```bash
   export NEW_COMPUTE_POOL_ID="<new-compute-pool-id>" # example: "lfcp-8m09g0"

   export JSON_DATA="{
     \"name\": \"${STATEMENT_NAME}\",
     \"organization_id\": \"${ORG_ID}\",
     \"environment_id\": \"${ENV_ID}\",
     \"spec\": {
       \"statement\": \"${SQL_CODE}\",
       \"compute_pool_id\": \"${NEW_COMPUTE_POOL_ID}\",
       \"stopped\": false
     },
     \"metadata\": {
       \"resource_version\": \"${RESOURCE_VERSION}\"
     }
   }"

   curl --request PUT \
     --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}" \
     --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
     --header 'content-type: application/json' \
     --data "${JSON_DATA}"
   ```

   Confluent Cloud resumes the statement under the new compute pool, continuing
   from the savepoint that was taken when the statement stopped.

<a id="flink-rest-api-patch-statement"></a>

### Patch a statement

Partially update a Flink SQL statement by sending a PATCH request to the
[Statements endpoint](/cloud/current/api.html#tag/Statements-(sqlv1)/operation/patchSqlv1Statement).

The PATCH endpoint enables you to make partial updates to a statement
without requiring the full statement object or resource version. This
avoids the 409 conflict errors that can occur with PUT requests when the
system updates the statement frequently.

The PATCH request uses JSON Patch format as specified in
[RFC 6902](https://tools.ietf.org/html/rfc6902), which enables you to
specify operations to add, replace, or remove specific fields.

Patching a statement requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export STATEMENT_NAME="<statement-name>" # example: "user-filter"
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
```

#### Stop or resume a statement

The following command stops a statement by setting `spec.stopped` to `true`.

```bash
curl --request PATCH \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header "Content-Type: application/json" \
  --data '[{"path":"/spec/stopped","op":"replace","value":true}]'
```

To resume a stopped statement, set `spec.stopped` to `false`:

```bash
curl --request PATCH \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header "Content-Type: application/json" \
  --data '[{"path":"/spec/stopped","op":"replace","value":false}]'
```

<a id="flink-rest-api-update-labels"></a>

#### Update labels

You can add, replace, or remove user labels (labels with the
`user.confluent.io/` prefix) using PATCH operations. The `/metadata/labels`
path replaces the entire label map in a single operation. Include every
label that you want the statement to keep, not only the one that you’re
adding or changing.

The following command adds a label to a statement that has no existing
labels:

```bash
curl --request PATCH \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header "Content-Type: application/json" \
  --data '[{"path":"/metadata/labels","op":"replace","value":{"user.confluent.io/hidden":"true"}}]'
```

If the statement already has other labels that you want to keep, include them
in the `value` map along with the label that you’re adding or changing.

The following command removes all labels from a statement:

```bash
curl --request PATCH \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header "Content-Type: application/json" \
  --data '[{"path":"/metadata/labels","op":"remove"}]'
```

To remove a single label while keeping others, use the `replace` operation
with a `value` map that omits the label that you want to remove.

After you add a label to a statement, you can use it to filter list requests.
For more information, see [Filter statements by label](#flink-rest-api-filter-statements-by-label).

<a id="flink-rest-api-delete-statement"></a>

### Delete a statement

Delete a statement from the compute pool by sending a DELETE request to the
[Statements endpoint](/cloud/current/api.html#tag/Statements-(sqlv1)/operation/deleteSqlv1Statement).

- After Confluent Cloud deletes a statement, you can’t undo the deletion.
- Confluent Cloud cleans up the state.
- After deletion completes, the statement is no longer accessible.

Deleting a statement requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export STATEMENT_NAME="<statement-name>" # example: "user-filter"
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
```

The following command deletes a statement in the specified organization and
environment.

```bash
curl --request DELETE \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/statements/${STATEMENT_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

<a id="flink-rest-api-manage-materialized-tables"></a>

## Manage materialized tables

Using requests to the Flink REST API, you can perform these actions:

- [Create a materialized table](#flink-rest-api-create-materialized-table)
- [Read a materialized table](#flink-rest-api-read-materialized-table)
- [List materialized tables](#flink-rest-api-list-materialized-tables)
- [Update a materialized table](#flink-rest-api-update-materialized-table)
- [Delete a materialized table](#flink-rest-api-delete-materialized-table)
- [List materialized table versions](#flink-rest-api-list-materialized-table-versions)
- [Read a materialized table version](#flink-rest-api-read-materialized-table-version)

For more information about materialized tables, see
[Materialized Tables](../concepts/materialized-tables.md#flink-sql-materialized-tables).

### Materialized table schema

A materialized table has the following schema:

```yaml
api_version: "sql/v1"
kind: "MaterializedTable"
name: "" # Name of the materialized table, unique within the database, max 100 chars
organization_id: "" # Identifier of your Confluent Cloud organization
environment_id: "" # Identifier of your Confluent Cloud environment
metadata:
  created_at: "" # Creation timestamp
  updated_at: "" # Last updated timestamp
  resource_version: "" # Opaque version string for optimistic locking (required for PUT)
  self: "" # An absolute URL to this resource
  uid: "" # Globally unique identifier
spec:
  query: "" # The SELECT query that defines the continuous pipeline, max 131072 chars
  compute_pool_id: "" # Optional. The ID of the compute pool
  principal: "" # Optional. User or service account ID that runs the query
  stopped: false # When true, the continuous query is paused
  table_options: {} # Optional. Table-level WITH clause properties
  session_options: {} # Optional. Session-level configuration (SET statement equivalent)
  columns: [] # Column definitions (inferred from query if not specified)
  watermark: {} # Watermark specification
  constraints: [] # Table constraints (for example, PRIMARY_KEY)
  distribution: {} # DISTRIBUTED BY specification
  kafka_cluster_id: "" # Read-only. Set from path parameter
status:
  phase: PENDING | RUNNING | CREATING | ALTERING | DEGRADED | STOPPING | STOPPED | FAILING | FAILED | COMPLETING | DELETING
  detail: "" # Human-readable description of the phase
  warnings: [] # Non-fatal warnings
  creation_statement: "" # The SQL statement that corresponds to this table
  scaling_status: {} # Scaling state information
  version: 1 # Current version number
  latest_version: 1 # Latest submitted version number
```

The table name has a maximum length of 100 characters and must satisfy
the following regular expression:

```none
[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*
```

<a id="flink-rest-api-create-materialized-table"></a>

### Create a materialized table

Create a materialized table in your environment by sending a POST request to the
[Materialized Tables endpoint](/cloud/current/api.html#tag/Materialized-Tables-(sqlv1)/operation/createSqlv1MaterializedTable).

Creating a materialized table requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export KAFKA_CLUSTER_ID="<kafka-cluster-id>" # example: "lkc-abc123"
export TABLE_NAME="<table-name>" # example: "high_value_orders"
export COMPUTE_POOL_ID="<compute-pool-id>" # (optional) example: "lfcp-8m03rm"
export PRINCIPAL_ID="<principal-id>" # (optional) example: "sa-23kgz4" for a service account
export JSON_DATA="<payload-string>"
```

The COMPUTE_POOL_ID and PRINCIPAL_ID parameters are optional. If you don’t
specify COMPUTE_POOL_ID, the table uses the default compute pool. If you don’t
specify PRINCIPAL_ID, Confluent Cloud infers the principal from the Flink API key you
provide.

The following JSON shows an example payload:

```json
{
  "name": "${TABLE_NAME}",
  "spec": {
    "query": "SELECT order_id, customer_id, price FROM examples.marketplace.orders WHERE price > 50.00",
    "compute_pool_id": "${COMPUTE_POOL_ID}",
    "principal": "${PRINCIPAL_ID}",
    "stopped": false
  }
}
```

The `kafka_cluster_id` is derived from the URL path and should not be included
in the request body. The `table_options`, `session_options`, `columns`,
`watermark`, `constraints`, and `distribution` fields are optional.

Escape the quotation mark characters in the JSON string. The payload
string to send resembles the following:

```bash
export JSON_DATA="{
  \"name\": \"${TABLE_NAME}\",
  \"spec\": {
    \"query\": \"SELECT order_id, customer_id, price FROM examples.marketplace.orders WHERE price > 50.00\",
    \"compute_pool_id\": \"${COMPUTE_POOL_ID}\",
    \"principal\": \"${PRINCIPAL_ID}\",
    \"stopped\": false
  }
}"
```

The following command sends a POST request that creates a materialized table.

```bash
curl --request POST \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/databases/${KAFKA_CLUSTER_ID}/materialized-tables" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to create a materialized table
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "MaterializedTable",
  "name": "high_value_orders",
  "organization_id": "b0b21724-4586-4a07-b787-d0bb5aacbf87",
  "environment_id": "env-z3y2x1",
  "metadata": {
    "created_at": "2026-04-14T10:00:00.000000Z",
    "updated_at": "2026-04-14T10:00:00.000000Z",
    "resource_version": "1",
    "self": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/databases/lkc-abc123/materialized-tables/high_value_orders",
    "uid": "12345678-1234-1234-1234-123456789012"
  },
  "spec": {
    "query": "SELECT order_id, customer_id, price FROM examples.marketplace.orders WHERE price > 50.00",
    "compute_pool_id": "lfcp-8m03rm",
    "principal": "sa-23kgz4",
    "stopped": false,
    "kafka_cluster_id": "lkc-abc123"
  },
  "status": {
    "phase": "PENDING",
    "detail": "Materialized table is being created",
    "version": 1,
    "latest_version": 1
  }
}
```

</details>
<br/>

<a id="flink-rest-api-read-materialized-table"></a>

### Read a materialized table

Get the details about a materialized table by sending a GET request to the
[Materialized Tables endpoint](/cloud/current/api.html#tag/Materialized-Tables-(sqlv1)/operation/getSqlv1MaterializedTable).

Getting details about a materialized table requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export KAFKA_CLUSTER_ID="<kafka-cluster-id>" # example: "lkc-abc123"
export TABLE_NAME="<table-name>" # example: "high_value_orders"
```

Run the following command to get details about the materialized
table specified by the TABLE_NAME environment variable.

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/databases/${KAFKA_CLUSTER_ID}/materialized-tables/${TABLE_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to get details about a materialized table
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "MaterializedTable",
  "name": "high_value_orders",
  "organization_id": "b0b21724-4586-4a07-b787-d0bb5aacbf87",
  "environment_id": "env-z3y2x1",
  "metadata": {
    "created_at": "2026-04-14T10:00:00.000000Z",
    "updated_at": "2026-04-14T10:05:00.000000Z",
    "resource_version": "2",
    "self": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/databases/lkc-abc123/materialized-tables/high_value_orders",
    "uid": "12345678-1234-1234-1234-123456789012"
  },
  "spec": {
    "query": "SELECT order_id, customer_id, price FROM examples.marketplace.orders WHERE price > 50.00",
    "compute_pool_id": "lfcp-8m03rm",
    "principal": "sa-23kgz4",
    "stopped": false,
    "kafka_cluster_id": "lkc-abc123",
    "columns": [
      {"name": "order_id", "type": "INT"},
      {"name": "customer_id", "type": "INT"},
      {"name": "price", "type": "DOUBLE"}
    ],
    "watermark": {
      "column": "$rowtime",
      "strategy": "Default"
    }
  },
  "status": {
    "phase": "RUNNING",
    "detail": "Materialized table is running",
    "creation_statement": "CREATE MATERIALIZED TABLE high_value_orders AS SELECT order_id, customer_id, price FROM examples.marketplace.orders WHERE price > 50.00",
    "scaling_status": {
      "scaling_state": "OK"
    },
    "version": 1,
    "latest_version": 1
  }
}
```

</details>
<br/>

<a id="flink-rest-api-list-materialized-tables"></a>

### List materialized tables

List all materialized tables across all databases in your environment by sending
a GET request to the
[Materialized Tables endpoint](/cloud/current/api.html#tag/Materialized-Tables-(sqlv1)/operation/listSqlv1MaterializedTables).

This endpoint returns materialized tables across all databases (Kafka clusters)
in the environment. The URL path does not include a database identifier.

Listing materialized tables requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
```

You can use the `page_size` query parameter to control the number of results
per page (default 10, maximum 100) and `page_token` for pagination.

Run the following command to list the materialized tables in your environment.

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/materialized-tables" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to list materialized tables
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "MaterializedTableList",
  "metadata": {
    "next": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/materialized-tables?page_token=eyJ..."
  },
  "data": [
    {
      "name": "high_value_orders",
      "organization_id": "b0b21724-4586-4a07-b787-d0bb5aacbf87",
      "environment_id": "env-z3y2x1",
      "spec": {
        "query": "SELECT order_id, customer_id, price FROM examples.marketplace.orders WHERE price > 50.00",
        "stopped": false,
        "kafka_cluster_id": "lkc-abc123"
      },
      "status": {
        "phase": "RUNNING",
        "version": 1,
        "latest_version": 1
      }
    },
    {
      "name": "customer_activity",
      "organization_id": "b0b21724-4586-4a07-b787-d0bb5aacbf87",
      "environment_id": "env-z3y2x1",
      "spec": {
        "query": "SELECT customer_id, COUNT(*) AS order_count FROM examples.marketplace.orders GROUP BY customer_id",
        "stopped": false,
        "kafka_cluster_id": "lkc-xyz789"
      },
      "status": {
        "phase": "RUNNING",
        "version": 2,
        "latest_version": 2
      }
    }
  ]
}
```

</details>
<br/>

<a id="flink-rest-api-update-materialized-table"></a>

### Update a materialized table

Update a materialized table by sending a PUT request to the
[Materialized Tables endpoint](/cloud/current/api.html#tag/Materialized-Tables-(sqlv1)/operation/updateSqlv1MaterializedTable).

You can update the following fields: `query` (triggers an evolution),
`stopped` (to stop or resume the continuous query), `compute_pool_id`, and
`principal`.

You must include `metadata.resource_version` in the PUT request. The resource
version changes every time the materialized table is updated. If the table has
been modified since you last fetched it, the request returns `409 Conflict`.
Fetch the latest version and retry.

Updating a materialized table requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export KAFKA_CLUSTER_ID="<kafka-cluster-id>" # example: "lkc-abc123"
export TABLE_NAME="<table-name>" # example: "high_value_orders"
export COMPUTE_POOL_ID="<compute-pool-id>" # (optional) example: "lfcp-8m03rm"
export PRINCIPAL_ID="<principal-id>" # (optional) example: "sa-23kgz4"
export RESOURCE_VERSION="<version>" # example: "2", must be fetched from the latest version of the table
export JSON_DATA="<payload-string>"
```

**Evolve a materialized table**

To evolve a materialized table, update the `query` field. The
following example raises the price threshold and adds a
`product_id` column:

```json
{
  "name": "${TABLE_NAME}",
  "spec": {
    "query": "SELECT order_id, customer_id, price, product_id FROM examples.marketplace.orders WHERE price > 100.00",
    "compute_pool_id": "${COMPUTE_POOL_ID}",
    "principal": "${PRINCIPAL_ID}",
    "stopped": false
  },
  "metadata": {
    "resource_version": "${RESOURCE_VERSION}"
  }
}
```

**Stop and resume a materialized table**

To stop a materialized table without dropping it, set `stopped` to `true`.
The backing Kafka topic and its data are preserved.

```json
{
  "name": "${TABLE_NAME}",
  "spec": {
    "stopped": true
  },
  "metadata": {
    "resource_version": "${RESOURCE_VERSION}"
  }
}
```

To resume, set `stopped` to `false`. The query restarts and continues
reading from the last committed offset.

Escape the quotation mark characters in the JSON string. The payload
string to send resembles the following:

```bash
export JSON_DATA="{
  \"name\": \"${TABLE_NAME}\",
  \"spec\": {
    \"query\": \"SELECT order_id, customer_id, price, product_id FROM examples.marketplace.orders WHERE price > 100.00\",
    \"compute_pool_id\": \"${COMPUTE_POOL_ID}\",
    \"principal\": \"${PRINCIPAL_ID}\",
    \"stopped\": false
  },
  \"metadata\": {
    \"resource_version\": \"${RESOURCE_VERSION}\"
  }
}"
```

The following command sends a PUT request that updates a materialized table.

```bash
curl --request PUT \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/databases/${KAFKA_CLUSTER_ID}/materialized-tables/${TABLE_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

The PUT request requires the resource version, which changes every time the
system or a user updates the materialized table. If the table changed since
you fetched it, submit a GET request, reapply your modifications, and retry
the update.

This means you must loop and retry on 409 errors. The following pseudocode
shows the loop.

```text
while true:
  table = getMaterializedTable()
  # make modifications to the current table
  table.spec.query = "SELECT ..."
  # send the update
  response = updateMaterializedTable(table)
  # if a conflict, retry
  if response.code == 409:
    continue
  elif response.code == 200:
    return "success"
  else:
    return response.error()
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to update a materialized table
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "MaterializedTable",
  "name": "high_value_orders",
  "organization_id": "b0b21724-4586-4a07-b787-d0bb5aacbf87",
  "environment_id": "env-z3y2x1",
  "metadata": {
    "created_at": "2026-04-14T10:00:00.000000Z",
    "updated_at": "2026-04-14T12:30:00.000000Z",
    "resource_version": "3",
    "self": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/databases/lkc-abc123/materialized-tables/high_value_orders",
    "uid": "12345678-1234-1234-1234-123456789012"
  },
  "spec": {
    "query": "SELECT order_id, customer_id, price, product_id FROM examples.marketplace.orders WHERE price > 100.00",
    "compute_pool_id": "lfcp-8m03rm",
    "principal": "sa-23kgz4",
    "stopped": false,
    "kafka_cluster_id": "lkc-abc123"
  },
  "status": {
    "phase": "RUNNING",
    "detail": "Materialized table is running",
    "version": 2,
    "latest_version": 2
  }
}
```

</details>
<br/>

<a id="flink-rest-api-delete-materialized-table"></a>

### Delete a materialized table

Delete a materialized table by sending a DELETE request to the
[Materialized Tables endpoint](/cloud/current/api.html#tag/Materialized-Tables-(sqlv1)/operation/deleteSqlv1MaterializedTable).

The DELETE request returns HTTP status `202 Accepted`. Deletion is
asynchronous: the materialized table transitions through the `DELETING` phase
before being fully removed. The backing Kafka topic is deleted along with the
table.

Deleting a materialized table requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export KAFKA_CLUSTER_ID="<kafka-cluster-id>" # example: "lkc-abc123"
export TABLE_NAME="<table-name>" # example: "high_value_orders"
```

The following command sends a DELETE request that deletes a materialized table.

```bash
curl --request DELETE \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/databases/${KAFKA_CLUSTER_ID}/materialized-tables/${TABLE_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

<a id="flink-rest-api-list-materialized-table-versions"></a>

### List materialized table versions

List the versions of a materialized table by sending a GET request to the
[Materialized Tables endpoint](/cloud/current/api.html#tag/Materialized-Tables-(sqlv1)/operation/listSqlv1MaterializedTableVersions).

Each time you evolve a materialized table by changing its `query` attribute,
Flink creates a new version. Use this endpoint to view the history of evolutions.

Listing materialized table versions requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export KAFKA_CLUSTER_ID="<kafka-cluster-id>" # example: "lkc-abc123"
export TABLE_NAME="<table-name>" # example: "high_value_orders"
```

Run the following command to list the versions of a materialized table.

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/databases/${KAFKA_CLUSTER_ID}/materialized-tables/${TABLE_NAME}/versions" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to list materialized table versions
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "MaterializedTableVersionList",
  "data": [
    {
      "spec": {
        "version": 1,
        "statement": "CREATE MATERIALIZED TABLE high_value_orders AS SELECT order_id, customer_id, price FROM examples.marketplace.orders WHERE price > 50.00",
        "changes": []
      },
      "metadata": {
        "created_at": "2026-04-14T10:00:00.000000Z"
      }
    },
    {
      "spec": {
        "version": 2,
        "statement": "CREATE OR ALTER MATERIALIZED TABLE high_value_orders AS SELECT order_id, customer_id, price, product_id FROM examples.marketplace.orders WHERE price > 100.00",
        "changes": [
          {
            "description": "Query updated"
          }
        ]
      },
      "metadata": {
        "created_at": "2026-04-14T12:30:00.000000Z"
      }
    }
  ]
}
```

</details>
<br/>

<a id="flink-rest-api-read-materialized-table-version"></a>

### Read a materialized table version

Get details about a specific version of a materialized table by sending a GET
request to the
[Materialized Tables endpoint](/cloud/current/api.html#tag/Materialized-Tables-(sqlv1)/operation/getSqlv1MaterializedTableVersion).

Getting details about a specific version requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export KAFKA_CLUSTER_ID="<kafka-cluster-id>" # example: "lkc-abc123"
export TABLE_NAME="<table-name>" # example: "high_value_orders"
export VERSION="<version-number>" # example: "1"
```

Run the following command to get details about a specific version.

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/databases/${KAFKA_CLUSTER_ID}/materialized-tables/${TABLE_NAME}/versions/${VERSION}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to get a materialized table version
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "MaterializedTableVersion",
  "spec": {
    "version": 1,
    "statement": "CREATE MATERIALIZED TABLE high_value_orders AS SELECT order_id, customer_id, price FROM examples.marketplace.orders WHERE price > 50.00",
    "changes": []
  },
  "metadata": {
    "created_at": "2026-04-14T10:00:00.000000Z"
  }
}
```

</details>
<br/>

<a id="flink-rest-api-manage-compute-pools"></a>

## Manage compute pools

Using requests to the Flink REST API, you can perform these actions:

- [List Flink compute pools](#flink-rest-api-list-compute-pools)
- [Create a Flink compute pool](#flink-rest-api-create-pool)
- [Read a Flink compute pool](#flink-rest-api-read-compute-pool)
- [Update a Flink compute pool](#flink-rest-api-update-compute-pool)
- [Delete a Flink compute pool](#flink-rest-api-delete-compute-pool)

You need authorization to create, update, delete (`FlinkAdmin`) or use
(`FlinkDeveloper`) a compute pool. For more information, see [Grant Role-Based Access in Confluent Cloud for Apache Flink](flink-rbac.md#flink-rbac).

<a id="flink-rest-api-list-compute-pools"></a>

### List Flink compute pools

List the compute pools in your environment by sending a GET request to the
[Compute Pools endpoint](/cloud/current/api.html#tag/Compute-Pools-(fcpmv2)/operation/listFcpmV2ComputePools).

- This request uses your Cloud API key instead of the Flink API key.

Listing the compute pools in your environment requires the following inputs:

```bash
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

Run the following command to list the compute pools in your environment.

```bash
curl --request GET \
     --url "https://confluent.cloud/api/fcpm/v2/compute-pools?environment=${ENV_ID}&page_size=100" \
     --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
     | jq -r '.data[] | .spec.display_name, {id}'
```

Your output should resemble:

```none
compute_pool_0
{
  "id": "lfcp-j123kl"
}
compute_pool_2
{
  "id": "lfcp-abc1de"
}
my-lfcp-01
{
  "id": "lfcp-l2mn3o"
}
...
```

Find your compute pool in the list and save its ID in an environment variable.

```bash
export COMPUTE_POOL_ID="<your-compute-pool-id>"
```

<a id="flink-rest-api-create-pool"></a>

### Create a Flink compute pool

Create a compute pool in your environment by sending a POST request to the
[Compute Pools endpoint](/cloud/current/api.html#tag/Compute-Pools-(fcpmv2)/operation/createFcpmV2ComputePool).

- This request uses your Cloud API key instead of the Flink API key.

Creating a compute pool requires the following inputs:

```bash
export COMPUTE_POOL_NAME="<compute-pool-name>" # human readable name, for example: "my-compute-pool"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export MAX_CFU="<max-cfu>" # example: 5
export JSON_DATA="<payload-string>"
```

The following JSON shows an example payload. The `network` and `default_pool` keys are optional.

```json
{
  "spec": {
    "display_name": "${COMPUTE_POOL_NAME}",
    "cloud": "${CLOUD_PROVIDER}",
    "region": "${CLOUD_REGION}",
    "max_cfu": ${MAX_CFU},
    "environment": {
      "id": "${ENV_ID}"
    },
    "network": {
      "id": "n-00000",
      "environment": "string"
    },
    "default_pool": false
  }
}
```

#### NOTE
Set `default_pool` to `true` to create a default compute pool manually.
This pre-provisions a default pool with your specified configuration before
users start running statements. Creating default pools requires the
`FlinkAdmin` role or higher.

Escape the quotation mark characters in the JSON string. The payload
string to send resembles the following:

```bash
export JSON_DATA="{
  \"spec\": {
    \"display_name\": \"${COMPUTE_POOL_NAME}\",
    \"cloud\": \"${CLOUD_PROVIDER}\",
    \"region\": \"${CLOUD_REGION}\",
    \"max_cfu\": ${MAX_CFU},
    \"environment\": {
      \"id\": \"${ENV_ID}\"
    }
  }
}"
```

The following command sends a POST request to create a compute pool.

```bash
curl --request POST \
  --url https://api.confluent.cloud/fcpm/v2/compute-pools \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to create a compute pool
</summary>
```json
{
    "api_version": "fcpm/v2",
    "id": "lfcp-6g7h8i",
    "kind": "ComputePool",
    "metadata": {
        "created_at": "2024-02-27T22:44:27.18964Z",
        "resource_name": "crn://confluent.cloud/organization=b0b21724-4586-4a07-b787-d0bb5aacbf87/environment=env-z3y2x1/flink-region=aws.us-east-1/compute-pool=lfcp-6g7h8i",
        "self": "https://api.confluent.cloud/fcpm/v2/compute-pools/lfcp-6g7h8i",
        "updated_at": "2024-02-27T22:44:27.18964Z"
    },
    "spec": {
        "cloud": "AWS",
        "display_name": "my-compute-pool",
        "environment": {
            "id": "env-z3y2x1",
            "related": "https://api.confluent.cloud/fcpm/v2/compute-pools/lfcp-6g7h8i",
            "resource_name": "crn://confluent.cloud/organization=b0b21724-4586-4a07-b787-d0bb5aacbf87/environment=env-z3y2x1"
        },
        "http_endpoint": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1",
        "max_cfu": 5,
        "region": "us-east-1"
    },
    "status": {
        "current_cfu": 0,
        "phase": "PROVISIONING"
    }
}
```

</details>
<br/>

<a id="flink-rest-api-read-compute-pool"></a>

### Read a Flink compute pool

Get the details about a compute pool in your environment by sending a GET request to the
[Compute Pools endpoint](/cloud/current/api.html#tag/Compute-Pools-(fcpmv2)/operation/getFcpmV2ComputePool).

- This request uses your Cloud API key instead of the Flink API key.

Getting details about a compute pool requires the following inputs:

```bash
export COMPUTE_POOL_ID="<compute-pool-id>" # example: "lfcp-8m03rm"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

Run the following command to get details about the compute pool specified in
the COMPUTE_POOL_ID environment variable.

```bash
curl --request GET \
  --url "https://api.confluent.cloud/fcpm/v2/compute-pools/${COMPUTE_POOL_ID}?environment=${ENV_ID}" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to read a compute pool
</summary>
```json
{
    "api_version": "fcpm/v2",
    "id": "lfcp-6g7h8i",
    "kind": "ComputePool",
    "metadata": {
        "created_at": "2024-02-27T22:44:27.18964Z",
        "resource_name": "crn://confluent.cloud/organization=b0b21724-4586-4a07-b787-d0bb5aacbf87/environment=env-z3y2x1/flink-region=aws.us-east-1/compute-pool=lfcp-6g7h8i",
        "self": "https://api.confluent.cloud/fcpm/v2/compute-pools/lfcp-6g7h8i",
        "updated_at": "2024-02-27T22:44:27.18964Z"
    },
    "spec": {
        "cloud": "AWS",
        "display_name": "my-compute-pool",
        "environment": {
            "id": "env-z3y2x1",
            "related": "https://api.confluent.cloud/fcpm/v2/compute-pools/lfcp-6g7h8i",
            "resource_name": "crn://confluent.cloud/organization=b0b21724-4586-4a07-b787-d0bb5aacbf87/environment=env-z3y2x1"
        },
        "http_endpoint": "https://flink.us-east-1.aws.confluent.cloud/sql/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1",
        "max_cfu": 5,
        "region": "us-east-1"
    },
    "status": {
        "current_cfu": 0,
        "phase": "PROVISIONED"
    }
}
```

</details>
<br/>

<a id="flink-rest-api-update-compute-pool"></a>

### Update a Flink compute pool

Update a compute pool in your environment by sending a PATCH request to the
[Compute Pools endpoint](/cloud/current/api.html#tag/Compute-Pools-(fcpmv2)/operation/updateFcpmV2ComputePool).

- This request uses your Cloud API key instead of the Flink API key.

Updating a compute pool requires the following inputs:

```bash
export COMPUTE_POOL_ID="<compute-pool-id>" # example: "lfcp-8m03rm"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export MAX_CFU="<max-cfu>" # example: 5
export JSON_DATA="<payload-string>"
```

The following JSON shows an example payload. The `network` key is optional.

```json
{
  "spec": {
    "display_name": "${COMPUTE_POOL_NAME}",
    "max_cfu": ${MAX_CFU},
    "environment": {
      "id": "${ENV_ID}"
    }
  }
}
```

Escape the quotation mark characters in the JSON string. The payload
string to send resembles the following:

```bash
export JSON_DATA="{
  \"spec\": {
    \"display_name\": \"${COMPUTE_POOL_NAME}\",
    \"max_cfu\": ${MAX_CFU},
    \"environment\": {
      \"id\": \"${ENV_ID}\"
    }
  }
}"
```

Run the following command to update the compute pool specified in the
COMPUTE_POOL_ID environment variable.

```bash
curl --request PATCH \
  --url "https://api.confluent.cloud/fcpm/v2/compute-pools/${COMPUTE_POOL_ID}" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

<a id="flink-rest-api-delete-compute-pool"></a>

### Delete a Flink compute pool

Delete a compute pool in your environment by sending a DELETE request to the
[Compute Pools endpoint](/cloud/current/api.html#tag/Compute-Pools-(fcpmv2)/operation/deleteFcpmV2ComputePool).

- This request uses your Cloud API key instead of the Flink API key.

Deleting a compute pool requires the following inputs:

```bash
export COMPUTE_POOL_ID="<compute-pool-id>" # example: "lfcp-8m03rm"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

Run the following command to delete the compute pool specified in the
COMPUTE_POOL_ID environment variable.

```bash
curl --request DELETE \
  --url "https://api.confluent.cloud/fcpm/v2/compute-pools/${COMPUTE_POOL_ID}?environment=${ENV_ID}" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}"
```

<a id="flink-rest-api-list-regions"></a>

## List Flink regions

List the regions where Flink is available by sending a GET request to the
[Regions endpoint](/cloud/current/api.html#tag/Regions-(fcpmv2)/operation/listFcpmV2Regions).

- This request uses your Cloud API key instead of the Flink API key.

Getting details about a compute pool requires the following inputs:

```bash
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
```

Run the following command to list the available Flink regions.

```bash
curl --request GET \
  --url "https://api.confluent.cloud/fcpm/v2/regions" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
  | jq -r '.data[].id'
```

Your output should resemble:

```none
aws.eu-central-1
aws.us-east-1
aws.eu-west-1
aws.us-east-2
...
```

<a id="flink-rest-api-manage-artifacts"></a>

## Manage Flink artifacts

Using requests to the Flink REST API, you can perform these actions:

- [List Flink artifacts](#flink-rest-api-list-artifacts)
- [Create a Flink artifact](#flink-rest-api-create-artifact)
- [Read an artifact](#flink-rest-api-read-artifact)
- [Update an artifact](#flink-rest-api-update-artifact)
- [Delete an artifact](#flink-rest-api-delete-artifact)

An artifact has the following schema:

```yaml
api_version: artifact/v1
kind: FlinkArtifact
id: dlz-f3a90de
metadata:
  self: 'https://api.confluent.cloud/artifact/v1/flink-artifacts/fa-12345'
  resource_name: crn://confluent.cloud/organization=<org-id>/flink-artifact=fa-12345
  created_at: '2006-01-02T15:04:05-07:00'
  updated_at: '2006-01-02T15:04:05-07:00'
  deleted_at: '2006-01-02T15:04:05-07:00'
cloud: AWS
region: us-east-1
environment: env-00000
display_name: string
class: io.confluent.example.SumScalarFunction
content_format: JAR
description: string
documentation_link: '^$|^(http://|https://).'
runtime_language: JAVA
versions:
  - version: cfa-ver-001
    release_notes: string
    is_beta: true
    artifact_id: {}
    upload_source:
      api_version: artifact.v1/UploadSource
      kind: PresignedUrl
      id: dlz-f3a90de
      metadata:
        self: https://api.confluent.cloud/artifact.v1/UploadSource/presigned-urls/pu-12345
        resource_name: crn://confluent.cloud/organization=<org-id>/presigned-url=pu-12345
        created_at: '2006-01-02T15:04:05-07:00'
        updated_at: '2006-01-02T15:04:05-07:00'
        deleted_at: '2006-01-02T15:04:05-07:00'
      location: PRESIGNED_URL_LOCATION
      upload_id: <guid>
```

<a id="flink-rest-api-list-artifacts"></a>

### List Flink artifacts

List the artifacts, like user-defined functions (UDFs), in your environment by
sending a GET request to the
[List Artifacts endpoint](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1)/operation/listArtifactV1FlinkArtifacts).

- This request uses your Cloud API key instead of the Flink API key.

Listing the artifacts in your environment requires the following inputs:

```bash
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

Run the following command to list the artifacts in your environment.

```bash
curl --request GET \
     --url "https://api.confluent.cloud/artifact/v1/flink-artifacts?cloud=${CLOUD_PROVIDER}&region=${CLOUD_REGION}&environment=${ENV_ID}" \
     --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
     | jq -r '.data[] | .spec.display_name, {id}'
```

Your output should resemble:

```json
{
  "id": "cfa-e8rzq7"
}
```

<a id="flink-rest-api-create-artifact"></a>

### Create a Flink artifact

Creating an artifact, like a user-defined function (UDF), requires these steps:

1. Request a presigned upload URL for a new Flink Artifact by sending a POST
   request to the
   [Presigned URLs endpoint](/cloud/current/api.html#tag/Presigned-Urls-(artifactv1)/The-Presigned-Urls-Model).
2. Upload your JAR file to the object storage provider by using the results from
   the presigned URL request.
3. Create the artifact in your environment by sending a POST request to the
   [Create Artifact endpoint](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1)/operation/createArtifactV1FlinkArtifact).

- These requests use your Cloud API key instead of the Flink API key.

Creating an artifact in your environment requires the following inputs:

```bash
export ARTIFACT_DISPLAY_NAME="<human-readable-name>" # example: "my-udf"
export ARTIFACT_DESCRIPTION="<description>" # example: "This is a demo UDF."
export ARTIFACT_DOC_LINK="<url-to-documentation>" # example: "https://docs.example.com/my-udf"
export CLASS_NAME="<java-class-name>" # example: "io.confluent.example.SumScalarFunction"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

The following JSON shows an example payload.

```json
{
  "content_format": "JAR",
  "cloud": "${CLOUD_PROVIDER}",
  "environment": "${ENV_ID}",
  "region": "${CLOUD_REGION}"
}
```

You must escape quotation mark characters in the JSON string, so the payload
string to send resembles the following:

```bash
export JSON_DATA="{
  \"content_format\": \"JAR\",
  \"cloud\": \"${CLOUD_PROVIDER}\",
  \"environment\": \"${ENV_ID}\",
  \"region\": \"${CLOUD_REGION}\"
}"
```

Run the following command to request the upload identifier and the presigned
upload URL for your artifact.

```bash
curl --request POST \
     --url https://api.confluent.cloud/artifact/v1/presigned-upload-url \
     --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
     --header 'content-type: application/json' \
     --data "${JSON_DATA}"
```

Your output should resemble:

```json
{
  "api_version": "artifact/v1",
  "kind": "PresignedUrl",
  "content_format": "JAR",
  "cloud": "AWS",
  "region": "us-east-1",
  "upload_id": "<upload-id-guid>",
  "upload_url": "https://confluent-custom-connectors-prod-us-east-1.s3.dualstack.us-east-1.amazonaws.com/",
  "upload_form_data": {
    "bucket": "confluent-custom-connectors-prod-us-east-1",
    "key": "staging/ccp/v1/<your-org-id>/custom-plugins/<guid>/plugin.jar",
    "policy": "ey…",
    "x-amz-algorithm": "AWS4-HMAC-SHA256",
    "x-amz-credential": "AS…/20241121/us-east-1/s3/aws4_request",
    "x-amz-date": "20241121T212232Z",
    "x-amz-security-token": "IQ…",
    "x-amz-signature": "52…"
  }
}
```

For convenience, save the security details in environment variables:

```bash
export UPLOAD_ID="<upload-id-guid>"
export UPLOAD_URL="<upload_url>"
export UPLOAD_BUCKET="<bucket>"
export UPLOAD_KEY="<key>"
export UPLOAD_POLICY="<policy>"
export UPLOAD_KEY="<key>"
export X_AMZ_ALGORITHM="<x-amz-algorithm>"
export X_AMZ_CREDENTIAL="<x-amz-credential>"
export X_AMZ_DATE="<x-amz-date>"
export X_AMZ_SECURITY_TOKEN="<x-amz-security-token>"
export X_AMZ_SIGNATURE="<x-amz-signature>"
```

After you have the presigned URL, ID, bucket policy, and other security
details, upload your JAR to the bucket. The following example provides a
curl command you can use to upload your JAR file.

#### NOTE
When specifying the JAR file to upload, you must use the `@` symbol
at the start of the file path. For example,
`-F file=@</path/to/upload/file>`. If you don’t use the `@` symbol,
you can see an error stating that
`Your proposed upload is smaller than the minimum allowed size.`

```bash
curl -X POST "${UPLOAD_URL}" \
  -F "bucket=${UPLOAD_BUCKET}" \
  -F "key=${UPLOAD_KEY}" \
  -F "policy=${UPLOAD_POLICY}" \
  -F "x-amz-algorithm=${X_AMZ_ALGORITHM}" \
  -F "x-amz-credential=${X_AMZ_CREDENTIAL}" \
  -F "x-amz-date=${X_AMZ_DATE}" \
  -F "x-amz-security-token=${X_AMZ_SECURITY_TOKEN}" \
  -F "x-amz-signature=${X_AMZ_SIGNATURE}" \
  -F file=@/path/to/udf_file.jar
```

After you upload your JAR file to the object store, you can create the
UDF in Confluent Cloud for Apache Flink by sending a POST request to the
[Create Artifact endpoint](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1)/operation/createArtifactV1FlinkArtifact).

The following JSON shows an example payload.

```json
{
  "cloud": "${CLOUD_PROVIDER}",
  "region": "${CLOUD_REGION}",
  "environment": "${ENV_ID}",
  "display_name": "${ARTIFACT_DISPLAY_NAME}",
  "class": "${CLASS_NAME}",
  "content_format": "JAR",
  "description": "${ARTIFACT_DESCRIPTION}",
  "documentation_link": "${ARTIFACT_DOC_LINK}",
  "runtime_language": "JAVA",
  "upload_source": {
    "location": "PRESIGNED_URL_LOCATION",
    "upload_id": "${UPLOAD_ID}"
  }
}
```

You must escape quotation mark characters in the JSON string, so the payload
string resembles the following:

```bash
export JSON_DATA="{
  \"cloud\": \"${CLOUD_PROVIDER}\",
  \"region\": \"${CLOUD_REGION}\",
  \"environment\": \"${ENV_ID}\",
  \"display_name\": \"${ARTIFACT_DISPLAY_NAME}\",
  \"class\": \"${CLASS_NAME}\",
  \"content_format\": \"JAR\",
  \"description\": \"${ARTIFACT_DESCRIPTION}\",
  \"documentation_link\": \"${ARTIFACT_DOC_LINK}\",
  \"runtime_language\": \"JAVA\",
  \"upload_source\": {
    \"location\": \"PRESIGNED_URL_LOCATION\",
    \"upload_id\": \"${UPLOAD_ID}\"
  }
}"
```

Run the following command to create the artifact in your environment.

```bash
curl --request POST \
     --url "https://api.confluent.cloud/artifact/v1/flink-artifacts?cloud=${CLOUD_REGION}&region=${CLOUD_REGION}&environment=${ENV_ID}" \
     --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
     --header 'content-type: application/json' \
     --data "${JSON_DATA}"
```

<a id="flink-rest-api-read-artifact"></a>

### Read an artifact

Get the details about an artifact in your environment by sending a GET request to the
[Read Artifact endpoint](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1)/operation/getArtifactV1FlinkArtifact).

- This request uses your Cloud API key instead of the Flink API key.

Getting details about an artifact requires the following inputs:

```bash
export ARTIFACT_ID="<artifact-id>" # example: cfa-e8rzq7
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

Run the following command to get details about the artifact specified by
the ARTIFACT_ID environment variable.

```bash
curl --request GET \
  --url "https://api.confluent.cloud/artifact/v1/flink-artifacts/${ARTIFACT_ID}?cloud=${CLOUD_PROVIDER}&region=${CLOUD_REGION}&environment=${ENV_ID}" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to get details about an artifact
</summary>
```json
{
  "api_version": "artifact/v1",
  "kind": "FlinkArtifact",
  "id": "cfa-e8rzq7",
  "metadata": {
    "self": "https://api.confluent.cloud/artifact/v1/flink-artifacts/cfa-e8rzq7",
    "resource_name": "crn://confluent.cloud/organization=<org-id>/flink-artifact=cfa-e8rzq7",
    "created_at": "2024-11-21T21:52:43.788042Z",
    "updated_at": "2024-11-21T21:52:44.625318Z"
  },
  "cloud": "AWS",
  "region": "us-east-1",
  "environment": "env-z3q9rd",
  "display_name": "udf_example",
  "class": "default",
  "content_format": "JAR",
  "description": "",
  "documentation_link": "",
  "runtime_language": "JAVA",
  "versions": [
    {
      "version": "ver-xq72dk",
      "release_notes": "",
      "is_beta": false,
      "artifact_id": {},
      "upload_source": {
        "location": "PRESIGNED_URL_LOCATION",
        "upload_id": ""
      }
    }
  ]
}
```

</details>
<br/>

<a id="flink-rest-api-update-artifact"></a>

### Update an artifact

Update an artifact in your environment by sending a PATCH request to the
[Update Artifact endpoint](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1)/operation/updateArtifactV1FlinkArtifact).

- This request uses your Cloud API key instead of the Flink API key.

Updating an artifact in your environment requires the following inputs:

```bash
export ARTIFACT_ID="<artifact-id>" # example: cfa-e8rzq7
export ARTIFACT_DISPLAY_NAME="<human-readable-name>" # example: "my-udf"
export ARTIFACT_DESCRIPTION="<description>" # example: "This is a demo UDF."
export ARTIFACT_DOC_LINK="<url-to-documentation>" # example: "https://docs.example.com/my-udf", "^$|^(http://|https://)."
export CLASS_NAME="<java-class-name>" # example: "io.confluent.example.SumScalarFunction"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

The following JSON shows an example payload.

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Request payload to update an artifact
</summary>
```json
{
  "cloud": "${CLOUD_PROVIDER}",
  "region": "${CLOUD_REGION}",
  "environment": "${ENV_ID}",
  "display_name": "${ARTIFACT_DISPLAY_NAME}",
  "content_format": "JAR",
  "description": "${ARTIFACT_DESCRIPTION}",
  "documentation_link": "${ARTIFACT_DOC_LINK}",
  "runtime_language": "JAVA",
  "versions": [
    {
      "version": "cfa-ver-001",
      "release_notes": "string",
      "is_beta": true,
      "artifact_id": {
        "cloud": "${CLOUD_PROVIDER}",
        "region": "${CLOUD_REGION}",
        "environment": "${ENV_ID}",
        "display_name": "${ARTIFACT_DISPLAY_NAME}",
        "class": "${CLASS_NAME}",
        "content_format": "JAR",
        "description": "${ARTIFACT_DESCRIPTION}",
        "documentation_link": "${ARTIFACT_DOC_LINK}",
        "runtime_language": "JAVA",
        "versions": [
          {}
        ]
      },
      "upload_source": {
        "location": "PRESIGNED_URL_LOCATION",
        "upload_id": "${UPLOAD_ID}"
      }
    }
  ]
}
```

</details>
<br/>

You must escape quotation mark characters in the JSON string, so the payload
string resembles the following:

```bash
export JSON_DATA="{
  \"cloud\": \"${CLOUD_PROVIDER}\",
  \"region\": \"${CLOUD_REGION}\",
  \"environment\": \"${ENV_ID}\",
  \"display_name\": \"${ARTIFACT_DISPLAY_NAME}\",
  \"content_format\": \"JAR\",
  \"description\": \"${ARTIFACT_DESCRIPTION}\",
  \"documentation_link\": \"${ARTIFACT_DOC_LINK}\",
  \"runtime_language\": \"JAVA\",
  \"versions\": [
    {
      \"version\": \"cfa-ver-001\",
      \"release_notes\": \"string\",
      \"is_beta\": true,
      \"artifact_id\": {
        \"cloud\": \"${CLOUD_PROVIDER}\",
        \"region\": \"${CLOUD_REGION}\",
        \"environment\": \"${ENV_ID}\",
        \"display_name\": \"${ARTIFACT_DISPLAY_NAME}\",
        \"class\": \"${CLASS_NAME}\",
        \"content_format\": \"JAR\",
        \"description\": \"${ARTIFACT_DESCRIPTION}\",
        \"documentation_link\": \"${ARTIFACT_DOC_LINK}\",
        \"runtime_language\": \"JAVA\",
        \"versions\": [
          {}
        ]
      },
      \"upload_source\": {
        \"location\": \"PRESIGNED_URL_LOCATION\",
        \"upload_id\": \"${UPLOAD_ID}\"
      }
    }
  ]
}"
```

Run the following command to update the artifact specified by
the ARTIFACT_ID environment variable.

```bash
curl --request PATCH \
     --url "https://api.confluent.cloud/artifact/v1/flink-artifacts/${ARTIFACT_ID}?cloud=${CLOUD_PROVIDER}&region=${CLOUD_REGION}&environment=${ENV_ID}" \
     --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
     --header 'content-type: application/json' \
     --data "${JSON_DATA}"
```

<a id="flink-rest-api-delete-artifact"></a>

### Delete an artifact

Delete an artifact in your environment by sending a DELETE request to the
[Delete Artifact endpoint](/cloud/current/api.html#tag/Flink-Artifacts-(artifactv1)/operation/deleteArtifactV1FlinkArtifact).

- This request uses your Cloud API key instead of the Flink API key.

Deleting an artifact in your environment requires the following inputs:

```bash
export ARTIFACT_ID="<artifact-id>" # example: cfa-e8rzq7
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

Run the following command to delete an artifact specified by the ARTIFACT_ID
environment variable.

```bash
curl --request DELETE \
  --url "https://api.confluent.cloud/artifact/v1/flink-artifacts/${ARTIFACT_ID}?cloud=${CLOUD_PROVIDER}&region=${CLOUD_REGION}&environment=${ENV_ID}" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}"
```

<a id="flink-rest-api-manage-connections"></a>

## Manage connections

To manage connections, you can use the following endpoints:

- [Create Connection](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/createConnectionV1Connection)
- [Delete Connection](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/deleteConnectionV1Connection)
- [Describe Connection](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/describeConnectionV1Connection)
- [List Connections](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/listConnectionsV1Connection)
- [Update Connection](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/updateConnectionV1Connection)

You need authorization to create, update, delete (`FlinkAdmin`) or use
(`FlinkDeveloper`) a connection. For more information, see [Grant Role-Based Access in Confluent Cloud for Apache Flink](flink-rbac.md#flink-rbac).

<a id="flink-rest-api-manage-connections-create"></a>

### Create a connection

Create a connection in your environment by sending a POST request to the
[Connections](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/createSqlv1Connection)
endpoint.

Creating a connection requires the following inputs. Credentials vary by
service.

```bash
export CONNECTION_NAME="<connection-name>" # example: "my-openai_connection"
export CONNECTION_TYPE="<connection-type>" # example: "OPENAI"
export ENDPOINT="<endpoint>" # example: "https://api.openai.com/v1/chat/completions"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-a1b2c3"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export JSON_DATA="<payload-string>"
```

The following JSON shows an example payload. The `auth_data` key varies by
service.

```json
{
  "name": "${CONNECTION_NAME}",
  "spec": {
    "connection_type": "${CONNECTION_TYPE}",
    "endpoint": "${ENDPOINT}",
    "auth_data": {
      "kind": "PlaintextProvider",
      "data": "string"
    }
  },
  "metadata": {}
}
```

Quotation mark characters in the JSON string must be escaped, so the payload
string to send resembles the following:

```bash
export JSON_DATA="{
  \"name\": \"${CONNECTION_NAME}\",
  \"spec\": {
    \"connection_type\": \"${CONNECTION_TYPE}\",
    \"endpoint\": \"${ENDPOINT}\",
    \"auth_data\": {
      \"kind\": \"PlaintextProvider\",
      \"data\": \"string\"
    }
  },
  \"metadata\": {}
}"
```

The following command sends a POST request to create a connection.

```bash
curl --request POST \
  --url "https://flink.region.provider.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/connections" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to create a connection
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "Connection",
  "metadata": {
    "self": "https://flink.us-west1.aws.confluent.cloud/sql/v1/organizations/org-abc/environments/env-a1b2c3/connections/my-openai_connection",
    "resource_name": "",
    "created_at": "2006-01-02T15:04:05-07:00",
    "updated_at": "2006-01-02T15:04:05-07:00",
    "deleted_at": "2006-01-02T15:04:05-07:00",
    "uid": "12345678-1234-1234-1234-123456789012",
    "resource_version": "a23av"
  },
  "name": "my-openai_connection",
  "spec": {
    "connection_type": "OPENAI",
    "endpoint": "https://api.openai.com/v1/chat/completions",
    "auth_data": {
      "kind": "PlaintextProvider",
      "data": "string"
    }
  },
  "status": {
    "phase": "READY",
    "detail": "Lookup failed: ai.openai.com"
     }
   }
}
```

<a id="flink-rest-api-manage-connections-delete"></a>

### Delete a connection

Delete a connection in your environment by sending a DELETE request to the
[Connections endpoint](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/deleteSqlv1Connection).

- This request uses your Cloud API key instead of the Flink API key.

Deleting a connection requires the following inputs:

```bash
export CONNECTION_NAME="<connection-name>" # example: "my-openai_connection"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-a1b2c3"
```

Run the following command to delete the connection specified in the
CONNECTION_NAME environment variable.

```bash
curl --request DELETE \
  --url "https://flink.region.provider.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/connections/${CONNECTION_NAME}" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}"
```

<a id="flink-rest-api-manage-connections-describe"></a>

### Describe a connection

Get the details about a connection in your environment by sending a GET request to the
[Connections endpoint](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/getSqlv1Connection).

- This request uses your Cloud API key instead of the Flink API key.

Getting details about a connection requires the following inputs:

```bash
export CONNECTION_NAME="<connection-name>" # example: "my-openai_connection"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-a1b2c3"
```

Run the following command to get details about the connection specified in
the CONNECTION_NAME environment variable.

```bash
curl --request GET \
  --url "https://flink.region.provider.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/connections/${CONNECTION_NAME}" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to read a connection
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "Connection",
  "metadata": {
    "self": "https://flink.us-west1.aws.confluent.cloud/sql/v1/organizations/org-abc/environments/env-123/connections/my-openai_connection",
    "resource_name": "",
    "created_at": "2006-01-02T15:04:05-07:00",
    "updated_at": "2006-01-02T15:04:05-07:00",
    "deleted_at": "2006-01-02T15:04:05-07:00",
    "uid": "12345678-1234-1234-1234-123456789012",
    "resource_version": "a23av"
  },
  "name": "my-openai_connection",
  "spec": {
    "connection_type": "OPENAI",
    "endpoint": "https://api.openai.com/v1/chat/completions",
    "auth_data": {
      "kind": "PlaintextProvider",
      "data": "string"
    }
  },
  "status": {
    "phase": "READY",
    "detail": "Lookup failed: ai.openai.com"
  }
}
```

<a id="flink-rest-api-manage-connections-list"></a>

### List connections

List the connections in your environment by sending a GET request to the
[Connections endpoint](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/listSqlv1Connections).

- This request uses your Cloud API key instead of the Flink API key.

Listing the connections in your environment requires the following inputs:

```bash
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-a1b2c3"
```

Run the following command to list the connections in your environment.

```bash
curl --request GET \
  --url "https://flink.region.provider.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/connections" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to list connections
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "ConnectionList",
  "metadata": {
    "first": "https://flink.us-west1.aws.confluent.cloud/sql/v1/environments/env-abc123/connections",
    "last": "",
    "prev": "",
    "next": "https://flink.us-west1.aws.confluent.cloud/sql/v1/environments/env-abc123/connections?page_token=UvmDWOB1iwfAIBPj6EYb",
    "total_size": 123,
    "self": "https://flink.us-west1.aws.confluent.cloud/sql/v1/environments/env-123/connections"
  },
  "data": [
    {
      "api_version": "sql/v1",
      "kind": "Connection",
      "metadata": {
        "self": "https://flink.us-west1.aws.confluent.cloud/sql/v1/organizations/org-abc/environments/env-123/connections/my-openai_connection",
        "resource_name": "",
        "created_at": "2006-01-02T15:04:05-07:00",
        "updated_at": "2006-01-02T15:04:05-07:00",
        "deleted_at": "2006-01-02T15:04:05-07:00",
        "uid": "12345678-1234-1234-1234-123456789012",
        "resource_version": "a23av"
      },
      "name": "my-openai_connection",
      "spec": {
        "connection_type": "OPENAI",
        "endpoint": "https://api.openai.com/v1/chat/completions",
        "auth_data": {
          "kind": "PlaintextProvider",
          "data": "string"
        }
     }
   },
   "status": {
     "phase": "READY",
     "detail": "Lookup failed: ai.openai.com"
      }
    }
  ]
}
```

<a id="flink-rest-api-manage-connections-update"></a>

### Update a connection

Update a connection in your environment by sending a PATCH request to the
[Connections endpoint](/cloud/current/api.html#tag/Connections-(sqlv1)/operation/updateSqlv1Connection).

- This request uses your Cloud API key instead of the Flink API key.

Updating a connection requires the following inputs:

```bash
export CONNECTION_NAME="<connection-name>" # example: "my-openai_connection"
export CONNECTION_TYPE="<connection-type>" # example: "OPENAI"
export ENDPOINT="<endpoint>" # example: "https://api.openai.com/v1/chat/completions"
export CLOUD_API_KEY="<cloud-api-key>"
export CLOUD_API_SECRET="<cloud-api-secret>"
export BASE64_CLOUD_KEY_AND_SECRET=$(echo -n "${CLOUD_API_KEY}:${CLOUD_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-a1b2c3"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export JSON_DATA="<payload-string>"
```

The following JSON shows an example payload. The `auth_data` key varies by
service.

```json
{
  "name": "${CONNECTION_NAME}",
  "spec": {
    "connection_type": "${CONNECTION_TYPE}",
    "endpoint": "${ENDPOINT}",
    "auth_data": {
      "kind": "PlaintextProvider",
      "data": "string"
    }
  },
  "metadata": {}
}
```

Quotation mark characters in the JSON string must be escaped, so the payload
string to send resembles the following:

```bash
export JSON_DATA="{
  \"name\": \"${CONNECTION_NAME}\",
  \"spec\": {
    \"connection_type\": \"${CONNECTION_TYPE}\",
    \"endpoint\": \"${ENDPOINT}\",
    \"auth_data\": {
      \"kind\": \"PlaintextProvider\",
      \"data\": \"string\"
    }
  },
  \"metadata\": {}
}"
```

The following command sends a PUT request to update a connection.

```bash
curl --request PUT \
  --url "https://flink.region.provider.confluent.cloud/sql/v1/organizations/${ORG_ID}/environments/${ENV_ID}/connections/${CONNECTION_NAME}" \
  --header "Authorization: Basic ${BASE64_CLOUD_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

Your output should resemble:

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  Response from a request to update a connection
</summary>
```json
{
  "api_version": "sql/v1",
  "kind": "Connection",
  "metadata": {
    "self": "https://flink.us-west1.aws.confluent.cloud/sql/v1/organizations/org-abc/environments/env-a1b2c3/connections/my-openai_connection",
    "resource_name": "",
    "created_at": "2006-01-02T15:04:05-07:00",
    "updated_at": "2006-01-02T15:04:05-07:00",
    "deleted_at": "2006-01-02T15:04:05-07:00",
    "uid": "12345678-1234-1234-1234-123456789012",
    "resource_version": "a23av"
  },
  "name": "my-openai_connection",
  "spec": {
    "connection_type": "OPENAI",
    "endpoint": "https://api.openai.com/v1/chat/completions",
    "auth_data": {
      "kind": "PlaintextProvider",
      "data": "string"
    }
  },
  "status": {
    "phase": "READY",
    "detail": "Lookup failed: ai.openai.com"
     }
   }
}
```

<a id="flink-rest-api-manage-workspaces"></a>

## Manage workspaces

A [workspace](../get-started/quick-start-cloud-console.md#flink-sql-quick-start-create-workspace) is a named,
ordered collection of Flink SQL code blocks. Using requests to the Flink
REST API, you can perform these actions:

- [List workspaces](#flink-rest-api-list-workspaces)
- [Create a workspace](#flink-rest-api-create-workspace)
- [Read a workspace](#flink-rest-api-read-workspace)
- [Update a workspace](#flink-rest-api-update-workspace)
- [Patch a workspace](#flink-rest-api-patch-workspace)
- [Delete a workspace](#flink-rest-api-delete-workspace)

### Workspace schema

A workspace has the following schema:

```yaml
api_version: "ws/v1"
kind: "Workspace"
organization_id: "" # Identifier of your Confluent Cloud organization
environment_id: "" # Identifier of your Confluent Cloud environment
name: "" # Primary identifier of the workspace, unique within the environment and region, 253 max length, [a-z0-9]([-.a-z0-9]*[a-z0-9])?
metadata:
  self: "" # An absolute URL to this resource
  uid: "" # Generated identifier that is unique across space and time
  resource_version: "" # Opaque version used for optimistic locking; required when updating the workspace
  created_by: "" # Principal ID of the user who created the workspace
  created_at: "" # Creation timestamp of this resource
  updated_at: "" # Last updated timestamp of this resource
spec:
  display_name: "" # A human-readable display name for the workspace, or generated if absent
  compute_pool: # Optional. The compute pool that clients can use to run the workspace's code blocks
    id: "" # example: "lfcp-8m03rm"
  service_account: # Optional. The service account that clients can use to run the workspace's code blocks
    id: "" # example: "sa-23kgz4"
  blocks: [] # Optional. The ordered code blocks for the workspace, 20 max
  properties: map[string]string # Optional. Key-value pairs that clients choose whether and how to use
status:
  network_kind: "" # Networking type used by the workspace (PUBLIC or PRIVATE)
```

The workspace name can have up to 253 characters and must match
the following regular expression:

```none
[a-z0-9]([-.a-z0-9]*[a-z0-9])?
```

<a id="flink-rest-api-list-workspaces"></a>

### List workspaces

List the workspaces in an environment by sending a GET request to the
[Workspaces endpoint](/cloud/current/api.html#tag/Workspaces-(ws/v1)/operation/listWsV1Workspaces).

By default, the response includes only the workspaces that you created. To
list every workspace in the environment, set the `all` query parameter to
`true`. This requires the `OrgAdmin` or `EnvAdmin` role.

Request query parameters
: - `spec.compute_pool` (optional): Filter the results to workspaces that
    use this compute pool ID.
  - `all` (optional): If `true` and the caller has the `OrgAdmin` or
    `EnvAdmin` role, return every workspace in the environment. Otherwise
    return only the workspaces created by the caller (default).
  - `page_token` (optional): Retrieve a page based on a token
    received earlier (from the `metadata.next` field of the response).
  - `page_size` (optional): The most items to return per page.

Listing the workspaces in your environment requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
```

Run the following command to list the workspaces that you created in the
environment.

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/ws/v1/organizations/${ORG_ID}/environments/${ENV_ID}/workspaces" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

Your output should resemble:

```json
{
  "api_version": "ws/v1",
  "kind": "WorkspaceList",
  "metadata": {
    "self": "https://flink.us-east-1.aws.confluent.cloud/ws/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/workspaces"
  },
  "data": [
    {
      "api_version": "ws/v1",
      "kind": "Workspace",
      "metadata": {
        "self": "https://flink.us-east-1.aws.confluent.cloud/ws/v1/organizations/b0b21724-4586-4a07-b787-d0bb5aacbf87/environments/env-z3y2x1/workspaces/my-workspace",
        "uid": "8241081c-a167-44ed-a920-7ec8f8473ee2",
        "resource_version": "YOp5OKKb6w6sm3IfizBj",
        "created_at": "2026-01-15T17:12:08.914198Z",
        "updated_at": "2026-01-15T17:12:08.914198Z"
      },
      "organization_id": "b0b21724-4586-4a07-b787-d0bb5aacbf87",
      "environment_id": "env-z3y2x1",
      "name": "my-workspace",
      "spec": {
        "display_name": "My first workspace",
        "compute_pool": {
          "id": "lfcp-8m03rm"
        }
      },
      "status": {
        "network_kind": "PUBLIC"
      }
    }
  ]
}
```

<a id="flink-rest-api-create-workspace"></a>

### Create a workspace

You can create a workspace by sending a POST request to the
[Workspaces endpoint](/cloud/current/api.html#tag/Workspaces-(ws/v1)/operation/createWsV1Workspace).

Creating a workspace requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export WORKSPACE_NAME="<workspace-name>" # example: "my-workspace"
export COMPUTE_POOL_ID="<compute-pool-id>" # (optional) example: "lfcp-8m03rm"
```

The following JSON shows an example payload that creates a workspace with a
single Flink SQL code block:

```json
{
  "name": "${WORKSPACE_NAME}",
  "spec": {
    "display_name": "My first workspace",
    "compute_pool": {
      "id": "${COMPUTE_POOL_ID}"
    },
    "blocks": [
      {
        "type": "CODE",
        "code_options": {
          "source": ["SELECT * FROM orders;"]
        }
      }
    ]
  }
}
```

You must escape quotation mark characters in the JSON string, so the
payload string to send resembles the following:

```bash
export JSON_DATA="{
  \"name\": \"${WORKSPACE_NAME}\",
  \"spec\": {
    \"display_name\": \"My first workspace\",
    \"compute_pool\": {
      \"id\": \"${COMPUTE_POOL_ID}\"
    },
    \"blocks\": [
      {
        \"type\": \"CODE\",
        \"code_options\": {
          \"source\": [\"SELECT * FROM orders;\"]
        }
      }
    ]
  }
}"
```

The following command sends a POST request that creates the workspace.

```bash
curl --request POST \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/ws/v1/organizations/${ORG_ID}/environments/${ENV_ID}/workspaces" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

<a id="flink-rest-api-read-workspace"></a>

### Read a workspace

Get the details about a workspace by sending a GET request to the
[Workspaces endpoint](/cloud/current/api.html#tag/Workspaces-(ws/v1)/operation/getWsV1Workspace).

Getting a workspace’s details requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export WORKSPACE_NAME="<workspace-name>" # example: "my-workspace"
```

The following command gets a workspace’s details by its name.

```bash
curl --request GET \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/ws/v1/organizations/${ORG_ID}/environments/${ENV_ID}/workspaces/${WORKSPACE_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

<a id="flink-rest-api-update-workspace"></a>

### Update a workspace

Update a workspace by sending a PUT request to the
[Workspaces endpoint](/cloud/current/api.html#tag/Workspaces-(ws/v1)/operation/putWsV1Workspace).

- You must specify the workspace’s current `resource_version` in the
  payload metadata.
- The request fails with a `409` error if the workspace changed since you
  fetched it. If this happens, send a GET request, reapply your
  modifications, and retry the update.

Updating a workspace requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export WORKSPACE_NAME="<workspace-name>" # example: "my-workspace"
export RESOURCE_VERSION="<version>" # must be fetched from the latest version of the workspace
export NEW_COMPUTE_POOL_ID="<new-compute-pool-id>" # example: "lfcp-8m09g0"
```

The following JSON shows an example payload that renames the workspace and
moves it to a different compute pool:

```json
{
  "metadata": {
    "resource_version": "${RESOURCE_VERSION}"
  },
  "spec": {
    "display_name": "My renamed workspace",
    "compute_pool": {
      "id": "${NEW_COMPUTE_POOL_ID}"
    }
  }
}
```

You must escape quotation mark characters in the JSON string, so the
payload string to send resembles the following:

```bash
export JSON_DATA="{
  \"metadata\": {
    \"resource_version\": \"${RESOURCE_VERSION}\"
  },
  \"spec\": {
    \"display_name\": \"My renamed workspace\",
    \"compute_pool\": {
      \"id\": \"${NEW_COMPUTE_POOL_ID}\"
    }
  }
}"
```

Run the following command to send a PUT request that updates the
workspace.

```bash
curl --request PUT \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/ws/v1/organizations/${ORG_ID}/environments/${ENV_ID}/workspaces/${WORKSPACE_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header 'content-type: application/json' \
  --data "${JSON_DATA}"
```

<a id="flink-rest-api-patch-workspace"></a>

### Patch a workspace

Partially update a workspace by sending a PATCH request to the
[Workspaces endpoint](/cloud/current/api.html#tag/Workspaces-(ws/v1)/operation/patchWsV1Workspace).

The PATCH request uses JSON Patch format as specified in
[RFC 6902](https://tools.ietf.org/html/rfc6902), which lets you add,
replace, or remove specific fields without sending the full workspace
representation or a resource version.

Patching a workspace requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export WORKSPACE_NAME="<workspace-name>" # example: "my-workspace"
```

The following command renames a workspace by replacing its
`spec.display_name` field.

```bash
curl --request PATCH \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/ws/v1/organizations/${ORG_ID}/environments/${ENV_ID}/workspaces/${WORKSPACE_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}" \
  --header "Content-Type: application/json-patch+json" \
  --data '[{"path":"/spec/display_name","op":"replace","value":"My renamed workspace"}]'
```

<a id="flink-rest-api-delete-workspace"></a>

### Delete a workspace

Delete a workspace by sending a DELETE request to the
[Workspaces endpoint](/cloud/current/api.html#tag/Workspaces-(ws/v1)/operation/deleteWsV1Workspace).

- After Confluent Cloud deletes a workspace, you can’t undo the deletion.
- Deleting a workspace doesn’t stop or delete the Flink SQL statements
  that its code blocks might have submitted.

Deleting a workspace requires the following inputs:

```bash
export FLINK_API_KEY="<flink-api-key>"
export FLINK_API_SECRET="<flink-api-secret>"
export BASE64_FLINK_KEY_AND_SECRET=$(echo -n "${FLINK_API_KEY}:${FLINK_API_SECRET}" | base64 -w 0)
export ORG_ID="<organization-id>" # example: "b0b21724-4586-4a07-b787-d0bb5aacbf87"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export WORKSPACE_NAME="<workspace-name>" # example: "my-workspace"
```

The following command deletes a workspace in the specified organization
and environment.

```bash
curl --request DELETE \
  --url "https://flink.${CLOUD_REGION}.${CLOUD_PROVIDER}.confluent.cloud/ws/v1/organizations/${ORG_ID}/environments/${ENV_ID}/workspaces/${WORKSPACE_NAME}" \
  --header "Authorization: Basic ${BASE64_FLINK_KEY_AND_SECRET}"
```

## Related content

- [Cloud Console](../get-started/quick-start-cloud-console.md#flink-sql-quick-start-run-sql-statement)
- [Confluent CLI](../reference/flink-sql-cli.md#flink-sql-confluent-cli)
- [SQL shell](../get-started/quick-start-shell.md#flink-sql-quick-start-shell)
- [Confluent Terraform Provider](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs)

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