<a id="cloud-streams-config"></a>

# Connect Self-Managed Kafka Streams to Confluent Cloud

Connect a self-managed Kafka Streams application to an Apache Kafka® cluster in
Confluent Cloud by getting the cluster’s bootstrap endpoint with the
Confluent CLI, creating an API key for authentication, and setting the
`bootstrap.servers`, SASL, and security properties in your application
configuration. The steps below walk through this process end to end.

#### NOTE
Kafka Streams applications aren’t supported on Freight clusters.
Freight clusters don’t support the idempotent producer or
transactions. Kafka Streams enables the idempotent producer by default, and it
requires transactions when you set `processing.guarantee` to
`exactly_once_v2`. For more information, see
[Freight cluster considerations](../clusters/cluster-types.md#freight-limitations).

**Prerequisites**

- [Confluent Platform](/platform/current/installation/index.html)

1. Use the Confluent CLI to log in to your Confluent Cloud cluster, and run the
   `confluent kafka cluster list` command to get the Kafka cluster ID.
   ```bash
   confluent kafka cluster list
   ```

   Your output should resemble:
   ```none
     Current |     ID     |    Name    | Type  | Cloud    |  Region  | Availability | Status
   ----------+------------+------------+-------+----------+----------+--------------+---------
     *       | lkc-a123b  | my-cluster | BASIC | azure    | westus2  | single-zone  | UP
   ```
2. Run the `confluent kafka cluster describe` command to get the endpoint for
   your Confluent Cloud cluster.
   ```bash
   confluent kafka cluster describe lkc-a123b
   ```

   Your output should resemble:
   ```text
    +----------------------+---------------------------------------------------------+
    | Current              | true                                                    |
    | ID                   | lkc-a123b                                               |
    | Name                 | wikiedits_cluster                                       |
    | Type                 | BASIC                                                   |
    | Ingress Limit (MB/s) |                                                     250 |
    | Egress Limit (MB/s)  |                                                     750 |
    | Storage              | 5 TB                                                    |
    | Cloud                | azure                                                   |
    | Region               | westus2                                                 |
    | Availability         | single-zone                                             |
    | Status               | UP                                                      |
    | Endpoint             | SASL_SSL://pkc-41973.westus2.azure.confluent.cloud:9092 |
    | REST Endpoint        | https://pkc-41973.westus2.azure.confluent.cloud:443     |
    | Topic Count          |                                                      30 |
    +----------------------+---------------------------------------------------------+
   ```

   Save the `Endpoint` value, which you’ll use in a later step.
3. Create a service account named `my-streams-app`. You must include a
   description.
   ```bash
   confluent iam service-account create my-streams-app --description "My Streams API and secrets service account."
   ```

   Your output should resemble:
   ```text
   +-------------+--------------------------------+
   | ID          | sa-ab01cd                      |
   | Name        | my-streams-app                 |
   | Description | My Streams API and secrets     |
   |             | service account.               |
   +-------------+--------------------------------+
   ```

   Save the service account ID, which you’ll use in later steps.
4. Create an API key and secret for service account `sa-ab01cd`. Be
   sure to replace the service account ID and Kafka cluster ID values shown
   here with your own:
   ```bash
   confluent api-key create --service-account sa-ab01cd --resource lkc-a123b
   ```

   Your output should resemble:
   ```text
   It may take a couple of minutes for the API key to be ready.
   Save the API key and secret. The secret is not retrievable later.
   +---------+------------------------------------------------------------------+
   | API Key | ABCXQHYDZXMMUDEF                                                 |
   | Secret  | aBCde3s54+4Xv36YKPLDKy2aklGr6x/ShUrEX5D1Te4AzRlphFlr6eghmPX81HTF |
   +---------+------------------------------------------------------------------+
   ```

   #### IMPORTANT
   **Save the API key and secret.** You require this information to
   configure your client applications. Be aware that this is the *only*
   time that you can access and view the key and secret.

To connect Kafka Streams to Confluent Cloud, update your existing [existing Stream](/platform/current/streams/developer-guide/config-streams.html)
configurations with the properties described here.

1. Create a `java.util.Properties` instance.
2. Configure your Kafka Streams application. Set the Kafka and Kafka Streams
   configuration options in the `java.util.Properties` instance before
   you start Kafka Streams. In this example, you must configure the Confluent Cloud
   broker endpoints (`StreamsConfig.BOOTSTRAP_SERVERS_CONFIG`) and SASL
   configuration (`SASL_JAAS_CONFIG`).
   ```java
   import java.util.Properties;
   import org.apache.kafka.clients.producer.ProducerConfig;
   import org.apache.kafka.common.config.SaslConfigs;
   import org.apache.kafka.streams.StreamsConfig;

   Properties props = new Properties();
   // Comma-separated list of the Confluent Cloud broker endpoints. For example:
   // r0.great-app.confluent.aws.prod.cloud:9092,r1.great-app.confluent.aws.prod.cloud:9093,r2.great-app.confluent.aws.prod.cloud:9094
   props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "<broker-endpoint1, broker-endpoint2, broker-endpoint3>");
   props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
   props.put(StreamsConfig.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
   props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
   props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required \
   username=\"<api-key>\" password=\"<api-secret>\";");

   // Recommended performance/resilience settings
   props.put(StreamsConfig.producerPrefix(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG), 2147483647);
   props.put(StreamsConfig.producerPrefix(ProducerConfig.MAX_BLOCK_MS_CONFIG), 9223372036854775807);

   // Any further settings
   props.put(... , ...);
   ```
3. (Optional) Add configurations for Confluent Cloud Schema Registry to your Kafka Streams
   application, as shown in
   [java_streams.delta](https://github.com/confluentinc/examples/tree/latest/ccloud/template_delta_configs/java_streams.delta)
   on GitHub at
   [ccloud/examples/template_delta_configs](https://github.com/confluentinc/examples/tree/latest/ccloud/template_delta_configs).
   ```java
   // Confluent Schema Registry for Java
   props.put("basic.auth.credentials.source", "USER_INFO");
   props.put("schema.registry.basic.auth.user.info", "<SCHEMA_REGISTRY_API_KEY>:<SCHEMA_REGISTRY_API_SECRET>");
   props.put("schema.registry.url", "https://<SCHEMA_REGISTRY_ENDPOINT>");
   ```

- For more information, see
  [Configuring a Streams Application](/platform/current/streams/developer-guide/config-streams.html).
- To view a working example of hybrid Apache Kafka® clusters from self-hosted to
  Confluent Cloud, see [cp-demo](/platform/current/tutorials/cp-demo/docs/index.html).
- For example configurations for all Confluent Platform components and clients
  connecting to Confluent Cloud, see
  [template examples for components](https://github.com/confluentinc/examples/tree/latest/ccloud/template_delta_configs).
- To look at all the code used in the Confluent Cloud demo, see the
  [Confluent Cloud demo examples](https://github.com/confluentinc/examples/tree/latest/ccloud).

## Create ACLs for Kafka Streams to access Confluent Cloud

If your Kafka cluster in Confluent Cloud has ACLs enabled, your Kafka Streams application
must be granted access to specific resources on the Kafka cluster. Use the
following Confluent CLI commands to create the necessary ACLs in the Kafka
cluster to allow Kafka Streams to operate on the specified topics.

In the following commands, replace `<service-account-id>` with the ID of the
service account that you created previously.

```bash
# Read access for input topic(s)
kafka-acls --bootstrap-server <ccloud-bootstrap-server> \
  --command-config <cloud.properties> --add --allow-principal User:<service-account-id> \
  --operation read --topic <input-topic>

# Write access for output topic(s)
kafka-acls --bootstrap-server <ccloud-bootstrap-server> \
  --command-config <cloud.properties> --add --allow-principal User:<service-account-id> \
  --operation write --topic <output-topic>

# Cluster access for idempotent write (needed for EOS)
kafka-acls --bootstrap-server <ccloud-bootstrap-server> \
  --command-config <cloud.properties> --add --allow-principal User:<service-account-id> \
  --operation idempotentWrite --cluster

# Full access to internal topics, consumer groups, and transactional ids
# for this application, prefixed by the Streams <application.id>
kafka-acls --bootstrap-server <ccloud-bootstrap-server> \
  --command-config <cloud.properties> --add --allow-principal User:<service-account-id> \
  --operation All --resource-pattern-type prefixed --topic <application.id> \
  --group <application.id> --transactional-id <application.id>
```

## Related content

- [Build Kafka Streams Client Applications for Confluent Cloud](../kafka-streams/overview.md#client-kafka-streams)
- [Kafka Streams Quick Start for Confluent Cloud](../kafka-streams/quickstart.md#client-kafka-streams-quick-start)
- [Monitor Kafka Streams Applications in Confluent Cloud](../kafka-streams/monitor-kafka-streams-apps.md#client-kafka-streams-monitor-apps)
- [Configuring a Streams Application](/platform/current/streams/developer-guide/config-streams.html)

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