# Module 2: Deploy Hybrid Confluent Platform and Confluent Cloud Environment
In a hybrid Apache Kafka® deployment scenario, you can have both an on-premises Confluent Platform deployment as well as a
[Confluent Cloud](https://confluent.cloud) deployment.
In this module, you will use [Cluster Linking](https://docs.confluent.io/cloud/current/multi-cloud/cluster-linking/index.html)
and [Schema Linking](/platform/current/schema-registry/schema-linking-cp.html) to send data and schemas to
Confluent Cloud, and monitor both deployments with [Confluent Health+](/platform/current/health-plus/index.html#confluent-health) and the [Confluent Cloud Metrics API](https://docs.confluent.io/cloud/current/monitoring/metrics-api.html).
Before you begin this module, make sure the cp-demo `start.sh` script successfully completed and Confluent Platform is already running [(see the on-premises module)](on-prem.md#cp-demo-run).
## Cost to Run
### Caution
Any Confluent Cloud example uses real Confluent Cloud resources that may be billable.
An example may create a new Confluent Cloud environment, Kafka cluster, topics, ACLs, and service accounts, as well as resources that have hourly charges like connectors and ksqlDB applications.
To avoid unexpected charges, carefully [evaluate the cost of resources](https://docs.confluent.io/cloud/current/billing/overview.html) before you start.
After you are done running a Confluent Cloud example, destroy all Confluent Cloud resources to avoid accruing hourly charges for services and verify that they have been deleted.
### Confluent Cloud Promo Code
To receive an additional $50 free usage in Confluent Cloud, enter promo code `CPDEMO50` in the Confluent Cloud Console’s Billing and payment section ([details](https://www.confluent.io/confluent-cloud-promo-disclaimer)).
This promo code should sufficiently cover up to one day of running this Confluent Cloud example, beyond which you may be billed for the services that have an hourly charge until you destroy the Confluent Cloud resources created by this example.
## Set Up Confluent Cloud
1. Create a Confluent Cloud account at [https://confluent.cloud](https://confluent.cloud).
2. Enter the promo code `CPDEMO50` in the Confluent Cloud UI Billing and payment section to receive an additional $50 free usage.
3. Go to [https://confluent.cloud/environments](https://confluent.cloud/environments) and click “+ Add cloud environment”. Name the environment **cp-demo-env**.
4. Inside the “cp-demo-env” environment, create a **Dedicated** Confluent Cloud cluster named **cp-demo-cluster** in the cloud provider and region of your choice with default configurations. Wait until your cluster is in a running state before proceeding.
#### NOTE
Cluster Linking requires a dedicated cluster
5. Create a Schema Registry for the “cp-demo-env” environment in the same region as your cluster.
## Set Up Confluent CLI and variables
1. [Install Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html) locally, v3.28.0 or later (if already installed, update the CLI as described in [Upgrade](https://docs.confluent.io/confluent-cli/current/install.html#upgrade)).
Verify the installation was successful.
```none
confluent version
```
2. Using the CLI, log in to Confluent Cloud with the command `confluent login`, and use your Confluent Cloud username and password. The `--save` argument saves your Confluent Cloud user login credentials for future use.
```shell
confluent login --save
```
3. Use the demo Confluent Cloud environment.
```shell
CC_ENV=$(confluent environment list -o json \
| jq -r '.[] | select(.name | contains("cp-demo")) | .id') \
&& echo "Your Confluent Cloud environment: $CC_ENV" \
&& confluent environment use $CC_ENV
```
4. Get the Confluent Cloud cluster ID and use the cluster.
```shell
CCLOUD_CLUSTER_ID=$(confluent kafka cluster list -o json \
| jq -r '.[] | select(.name | contains("cp-demo")) | .id') \
&& echo "Your Confluent Cloud cluster ID: $CCLOUD_CLUSTER_ID" \
&& confluent kafka cluster use $CCLOUD_CLUSTER_ID
```
5. Get the bootstrap endpoint for the Confluent Cloud cluster.
```shell
CC_BOOTSTRAP_ENDPOINT=$(confluent kafka cluster describe -o json | jq -r .endpoint) \
&& echo "Your Cluster's endpoint: $CC_BOOTSTRAP_ENDPOINT"
```
6. Create a Confluent Cloud service account for CP Demo and get its ID.
```shell
confluent iam service-account create cp-demo-sa --description "service account for cp-demo" \
&& SERVICE_ACCOUNT_ID=$(confluent iam service-account list -o json \
| jq -r '.[] | select(.name | contains("cp-demo")) | .id') \
&& echo "Your cp-demo service account ID: $SERVICE_ACCOUNT_ID"
```
7. Get the ID and endpoint URL for your Schema Registry cluster.
(**Note:** The Schema Registry cluster was created by default when you [added your cloud environment](/cloud/current/get-started/schema-registry.html#cloud-sr-enable-zones).)
```shell
CC_SR_CLUSTER_ID=$(confluent schema-registry cluster describe -o json | jq -r .cluster_id) \
&& CC_SR_ENDPOINT=$(confluent schema-registry cluster describe -o json | jq -r .endpoint_url) \
&& echo "Schema Registry Cluster ID: $CC_SR_CLUSTER_ID" \
&& echo "Schema Registry Endpoint: $CC_SR_ENDPOINT"
```
8. Create a Schema Registry API key for the cp-demo service account.
```shell
confluent api-key create \
--service-account $SERVICE_ACCOUNT_ID \
--resource $CC_SR_CLUSTER_ID \
--description "SR key for cp-demo schema link"
```
Verify your output resembles
```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 | SZBKJLD67XK5NZNZ |
| Secret | NTqs/A3Mt0Ohkk4fkaIsC0oLQ5Q/F0lLowYo/UrsTrEAM5ozxY7fjqxDdVwMJz99 |
+---------+------------------------------------------------------------------+
```
Set variables to reference the Schema Registry credentials returned in the previous step.
```shell
SR_API_KEY=SZBKJLD67XK5NZNZ
SR_API_SECRET=NTqs/A3Mt0Ohkk4fkaIsC0oLQ5Q/F0lLowYo/UrsTrEAM5ozxY7fjqxDdVwMJz99
```
9. Create a Kafka cluster API key for the cp-demo service account.
```shell
confluent api-key create \
--service-account $SERVICE_ACCOUNT_ID \
--resource $CCLOUD_CLUSTER_ID \
--description "Kafka key for cp-demo cluster link"
```
Verify your output resembles
```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 | SZBKLMG61XK9NZAB |
| Secret | QTpi/A3Mt0Ohkk4fkaIsGR3ATQ5Q/F0lLowYo/UrsTr3AMsozxY7fjqxDdVwMJz02 |
+---------+-------------------------------------------------------------------+
```
Set variables to reference the Kafka credentials returned in the previous step.
```shell
CCLOUD_CLUSTER_API_KEY=SZBKLMG61XK9NZAB
CCLOUD_CLUSTER_API_SECRET=QTpi/A3Mt0Ohkk4fkaIsGR3ATQ5Q/F0lLowYo/UrsTr3AMsozxY7fjqxDdVwMJz02
```
10. We will also need the cluster ID for the on-premises Confluent Platform cluster.
```shell
CP_CLUSTER_ID=$(curl -s https://localhost:8091/v1/metadata/id \
--tlsv1.2 --cacert ./scripts/security/snakeoil-ca-1.crt \
| jq -r ".id") \
&& echo "Your on-premises Confluent Platform cluster ID: $CP_CLUSTER_ID"
```
#### NOTE
For security purposes, you may be automatically logged out of the `confluent` CLI at some point. If this happens,
run the following command:
```shell
confluent login && \
confluent environment use $CC_ENV && \
confluent kafka cluster use $CCLOUD_CLUSTER_ID
```
## Export Schemas to Confluent Cloud with Schema Linking
Confluent Schema Registry is critical for evolving schemas alongside your business needs and ensuring high data quality.
With [Schema Linking](/platform/current/schema-registry/schema-linking-cp.html)
, you can easily export your schemas from your on-premises Schema Registry to Confluent Cloud.
In this section, you will export the schema subjects `wikipedia.parsed-value` and `wikipedia.parsed.count-by-domain-value`
from Confluent Platform to Confluent Cloud with schema linking.
These schema subjects will be exported to a new [schema context](/platform/current/schema-registry/schema-linking-cp.html#what-is-a-schema-context)
called “cp-demo”, so their qualified subject names in Confluent Cloud will be `:.cp-demo:wikipedia.parsed-value` and `:.cp-demo:wikipedia.parsed.count-by-domain-value`.
1. From here, we will switch back and forth between using Confluent Cloud
and Confluent Platform. We can streamline this “context switching”
with the `confluent context` CLI subcommand.
Here let’s create a context called “ccloud” from the current context.
```shell
confluent context update --name ccloud
```
2. Next, log into Confluent Platform and create a context called “cp”.
To create a cluster link, the CLI user must have `ClusterAdmin`
privileges. For simplicity, sign in as a super user using the username **superUser**
and password **superUser**
```shell
confluent login --save --url https://localhost:8091 \
--certificate-authority-path scripts/security/snakeoil-ca-1.crt
```
and create a CLI context called “cp”.
```shell
confluent context update --name cp
```
3. Inspect the schema exporter configuration file at `scripts/ccloud/schema-link-example.properties`.
```none
# Copy the contents of this file to schema-link.properties and
# add your Confluent Cloud Schema Registry credentials
# *****DO NOT***** add schema-link.properties to version control
# with your Confluent Cloud credentials
schema.registry.url=
basic.auth.credentials.source=USER_INFO
basic.auth.user.info=:
```
4. Run the following command copy the contents of the configuration file to a new file called `schema-link.properties` that includes your Schema Registry credentials.
```shell
sed -e "s||${CC_SR_ENDPOINT}|g" \
-e "s||${SR_API_KEY}|g" \
-e "s||${SR_API_SECRET}|g" \
scripts/ccloud/schema-link-example.properties > scripts/ccloud/schema-link.properties
```
5. Create a schema exporter called “cp-cc-schema-exporter” for the on-premises Schema Registry.
```shell
confluent schema-registry exporter create cp-cc-schema-exporter \
--subjects "wikipedia.parsed*" \
--context-name cp-demo \
--context-type CUSTOM \
--schema-registry-endpoint https://localhost:8085 \
--certificate-authority-path scripts/security/snakeoil-ca-1.crt \
--config scripts/ccloud/schema-link.properties
```
Notice we can use a wildcard `*` to export multiple subjects.
#### NOTE
Whether using the REST API or the CLI, the user
making the request needs permission
to create the schema exporter and to read the schema subjects.
For educational purposes, here is an equivalent command that uses `curl` on Confluent Server’s
embedded REST API with the `schemaregistryUser` principal:
```shell
curl -X POST -H "Content-Type: application/json" \
-d @<(cat <<-EOF
{
"name": "cp-cc-schema-exporter",
"contextType": "CUSTOM",
"context": "cp-demo",
"subjects": ["wikipedia.parsed*"],
"config": {
"schema.registry.url": "${CC_SR_ENDPOINT}",
"basic.auth.credentials.source": "USER_INFO",
"basic.auth.user.info": "${SR_API_KEY}:${SR_API_SECRET}"
}
}
EOF
) \
--user schemaregistryUser:schemaregistryUser \
--cacert scripts/security/snakeoil-ca-1.crt \
https://localhost:8085/exporters
```
6. Verify the schema exporter is running.
```shell
confluent schema-registry exporter status describe cp-cc-schema-exporter \
--schema-registry-endpoint https://localhost:8085 \
--certificate-authority-path scripts/security/snakeoil-ca-1.crt
```
7. Switch back to the `ccloud` CLI context (not to be confused with Schema Registry context!).
```none
confluent context use ccloud
```
8. Verify that the schema subjects are being exported to Confluent Cloud.
```shell
confluent schema-registry subject list --prefix ":.cp-demo:"
```
The output should resemble
```text
Subject
----------------------------------------------------
:.cp-demo:wikipedia.parsed-value
:.cp-demo:wikipedia.parsed.count-by-domain-value
```
Schema subjects have been successfully exported from Confluent Platform to
Confluent Cloud with schema linking! As schemas evolve on-premises, those changes will
automatically propagate to Confluent Cloud as long as the exporter is running.
## Mirror Data to Confluent Cloud with Cluster Linking
In this section, you will create a source-initiated cluster link
to mirror the topic `wikipedia.parsed` from Confluent Platform to Confluent Cloud.
For security reasons, most on-premises datacenters don’t
allow inbound connections,
so Confluent recommends source-initiated cluster linking to easily and securely
mirror Kafka topics from your on-premises cluster to Confluent Cloud.
1. Verify that you’re still using the `ccloud` CLI context.
```none
confluent context list
```
2. Give the cp-demo service account the `CloudClusterAdmin` role in Confluent Cloud
to authorize it to create cluster links and mirror topics in Confluent Cloud.
```shell
confluent iam rbac role-binding create \
--principal User:$SERVICE_ACCOUNT_ID \
--role CloudClusterAdmin \
--cloud-cluster $CCLOUD_CLUSTER_ID --environment $CC_ENV
```
Verify that the role-binding was created. The output should show the role has been created.
```shell
confluent iam rbac role-binding list \
--principal User:$SERVICE_ACCOUNT_ID \
-o json | jq
```
3. Inspect the file `scripts/ccloud/cluster-link-ccloud.properties`
```none
# This is the Confluent Cloud half of the cluster link
# Confluent Cloud dedicated cluster is the destination
link.mode=DESTINATION
# Link connection comes in from Confluent Platform so you don't have to open your on-prem firewall
connection.mode=INBOUND
```
4. Create the Confluent Cloud half of the cluster link with the name **cp-cc-cluster-link**.
```shell
confluent kafka link create cp-cc-cluster-link \
--cluster $CCLOUD_CLUSTER_ID \
--source-cluster $CP_CLUSTER_ID \
--config-file ./scripts/ccloud/cluster-link-ccloud.properties
```
5. Inspect the file `scripts/ccloud/cluster-link-cp-example.properties` and read the comments to understand what each property does.
```none
# Configuration for the Confluent Platform half of the cluster link
# Copy the contents of this file to cluster-link-cp.properties and
# add your Confluent Cloud credentials
# *****DO NOT***** add cluster-link-cp.properties to version control
# with your Confluent Cloud credentials
# Confluent Platform is the source cluster
link.mode=SOURCE
# The link is initiated at the source so you don't have to open your firewall
connection.mode=OUTBOUND
# Authenticate to Confluent Cloud
bootstrap.servers=
ssl.endpoint.identification.algorithm=https
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username='' \
password='';
# We are using the CP's SASL OAUTHBEARER token listener
local.listener.name=TOKEN
local.sasl.mechanism=OAUTHBEARER
local.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
local.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
username="connectorSA" \
password="connectorSA" \
metadataServerUrls="https://kafka1:8091,https://kafka2:8092";
```
6. Run the following command to copy the file to `scripts/ccloud/cluster-link-cp.properties`
with credentials and bootstrap endpoint for your own Confluent Cloud cluster.
```shell
sed -e "s||${CCLOUD_CLUSTER_API_KEY}|g" \
-e "s||${CCLOUD_CLUSTER_API_SECRET}|g" \
-e "s||${CC_BOOTSTRAP_ENDPOINT}|g" \
scripts/ccloud/cluster-link-cp-example.properties > scripts/ccloud/cluster-link-cp.properties
```
7. Next, use the `cp` CLI context to log into Confluent Platform.
To create a cluster link, the CLI user must have `ClusterAdmin`
privileges. For simplicity, we are continuing to use a super user instead of a `ClusterAdmin`.
```shell
confluent context use cp
```
8. The cluster link itself needs the `DeveloperRead` and `DeveloperManage`
roles for any topics it plans to mirror, as well as the `ClusterAdmin` role for the Kafka cluster.
Our cluster link uses the `connectorSA` principal, which already has
`ResourceOwner` permissions on the `wikipedia.parsed` topic, so we just
need to add the `ClusterAdmin` role.
```shell
confluent iam rbac role-binding create \
--principal User:connectorSA \
--role ClusterAdmin \
--kafka-cluster $CP_CLUSTER_ID
```
9. Create the Confluent Platform half of the cluster link, still called **cp-cc-cluster-link**.
```shell
confluent kafka link create cp-cc-cluster-link \
--destination-bootstrap-server $CC_BOOTSTRAP_ENDPOINT \
--destination-cluster $CCLOUD_CLUSTER_ID \
--config ./scripts/ccloud/cluster-link-cp.properties \
--url https://localhost:8091/kafka \
--certificate-authority-path scripts/security/snakeoil-ca-1.crt
```
10. Switch contexts back to “ccloud” and create the mirror topic for `wikipedia.parsed` in Confluent Cloud.
```shell
confluent context use ccloud \
&& confluent kafka mirror create wikipedia.parsed --link cp-cc-cluster-link
```
11. Consume records from the mirror topic using the schema context “cp-demo”.
Press `Ctrl+C` to stop the consumer when you are ready.
```shell
confluent kafka topic consume \
--api-key $CCLOUD_CLUSTER_API_KEY \
--api-secret $CCLOUD_CLUSTER_API_SECRET \
--schema-registry-endpoint $CC_SR_ENDPOINT/contexts/:.cp-demo: \
--schema-registry-api-key $SR_API_KEY \
--schema-registry-api-secret $SR_API_SECRET \
--value-format avro \
wikipedia.parsed | jq
```
You successfully created a source-initiated cluster link to seamlessly
move data from on-premises to cloud in real time. Cluster linking opens up
real-time hybrid cloud, multi-cloud, and disaster recovery use cases.
See the [Cluster Linking documentation](https://docs.confluent.io/cloud/current/multi-cloud/overview.html)
for more information.
## Confluent Cloud ksqlDB
In this section, you will create a Confluent Cloud ksqlDB cluster to processes data from the `wikipedia.parsed` mirror topic.
1. Log into the Confluent Cloud Console at [https://confluent.cloud](https://confluent.cloud) and navigate to the **cp-demo-env** environment and then to the **cp-demo-cluster** cluster within that environment.
2. Select “ksqlDB” from the left side menu, click “Create cluster myself”. Select “Global access”. Name the cluster **cp-demo-ksql** and choose a cluster size of 1 CKU. It will take a minute or so to provision.
3. Once the ksqlDB cluster is provisioned, click into it and enter these query statements into the editor:
```sql
CREATE STREAM wikipedia WITH (kafka_topic='wikipedia.parsed', value_format='AVRO');
CREATE STREAM wikipedianobot AS
SELECT *, (length->new - length->old) AS BYTECHANGE
FROM wikipedia
WHERE bot = false
AND length IS NOT NULL
AND length->new IS NOT NULL
AND length->old IS NOT NULL;
```
4. Click the “Flow” tab to see the stream processing topology.
5. View the events in the ksqlDB streams in Confluent Cloud by pasting in `SELECT * FROM WIKIPEDIANOBOT EMIT CHANGES;` and clicking “Run query”. Stop the query when you are finished.
#### IMPORTANT
The ksqlDB cluster in Confluent Cloud has hourly charges even if you are not actively using it. Make sure to go to [Stop the Confluent Cloud environment](teardown.md#cp-demo-ccloud-cleanup)
in the Teardown module to destroy all cloud resources when you are finished.
## Metrics API
The Confluent Cloud Metrics API is a REST API you can use to query timeseries metrics.
You can use the Metrics API to get telemetry data for both the on-premises Confluent Platform cluster as well as the Confluent Cloud cluster.
- On-premises metrics (enabled by Telemetry Reporter) using the endpoint [https://api.telemetry.confluent.cloud/v2/metrics/hosted-monitoring/query](https://api.telemetry.confluent.cloud/v2/metrics/hosted-monitoring/query)
#### NOTE
The hosted monitoring endpoint is in preview and the endpont will eventually be renamed [https://api.telemetry.confluent.cloud/v2/metrics/health-plus/query](https://api.telemetry.confluent.cloud/v2/metrics/health-plus/query)
- Confluent Cloud metrics using the endpoint [https://api.telemetry.confluent.cloud/v2/metrics/cloud/query](https://api.telemetry.confluent.cloud/v2/metrics/cloud/query)
- See the [Confluent Cloud Metrics API Reference](https://api.telemetry.confluent.cloud/docs) for more information.
The Metrics API and Telemetry Reporter powers [Health+](/platform/current/health-plus/index.html), the fully-managed monitoring
solution for Confluent Platform. You can enable Health+ for free and add premium capabilities as you see fit.
Popular [third-party monitoring tools](https://docs.confluent.io/cloud/current/monitoring/metrics-api.html#integrate-with-third-party-monitoring)
like Datadog and Grafana Cloud integrate with the Metrics API out-of-the-box,
or if you manage your own Prometheus database, the Metrics API can also export metrics in Prometheus format.
### Configure Confluent Health+ with the Telemetry Reporter
1. Verify that you’re still using the `ccloud` CLI context.
```none
confluent context list
```
2. Create a new `Cloud` API key and secret to authenticate to Confluent Cloud. These credentials will be used to configure the Telemetry Reporter in Confluent Platform for Health+, as well as to access the Confluent Cloud Metrics API directly.
```shell
confluent api-key create --resource cloud -o json \
--service-account $SERVICE_ACCOUNT_ID \
--description "cloud api key for cp-demo"
```
Verify your output resembles:
```text
{
"key": "QX7X4VA4DFJTTOIA",
"secret": "fjcDDyr0Nm84zZr77ku/AQqCKQOOmb35Ql68HQnb60VuU+xLKiu/n2UNQ0WYXp/D"
}
```
The value of the API key, in this case `QX7X4VA4DFJTTOIA`, and API secret,
in this case `fjcDDyr0Nm84zZr77ku/AQqCKQOOmb35Ql68HQnb60VuU+xLKiu/n2UNQ0WYXp/D`,
will differ in your output.
3. Set variables to reference these credentials returned in the previous step.
```text
METRICS_API_KEY=QX7X4VA4DFJTTOIA
METRICS_API_SECRET=fjcDDyr0Nm84zZr77ku/AQqCKQOOmb35Ql68HQnb60VuU+xLKiu/n2UNQ0WYXp/D
```
4. [Dynamically configure](../../kafka/dynamic-config.md#kafka-dynamic-configurations) the on-premises `cp-demo` cluster to use the Telemetry Reporter, which sends metrics to Confluent Cloud. This requires setting 3 configuration parameters: `confluent.telemetry.enabled=true`, `confluent.telemetry.api.key`, and `confluent.telemetry.api.secret`.
```text
docker compose exec kafka1 kafka-configs \
--bootstrap-server kafka1:12091 \
--alter \
--entity-type brokers \
--entity-default \
--add-config confluent.telemetry.enabled=true,confluent.telemetry.api.key=${METRICS_API_KEY},confluent.telemetry.api.secret=${METRICS_API_SECRET}
```
5. Check the broker logs to verify the brokers were dynamically configured.
```bash
docker logs --since=5m kafka1 | grep confluent.telemetry.api
```
Your output should resemble the following, but the `confluent.telemetry.api.key` value will be different in your environment.
```text
...
confluent.telemetry.api.key = QX7X4VA4DFJTTOIA
confluent.telemetry.api.secret = [hidden]
...
```
6. Navigate to the Health+ section of the Confluent Cloud Console at [https://confluent.cloud/health-plus](https://confluent.cloud/health-plus) and verify you see your cluster’s Health+ dashboard.
### Query Metrics
1. First we will query the Metrics API for on-premises metrics. Here are the content of the query file [metrics query file](https://github.com/confluentinc/cp-demo/tree/latest/scripts/ccloud/metrics_query_onprem.json), which requests `io.confluent.kafka.server/received_bytes` for the topic `wikipedia.parsed` in the on-premises cluster (for all queryable metrics examples, see [Metrics API](https://docs.confluent.io/cloud/current/monitoring/metrics-api.html)):
```none
{
"aggregations": [
{
"agg": "SUM",
"metric": "io.confluent.kafka.server/received_bytes"
}
],
"filter": {
"filters": [
{
"field": "metric.topic",
"op": "EQ",
"value": "wikipedia.parsed"
}
],
"op": "AND"
},
"intervals": ["now-1h/now+1h"],
"granularity": "PT1M",
"group_by": [
"metric.topic"
],
"limit": 5
}
```
2. Send this query to the Metrics API endpoint at [https://api.telemetry.confluent.cloud/v2/metrics/hosted-monitoring/query](https://api.telemetry.confluent.cloud/v2/metrics/hosted-monitoring/query).
```text
curl -s -u ${METRICS_API_KEY}:${METRICS_API_SECRET} \
--header 'content-type: application/json' \
--data @scripts/ccloud/metrics_query_onprem.json \
https://api.telemetry.confluent.cloud/v2/metrics/hosted-monitoring/query \
| jq .
```
3. Your output should resemble the output below, showing metrics for the on-premises topic `wikipedia.parsed`:
```text
{
"data": [
{
"timestamp": "2020-12-14T20:52:00Z",
"value": 1744066,
"metric.topic": "wikipedia.parsed"
},
{
"timestamp": "2020-12-14T20:53:00Z",
"value": 1847596,
"metric.topic": "wikipedia.parsed"
}
]
}
```
4. For the Confluent Cloud metrics: view the [metrics query file](https://github.com/confluentinc/cp-demo/tree/latest/scripts/ccloud/metrics_query_ccloud.json), which requests `io.confluent.kafka.server/cluster_link_mirror_topic_bytes` for the cluster link `cp-cc-cluster-link` in Confluent Cloud, which includes metrics for the `wikipedia.parsed` mirror topic.
```none
{
"aggregations": [
{
"metric": "io.confluent.kafka.server/cluster_link_mirror_topic_bytes"
}
],
"filter": {
"filters": [
{
"field": "metric.link_name",
"op": "EQ",
"value": "cp-cc-cluster-link"
},
{
"field": "resource.kafka.id",
"op": "EQ",
"value": ""
}
],
"op": "AND"
},
"intervals": ["now-1h/now+1h"],
"granularity": "PT1H",
"group_by": [
"metric.topic"
],
"limit": 5
}
```
5. Send this query to the Metrics API endpoint at [https://api.telemetry.confluent.cloud/v2/metrics/cloud/query](https://api.telemetry.confluent.cloud/v2/metrics/cloud/query).
```text
sed "s//${CCLOUD_CLUSTER_ID}/g" scripts/ccloud/metrics_query_ccloud.json \
| curl -s -u ${METRICS_API_KEY}:${METRICS_API_SECRET} \
--header 'content-type: application/json' \
--data @- \
https://api.telemetry.confluent.cloud/v2/metrics/cloud/query \
| jq .
```
6. Your output should resemble the output below, showing metrics for the cluster link `cp-cc-cluster-link`, including the Confluent Cloud mirror topic `wikipedia.parsed`:
```text
{
"data": [
{
"timestamp": "2020-12-14T20:00:00Z",
"value": 1690522,
"metric.topic": "wikipedia.parsed"
}
]
}
```
## Cleanup
Any Confluent Cloud example uses real Confluent Cloud resources.
After you are done running a Confluent Cloud example, manually verify that all Confluent Cloud resources are destroyed to avoid unexpected charges.
Follow the clean up procedure in [Stop the Confluent Cloud environment](teardown.md#cp-demo-ccloud-cleanup) to avoid unexpected Confluent Cloud charges.
## Related content
- [Scripted Confluent Platform Demo](overview.md#scripted-demo)
- [Module 2: Deploy Hybrid Confluent Platform and Confluent Cloud Environment](#cp-demo-hybrid)