<a id="manage-connections"></a>

# Manage Connections with External Services in Confluent Cloud

#### NOTE
Currently, the only connections supported are Flink connections. As more connection
types are added, this page will be updated to reflect the changes.

Confluent Cloud offers provider integrations with Confluent Cloud for Apache Flink to provide security and
reusability of Flink connections for Flink statements. A Flink connection defines
the configuration parameters required to connect to a specific third-party
service, assuming a network path is available for that service. A Flink
connection defines information about the endpoint and authentication type
required to connect to the third-party service. You can reuse a Flink connection
across multiple Flink statements when explicit permissions are granted to the
Flink application.

Connections support a broad range of third-party services, including but not
limited to AI/ML providers, databases, cloud storage platforms, messaging systems,
databases (search and vector), APIs, data catalogs, notification tools, and secrets
management systems. This flexibility supports diverse streaming data workloads that
need access to both cloud-native and external enterprise data services.

Connections can be used in Confluent Cloud environments that use private networking. For
details on using Flink connections in a private network, see [Private Networking with Confluent Cloud for Apache Flink](../../flink/concepts/flink-private-networking.md#flink-sql-private-networking).

For conceptual information about connections, supported services, and security
considerations, see [Reuse Confluent Cloud Connections With External Services](overview.md#connections-overview).

<a id="manage-connections-prerequisites"></a>

## Prerequisites

Before creating and using Flink connections, ensure you have:

- A Confluent Cloud account with appropriate permissions to create and manage Flink
  connections. For details on required roles, see [Security](overview.md#connections-security).
- Network access to the target third-party service.
- Required authentication credentials for the target service.
- Understanding of the target service’s connection requirements and limitations.

<a id="connection-types"></a>

## Connection types

Using a Confluent Cloud provider integration, you can currently create and use the following
connection types for Flink statements:

### Flink AI connection

Confluent Cloud for Apache Flink supports AI model inference and enables you to use your models as
resources in your Flink SQL statements, just like you would use a table or a
function. You can use Flink AI connections to integrate with the following
external service providers:

* OpenAI
* Azure ML
* Azure OpenAI
* Amazon Bedrock
* Amazon SageMaker
* Couchbase
* Elasticsearch
* Google AI
* Google Vertex AI
* Pinecone

For details about configuration parameters for each service, see [Reuse Confluent Cloud Connections With External Services](overview.md#connections-overview).

To learn more about using Flink AI with Confluent Cloud, see:

* [Run an AI Model with Confluent Cloud](../../ai/ai-model-inference.md#flink-sql-ai-model)
* [Flink AI: Real-Time ML and GenAI Enrichment of Streaming Data with Flink SQL on Confluent Cloud](https://www.confluent.io/blog/flink-ai-real-time-ml-and-genai-enrichment-of-streaming-data-with-flink-sql-on-confluent-cloud/)

You can use the Confluent Cloud Console, Confluent CLI, and Confluent Cloud APIs to
create, delete, and describe your connections.

For details on the Confluent CLI `confluent flink connection` commands, see
[confluent flink connection](https://docs.confluent.io/confluent-cli/current/command-reference/flink/connection/index.html#confluent-flink-connection).

## Create a connection

Follow the steps in the following tabs to create a connection that defines an
endpoint and authentication method for Flink to connect to external services.
After you create a connection, you can reuse it in the same environment to
connect to other external services.

### Cloud Console

To create a connection using the Confluent Cloud Console, follow these steps:

1. Open the Confluent Cloud Console at [https://confluent.cloud](https://confluent.cloud).
2. Go to the environment where you want to create the connection.
3. In the left pane, click **Integrations**.
4. Click the **Connections** tab.
5. Click **Add connection**.

   The **Service Provider** page displays the list of supported service
   providers.
6. Click the service provider you want to use, and then click **Continue**.

   The **Authentication** page displays.
7. Enter values to define the endpoint and credentials for the connection,
   and then click **Continue**.

   The **Review** page displays.
8. Enter the requested information for the connection, and then click
   **Create connection**.

   | Connection name   | Enter a meaningful name for the connection.               |
   |-------------------|-----------------------------------------------------------|
   | Cloud provider    | Select the cloud service provider from the dropdown list. |
   | Region            | Select the region from the dropdown list.                 |

The **Integrations** page displays the list of connections, including the
connection you just created.

### Confluent CLI

Use the following command to create a connection, replacing the placeholders
with your values:

```shell
confluent flink connection create \
--name <connection-name> \
--type <connection-type> \
--config <config-file>
```

Example configuration file for OpenAI (`openai-config.json`):

```json
{
"endpoint": "https://api.openai.com/v1/chat/completions",
"api.key": "${secret:openai-api-key}"
}
```

Example configuration file for Amazon Bedrock (`bedrock-config.json`):

```json
{
"endpoint": "https://bedrock-runtime.us-east-1.amazonaws.com",
"aws.access.key.id": "${secret:aws-access-key}",
"aws.secret.access.key": "${secret:aws-secret-key}"
}
```

For details on the Confluent CLI `confluent flink connection create`
command, see [confluent flink connection create](https://docs.confluent.io/confluent-cli/current/command-reference/flink/connection/confluent_flink_connection_create.html).

### REST API

To create a connection using the REST API, use the following API endpoint:

```bash
POST /flink/v1/connections
Content-Type: application/json

{
"name": "<connection-name>",
"type": "<connection-type>",
"config": {
   "endpoint": "https://api.openai.com/v1/chat/completions",
   "api.key": "${secret:openai-api-key}"
}
}
```

Replace the placeholders with your values. For example, replace
`<connection-name>` with `my_openai_connection` and
`<connection-type>` with `ai`.

Example response:

```json
{
"name": "my_openai_connection",
"type": "ai",
"config": {
   "endpoint": "https://api.openai.com/v1/chat/completions",
   "api.key": "${secret:openai-api-key}"
}
}
```

For details on creating a connection using the POST method, see
[Create a Connection](https://docs.confluent.io/cloud/current/ccloud/create-sqlv-1-connection/)
in the Confluent Cloud APIs Reference.

## Delete a connection

You can delete a connection using the Confluent Cloud Console, Confluent CLI,
and Confluent Cloud APIs.

### Cloud Console

To delete a connection using the Confluent Cloud Console, follow these steps:

1. Open the Confluent Cloud Console at [https://confluent.cloud](https://confluent.cloud).
2. Go to the environment where you want to delete the connection.
3. In the left pane, click **Integrations**.
4. Click the **Connections** tab.
5. Click the connection you want to delete.
6. Click **Delete**.

### Confluent CLI

To delete a connection using the Confluent CLI, use the
`confluent flink connection delete` command:

```bash
confluent flink connection delete <connection-name>
```

Replace the placeholders with your values. For example, replace
`<connection-name>` with `my_openai_connection`.

For details on the Confluent CLI `confluent flink connection delete`
command, see [confluent flink connection delete](https://docs.confluent.io/confluent-cli/current/command-reference/flink/connection/confluent_flink_connection_delete.html).

### REST API

To delete a connection using the REST API, use the following API endpoint:

```bash
DELETE /flink/v1/connections/{connection-name}
```

Replace the placeholders with your values. For example, replace
`<connection-name>` with `my_openai_connection`.

Example response:

```json
{
"status": "deleted"
}
```

For details on deleting a connection using the DELETE method, see
[Delete a Connection](https://docs.confluent.io/cloud/current/ccloud/delete-sqlv-1-connection/)
in the Confluent Cloud APIs Reference.

## Describe a connection

You can describe a connection using the Confluent Cloud Console, Confluent CLI,
and Confluent Cloud APIs.

### Cloud Console

To describe a connection using the Confluent Cloud Console, follow these steps:

1. Open the Confluent Cloud Console at [https://confluent.cloud](https://confluent.cloud).
2. Go to the environment where you want to describe the connection.
3. In the left pane, click **Integrations**.

The **Integrations** page displays the list of connections, including details
about the connection you want to describe. You can view the following
connection details:

- Connection name
- Connection type (service)
- Cloud provider
- Region
- Date created
- Date modified

1. Click the connection name to view details.

### Confluent CLI

To describe a connection using the Confluent CLI, use the
`confluent flink connection describe` command:

```bash
confluent flink connection describe <connection-name>
```

Replace the placeholder `<connection-name>` with your value. For example, replace
`<connection-name>` with `my_openai_connection`.

For details on the Confluent CLI `confluent flink connection describe`
command, see [confluent flink connection describe](https://docs.confluent.io/confluent-cli/current/command-reference/flink/connection/confluent_flink_connection_describe.html).

### REST API

To describe a connection using the REST API, use the following GET method:

```bash
GET /flink/v1/connections/{connection-name}
```

Replace the placeholder `<connection-name>` with your value. For example, replace
`<connection-name>` with `my_openai_connection`.

Example response:

```json
{
"name": "my_openai_connection",
"type": "ai",
"config": {
   "endpoint": "https://api.openai.com/v1/chat/completions",
   "api.key": "${secret:openai-api-key}"
}
}
```

For details on describing a connection using the GET method, see
[Read a Connection](https://docs.confluent.io/cloud/current/ccloud/get-sqlv-1-connection/)
in the Confluent Cloud APIs Reference.

## List connections

You can list all connections in a specified environment using the
Confluent Cloud Console, Confluent CLI, and Confluent Cloud APIs.

### Cloud Console

To view a list of connections using the Confluent Cloud Console, follow these steps:

1. Open the Confluent Cloud Console at [https://confluent.cloud](https://confluent.cloud).
2. Go to the environment where you want to view connections.
3. In the left pane, click **Integrations**.
4. Click the **Connections** tab.
5. View the list of all connections.
6. Use filters to search or sort connections.

### Confluent CLI

To list connections using the Confluent CLI, use the
`confluent flink connection list` command:

```bash
confluent flink connection list
```

For details on the Confluent CLI `confluent flink connection list`
command, see [confluent flink connection list](https://docs.confluent.io/confluent-cli/current/command-reference/flink/connection/confluent_flink_connection_list.html).

### REST API

To view a list of connections, use the following GET method:

```bash
GET /flink/v1/connections
```

Example response:

```json
{
"connections": [
   {
      "name": "my_openai_connection",
      "type": "ai"
   },
   {
      "name": "my_bedrock_connection",
      "type": "ai"
   }
]
}
```

For details on listing connections using the GET method, see
[List Connections](https://docs.confluent.io/cloud/current/ccloud/list-sqlv-1-connections/)
in the Confluent Cloud APIs Reference.

## Update a connection

You can update the configuration parameters for a Flink Connection using the
Confluent Cloud Console, Confluent CLI, and Confluent Cloud APIs.

### Cloud Console

To update a connection using the Confluent Cloud Console, follow these steps:

1. Open the Confluent Cloud Console at [https://confluent.cloud](https://confluent.cloud).
2. Go to the environment where you want to update the connection.
3. In the left pane, click **Integrations**.
4. Click the **Connections** tab.
5. Click the connection to update.
6. Click **Edit**.
7. Modify the configuration, and then click **Save**.

### Confluent CLI

To update a connection using the Confluent CLI, use the
`confluent flink connection update` command:

```bash
confluent flink connection update <connection-name> \
--config <updated-config-file>
```

Replace the placeholders with your values. For example, replace
`<connection-name>` with `my_openai_connection` and
`<updated-config-file>` with `updated-openai-config.json`.

For details on the Confluent CLI `confluent flink connection update`
command, see [confluent flink connection update](https://docs.confluent.io/confluent-cli/current/command-reference/flink/connection/confluent_flink_connection_update.html).

### REST API

To update a connection using the REST API, use the following PUT method:

```bash
PUT /flink/v1/connections/{connection-name}
Content-Type: application/json

{
"config": {
   "endpoint": "https://api.openai.com/v1/chat/completions",
   "api.key": "${secret:updated-openai-api-key}"
}
}
```

Replace the placeholders with your values. For example, replace

Example response:

```json
{
"status": "updated"
}
```

For details on updating a connection using the PUT method, see
[Update a Connection](https://docs.confluent.io/cloud/current/ccloud/update-sqlv-1-connection/)
in the Confluent Cloud APIs Reference.

## Use a connection in Flink statements

After creating a connection, you can use it in your Flink SQL statements to interact
with external services. The following examples show how to use connections in different
scenarios:

**Using an AI connection for model inference:**

```sql
CREATE MODEL my_sentiment_model
INPUT (review_text STRING)
OUTPUT (sentiment STRING, confidence DOUBLE)
WITH (
  'provider' = 'openai',
  'connection' = 'my_openai_connection',
  'model' = 'gpt-3.5-turbo',
  'task' = 'classification'
);
```

**Using a vector database connection:**

```sql
CREATE TABLE embeddings_table (
  id STRING,
  vector ARRAY<DOUBLE>,
  metadata MAP<STRING, STRING>
) WITH (
  'connector' = 'pinecone',
  'connection' = 'my_pinecone_connection',
  'index.name' = 'my-index'
);
```

For more examples and detailed usage instructions, see the Confluent documentation
for specific Flink AI functions and connectors.

## Troubleshooting

Common issues and solutions when working with connections:

### Authentication errors

If you encounter authentication errors:

* Verify that your API keys or credentials are correct and active.
* Check that the secret references (`${secret:key-name}`) point to valid secrets.
* Ensure your service account has the necessary permissions.
* For AWS services, verify that IAM roles and policies are correctly configured.

### Connection timeouts

If connections time out:

* Verify network connectivity to the external service.
* Check if the service endpoint URL is correct and accessible.
* For private networking setups, ensure proper network configuration.
* Review service-specific rate limits and quotas.

### Invalid configuration

If you receive configuration errors:

* Verify that all required configuration parameters are provided.
* Check the service-specific configuration requirements in [Reuse Confluent Cloud Connections With External Services](overview.md#connections-overview).
* Ensure the connection type matches the target service.
* Validate JSON syntax in configuration files.

## Best practices

Follow these best practices when working with connections:

### Security

* Use secrets management to store sensitive credentials like API keys.
* Regularly rotate API keys and access credentials.
* Use the principle of least privilege when configuring service permissions.
* Monitor connection usage through Confluent Cloud audit logs.

### Performance

* Reuse connections across multiple Flink statements in the same environment.
* Monitor connection usage and adjust rate limits as needed.
* Configure appropriate timeout values for external service calls.
* Test connections in development environments before production deployment.

### Management

* Use descriptive names for connections to make them easy to identify.
* Document connection purposes and dependencies.
* Implement proper lifecycle management for connections.
* Monitor connection health and status regularly.

## Related content

- [External Services for Confluent Cloud Connections](services.md#connections-services)
- [Reuse Confluent Cloud Connections With External Services](overview.md#connections-overview)
- [Run an AI Model with Confluent Cloud](../../ai/ai-model-inference.md#flink-sql-ai-model)
- [Flink AI: Real-Time ML and GenAI Enrichment of Streaming Data with Flink SQL on Confluent Cloud [Confluent Blog]](https://www.confluent.io/blog/flink-ai-real-time-ml-and-genai-enrichment-of-streaming-data-with-flink-sql-on-confluent-cloud/)
- [How to generate vector embeddings for RAG (GenAI) with Flink SQL in Confluent Cloud [Confluent Developer]](https://developer.confluent.io/confluent-tutorials/gen-ai-vector-embedding/flinksql/)