# Manage Service Accounts for Connectors in Confluent Cloud All Confluent Cloud connectors require credentials to allow the connector to operate and access Kafka. You can either create and use an API key and secret or use a service account. When you create a service account, you configure access control list (ACL) DESCRIBE, CREATE, READ, and WRITE access to topics and create the API key and secret. Once the service account is created, a user creating a connector can select the service account ID when configuring the connector. #### IMPORTANT A connector configuration must include either an API key and secret or a service account ID. For additional Confluent Cloud service account information, see [Service Accounts on Confluent Cloud](../security/authenticate/workload-identities/service-accounts/overview.md#service-accounts). ## Create a service account using the Confluent Cloud Console When you create a new connector using the Cloud Console, you have the option to select an existing service account or create a new one. Complete the following steps to create a new service account while creating a new connector. For additional Confluent Cloud service account information, see [Service Accounts on Confluent Cloud](../security/authenticate/workload-identities/service-accounts/overview.md#service-accounts). 1. Select your new connector from the **Connector Plugins** screen and, if applicable, select an existing topic or create a new topic. 2. Click **Service account** on the **Kafka credentials** screen. ![Select service account](images/ccloud-connect-service-account-1.png) 3. You can select an existing service account or create a new one. Click **Create a new one**. Create a name and description for the service account. ![Create a new service account](images/ccloud-connect-service-account-2.png) 4. Click **Add all required ACLs…** to be sure that the connector can read, create, and write to any topics it may require for operation. 5. Click **Continue** and configure the connector. Your new service account ID is similar to the ID `sa-vky17n` highlighted below. ![Service account ID](images/ccloud-connect-service-account-3.png) ## Create a service account using the Confluent CLI The following examples show how to set up a service account using the [Confluent Cloud CLI](https://docs.confluent.io/confluent-cli/current/overview.html). These steps can be used for a cluster running on any cloud provider. ### Sink connector service account This example assumes the following: * You have a Kafka cluster with cluster ID `lkc-abcd123`. * You want the sink connector to read from a topic named `pageviews`. Use the following example steps to create a service account, set ACLs, and add the API key and secret. #### NOTE The following steps show basic ACL entries for sink connector service accounts. Be sure to review the [Sink connector SUCCESS and ERROR topics](#cloud-service-account-sink-additional-acls) and [Sink connector offset management](#cloud-service-account-sink-offset-management-acls) sections for additional ACL entries that may be required for certain connectors or tasks. 1. Create a service account named `myserviceaccount`: ```none confluent iam service-account create myserviceaccount --description "test service account" ``` 2. Find the service account ID for `myserviceaccount`: ```none confluent iam service-account list ``` 3. Set a DESCRIBE ACL to the cluster. ```none confluent kafka acl create --allow --service-account "" --operations describe --cluster-scope ``` 4. Set a READ ACL to `pageviews`: ```none confluent kafka acl create --allow --service-account "" --operations read --topic pageviews ``` 5. Set a CREATE ACL to the following topic prefix: ```none confluent kafka acl create --allow --service-account "" --operations create --prefix --topic "dlq-lcc-" ``` 6. Set a WRITE ACL to the following topic prefix: ```none confluent kafka acl create --allow --service-account "" --operations write --prefix --topic "dlq-lcc-" ``` 7. Set a READ ACL to a consumer group with the following prefix: ```none confluent kafka acl create --allow --service-account "" --operations read --prefix --consumer-group "connect-lcc-" ``` 8. Create a Kafka API key and secret for ``: ```none confluent api-key create --resource "lkc-abcd123" --service-account "" ``` 9. Save the API key and secret. The connector configuration must include either an API key and secret or a service account ID. For additional service account information, see [Service Accounts on Confluent Cloud](../security/authenticate/workload-identities/service-accounts/overview.md#service-accounts). ### Source connector service account This example assumes the following: * You have a Kafka cluster with cluster ID `lkc-abcd123`. * You want the source connector to write to a topic named `passengers`. Use the following example steps to create a service account, set ACLs, and add the API key and secret. #### NOTE The following steps show basic ACL entries for source connector service accounts. Make sure to review [Debezium [Legacy] Source Connectors](#cloud-service-account-debezium-acls) and [JDBC-based Source Connectors and the MongoDB Atlas Source Connector](#cloud-service-account-jdbc-mongo-acls) for additional ACL entries that may be required for certain connectors. 1. Create a service account named `myserviceaccount`: ```none confluent iam service-account create myserviceaccount --description "test service account" ``` 2. Find the service account ID for `myserviceaccount`: ```none confluent iam service-account list ``` 3. Set a DESCRIBE ACL to the cluster. ```none confluent kafka acl create --allow --service-account "" --operations describe --cluster-scope ``` 4. Set a WRITE ACL to `passengers`: ```none confluent kafka acl create --allow --service-account "" --operations write --topic "passengers" ``` 5. Create a Kafka API key and secret for ``: ```none confluent api-key create --resource "lkc-abcd123" --service-account "" ``` 6. Save the API key and secret. The connector configuration must include either an API key and secret or a service account ID. For additional service account information, see [Service Accounts on Confluent Cloud](../security/authenticate/workload-identities/service-accounts/overview.md#service-accounts). ### Additional ACL entries Certain connectors require additional ACL entries. #### Debezium [Legacy] Source Connectors The [Source connector service account](#cloud-service-account-source-connectors) section provides basic ACL entries for source connector service accounts. Debezium [Legacy] Source connectors require additional ACL entries. Add the following ACL entries for Debezium [Legacy] Source connectors: * ACLs to create and write to table related topics prefixed with ``. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "" ``` * ACLs to describe configurations at the cluster scope level. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --cluster-scope --operations describe ``` ```none confluent kafka acl create --allow --service-account "" \ --cluster-scope --operations describe_configs ``` The Debezium [MySQL CDC Source (Debezium) [Legacy]](cc-mysql-source-cdc-debezium.md#cc-mysql-source-cdc-debezium) and the Debezium [Microsoft SQL Source (Debezium) [Legacy]](cc-microsoft-sql-server-source-cdc-debezium.md#cc-microsoft-sql-server-source-cdc-debezium) connectors require the following additional ACL entries: * ACLs to read, create, and write to database history topics prefixed with `dbhistory..lcc-`. For example, the server name is `cdc` in the configuration property `"database.server.name": "cdc"`. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations read --prefix --topic "dbhistory..lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "dbhistory..lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "dbhistory..lcc-" ``` * ACLs to read database history consumer group named `-dbhistory`. For example, the server name is `cdc` in the configuration property `"database.server.name": "cdc"`. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations "read" --consumer-group "-dbhistory" ``` If you configure the connector to generate heartbeat events, it periodically produces produces dummy records to the heartbeat topic `__debezium-heartbeat.` at intervals specified by `heartbeat.interval.ms`. To ensure the connector can publish messages to the heartbeat topic, grant ACL permissions for `describe`, `create`, and `write` on topics prefixed with `__debezium-heartbeat.`. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations describe --prefix --topic "__debezium-heartbeat." ``` ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "__debezium-heartbeat." ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "__debezium-heartbeat." ``` #### Debezium V2 Source Connectors The [Source connector service account](#cloud-service-account-source-connectors) section provides basic ACL entries for source connector service accounts. Debezium V2 Source connectors require additional ACL entries. Add the following ACL entries for Debezium V2 Source connectors: * ACLs to create and write to table-related topics prefixed with ``. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "" ``` * ACLs to describe configurations at the cluster scope. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --cluster-scope --operations describe ``` ```none confluent kafka acl create --allow --service-account "" \ --cluster-scope --operations describe_configs ``` The Debezium [MySQL CDC Source V2 (Debezium)](cc-mysql-source-cdc-v2-debezium/index.md#cc-mysql-source-cdc-v2-debezium), Debezium [Microsoft SQL Source V2 (Debezium)](cc-microsoft-sql-server-source-cdc-v2-debezium/index.md#cc-microsoft-sql-server-source-cdc-v2-debezium), and the [MariaDB CDC Source (Debezium)](cc-mariadb-cdc-source-debezium.md#cc-mariadb-source-cdc-debezium-configure-connector) connectors require the following additional ACL entries: * ACLs to read, create, and write to schema history topics prefixed with `dbhistory..lcc-`. For example, the prefix value is `cdc` in the configuration property `"topic.prefix": "cdc"`. Use the following commands to set these ACLs: #### NOTE These steps are applicable only when no custom value is specified for the configuration `Database schema history topic name` in Confluent Cloud Console or `schema.history.internal.kafka.topic` in the Confluent CLI. In case a value is specified, replace the `--topic` parameter with the specified value. ```none confluent kafka acl create --allow --service-account "" \ --operations read --prefix --topic "dbhistory..lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "dbhistory..lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "dbhistory..lcc-" ``` * ACLs to read schema history consumer group named `-schemahistory`. For example, the prefix value is `cdc` in the configuration property `"topic.prefix": "cdc"`. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations "read" --consumer-group "-schemahistory" ``` If you configure the connector to generate heartbeat events, it periodically produces dummy records to the heartbeat topic `__debezium-heartbeat-.` at intervals specified by `heartbeat.interval.ms`. To ensure the connector can publish messages to the heartbeat topic, grant ACL permissions for `describe`, `create`, and `write` on topics prefixed with `__debezium-heartbeat-lcc-`. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations describe --prefix --topic "__debezium-heartbeat-lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "__debezium-heartbeat-lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "__debezium-heartbeat-lcc-" ``` #### JDBC-based Source Connectors and the MongoDB Atlas Source Connector The [Source connector service account](#cloud-service-account-source-connectors) section provides basic ACL entries for source connector service accounts. Several source connectors allow a topic prefix. When a prefix is used and the following connectors are created using the CLI or API, you need to add ACL entries. * [MySQL Source (JDBC) Connector for Confluent Cloud](cc-mysql-source.md#cc-mysql-source) * [PostgreSQL Source (JDBC) Connector for Confluent Cloud](cc-postgresql-source.md#cc-postgresql-source) * [Microsoft SQL Server Source (JDBC) Connector for Confluent Cloud](cc-microsoft-sql-server-source.md#cc-microsoft-sql-server-source) * [Oracle Database Source (JDBC) Connector for Confluent Cloud](cc-oracle-db-source.md#cc-oracle-db-source) * [MongoDB Atlas Source Connector for Confluent Cloud](cc-mongo-db-source.md#cc-mongo-db-source) * [Snowflake Source Connector for Confluent Cloud](cc-snowflake-source/cc-snowflake-source.md#cc-snowflake-source) Add the following ACL entries for these source connectors: ```none confluent kafka acl create --allow --service-account "" --operations create --prefix --topic "" ``` ```none confluent kafka acl create --allow --service-account "" --operations write --prefix --topic "" ``` #### Oracle CDC Source connector To access redo log topics, you must grant the connector a corresponding operation–that is, CREATE, READ, or WRITE in an ACL. The default redo log topic for the Oracle CDC Source connector is `${connectorName}-${databaseName}-redo-log`. When this topic is created by the connector, it appends the `lcc-` prefix. Add the following ACL entries for Redo Log topic access: ```none confluent kafka acl create --allow --service-account "" --operations create --prefix --topic "lcc-" ``` ```none confluent kafka acl create --allow --service-account "" --operations read --prefix --topic "lcc-" ``` ```none confluent kafka acl create --allow --service-account "" --operations write --prefix --topic "lcc-" ``` Add the following ACL entry for Consumer Group READ access: ```none confluent kafka acl create --allow --service-account "" --operations read --prefix --consumer-group "lcc-" ``` If you set the following [configuration properties](cc-oracle-cdc-source/cc-oracle-cdc-source.md#cc-oracle-cdc-source-config-properties), you need to set ACLs for the resulting output topics: * `table.topic.name.template` for table-specific topics. * `lob.topic.name.template` for LOB objects. * `redo.log-corruption.topic` for corrupted redo log records. For these output topics, you must grant the connector either CREATE or WRITE. When granted READ, WRITE, or DELETE, the connector implicitly derives the DESCRIBE operation. #### Oracle XStream CDC Source connector The [Source connector service account](#cloud-service-account-source-connectors) section provides basic ACL entries for source connector service accounts. Oracle XStream CDC Source connector require additional ACL entries. Add the following ACL entries for Oracle XStream CDC Source connector: * ACLs to create and write to change event topics prefixed with ``. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "" ``` * ACLs to describe configurations at the cluster scope level. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --cluster-scope --operations describe ``` ```none confluent kafka acl create --allow --service-account "" \ --cluster-scope --operations describe_configs ``` * ACLs to read, create, and write to schema history topics prefixed with `__orcl-schema-changes..lcc-`. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations read --prefix --topic "__orcl-schema-changes..lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "__orcl-schema-changes..lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "__orcl-schema-changes..lcc-" ``` The following additional ACL entries are required if heartbeats are enabled for the connector using the `heartbeat.interval.ms` configuration property. * ACLs to read, create, and write to heartbeat topics prefixed with `__orcl-heartbeat.lcc-`. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations read --prefix --topic "__orcl-heartbeat.lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "__orcl-heartbeat.lcc-" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "__orcl-heartbeat.lcc-" ``` The following additional ACL entries are required if signaling using Kafka topic is enabled and configured for the connector using the `signal.enabled.channels` and `signal.kafka.topic` configuration properties. * ACLs to read from the signaling topic. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations read --topic "" ``` * ACLs to read Kafka signaling consumer group named `kafka-signal`. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" \ --operations "read" --consumer-group "kafka-signal" ``` #### Azure Cosmos DB Source V2 Connector #### NOTE It is recommended to download the API key and secret for this connector instead of using a service account, as the source connector creates an internal topic, and using a service account will cause an error. The [Source connector service account](#cloud-service-account-source-connectors) section provides basic ACL entries for source connector service accounts. Azure Cosmos DB Source V2 connectors require additional ACL entries. Add the following ACL entries for Azure Cosmos DB Source V2 connectors: - ACLs to read, create and write to table-related topics prefixed with ``. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" --operations read --prefix --topic "" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations create --prefix --topic "" ``` ```none confluent kafka acl create --allow --service-account "" \ --operations write --prefix --topic "" ``` - ACLs to read, create, and write to metadata storage prefixed with `cosmos.metadata.topic/`. Use the following commands to set these ACLs: ```none confluent kafka acl create --allow --service-account "" --operations create --prefix --topic "cosmos.metadata.topic/" ``` ```none confluent kafka acl create --allow --service-account "" --operations read --prefix --topic "cosmos.metadata.topic/" ``` ```none confluent kafka acl create --allow --service-account "" --operations write --prefix --topic "cosmos.metadata.topic/" ``` #### Sink connector SUCCESS and ERROR topics The [Sink connector service account](#cloud-service-account-sink-connectors) section provides basic ACL entries for sink connector service accounts. Several sink connectors create additional `success-lcc` and `error-lcc` topics when the connector is launched. The following sink connectors create these topics and require additional ACL entries: * [AWS Lambda Sink Connector for Confluent Cloud](cc-aws-lambda-sink.md#cc-aws-lambda-sink) * [Azure Functions Sink Connector for Confluent Cloud](cc-azure-functions-sink.md#cc-azure-functions-sink) * [Datadog Metrics Sink for Confluent Cloud](cc-datadog-metrics-sink.md#cc-datadog-metrics-sink) * [Google Cloud Functions Sink Connector [End of Life] for Confluent Cloud](cc-google-functions-sink.md#cc-google-functions-sink) * [HTTP Sink Connector for Confluent Cloud](cc-http-sink.md#cc-http-sink) * [Salesforce Platform Event Sink Connector for Confluent Cloud](cc-salesforce-platform-event-sink.md#cc-salesforce-platform-event-sink) Add the following ACL entries for these sink connectors: ```none confluent kafka acl create --allow --service-account "" --operations create --prefix --topic "success-lcc" ``` ```none confluent kafka acl create --allow --service-account "" --operations write --prefix --topic "success-lcc" ``` ```none confluent kafka acl create --allow --service-account "" --operations create --prefix --topic "error-lcc" ``` ```none confluent kafka acl create --allow --service-account "" --operations write --prefix --topic "error-lcc" ``` #### Sink connector offset management The [Sink connector service account](#cloud-service-account-sink-connectors) section provides basic ACL entries for sink connector service accounts. Sink connectors require additional permissions to manage offsets. You must assign READ, DESCRIBE, and DELETE permissions on the consumer group for the sink connector. Add the following role binding to configure sink connectors: ```none confluent iam rbac role-binding create --principal User:sa-lq5v76 --role ResourceOwner --resource Group:connect-lcc-xyz --kafka-cluster lkc-defg123 --environment env-dvr9z --cloud-cluster lkc-defg123 ``` #### Exactly once semantics for source connectors The [Source connector service account](#cloud-service-account-source-connectors) section provides basic ACL entries for source connector service accounts to use exactly once semantics (EOS). Source connectors enabling EOS automatically create resources with specific naming conventions: - **Offset Topic:** Used for storing connector offsets. * **Default Name:** `connect-offsets-lcc_id` * **Custom Name:** Can be user-provided. - **Transaction Resources:** Used for transactional guarantees. * **Default Name:** `transaction-lcc_id-task_id` * **Custom Name:** Based on the user-provided offset topic name, followed by `-lcc_id-task_id`. Add the following ACL entries with supported source connectors to provide necessary permissions for EOS: - ACLs to `DESCRIBE`, `CREATE`, `READ`, `WRITE` offset topic prefixed with ``: ```bash confluent kafka acl create --allow \ --operations describe,create,read,write \ --service-account \ --topic \ --cluster ``` - ACLs to `DESCRIBE`, `WRITE` transaction resources: ```bash confluent kafka acl create --allow \ --operations describe,write \ --service-account \ --transactional-id --prefix \ --cluster ```