Run Example Queries for the Confluent Cloud Metrics API

The Confluent Cloud Metrics API has an expressive query language that lets you flexibly filter and group time-series data. The following example queries serve as a template.

Find more examples in the Cloud Console, which also uses the Confluent Cloud Metrics API.

To learn how to identify clients sending deprecated requests to a cluster, see Client Deprecation in Confluent Cloud.

Timestamps in metrics queries use Coordinated Universal Time (UTC). Use either UTC or an offset appropriate for your location.

In cases where a principal_id is returned, the response also contains a principal_name field with the human-readable name for the principal_id. You should use the principal_name field for display purposes, and the principal_id field for any programmatic use cases such as alerting or monitoring. For an example, see Query for metrics for a specific principal ID.

Monitor Kafka clusters and clients

The following examples query metrics for Kafka clusters, topics, and clients.

Query for bytes produced to the cluster per minute grouped by topic

This query measures bytes produced (ingress). If you want to query bytes consumed (egress), see Query for bytes consumed from the cluster per minute grouped by topic. If you are using Cluster Linking, the received_bytes does not include the mirror-in bytes to the cluster. You can use the cluster_link_destination_response_bytes metrics to query the mirror-in bytes instead.

1. Create a file named received_bytes_query.json using the following template. Be sure to change lkc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.server/received_bytes"
       }
     ],
     "filter": {
       "field": "resource.kafka.id",
       "op": "EQ",
       "value": "lkc-XXXXX"
     },
     "granularity": "PT1M",
     "group_by": [
       "metric.topic"
     ],
     "intervals": [
       "2019-12-19T11:00:00-05:00/2019-12-19T11:05:00-05:00"
     ],
     "limit": 25
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < received_bytes_query.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @received_bytes_query.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2019-12-19T16:00:00Z",
         "metric.topic": "test-topic",
         "value": 72.0
       },
       {
         "timestamp": "2019-12-19T16:01:00Z",
         "metric.topic": "test-topic",
         "value": 139.0
       },
       {
         "timestamp": "2019-12-19T16:02:00Z",
         "metric.topic": "test-topic",
         "value": 232.0
       },
       {
         "timestamp": "2019-12-19T16:03:00Z",
         "metric.topic": "test-topic",
         "value": 0.0
       },
       {
         "timestamp": "2019-12-19T16:04:00Z",
         "metric.topic": "test-topic",
         "value": 0.0
       }
     ]
   }
   

Query for bytes consumed from the cluster per minute grouped by topic

This query measures bytes consumed (egress). If you want to query bytes produced (ingress), see Query for bytes produced to the cluster per minute grouped by topic. If you are using Cluster Linking, the sent_bytes metric also includes the mirror-out bytes from the cluster. For details about sent_bytes and received_bytes with Cluster Linking, see Cluster Linking Performance Limits.

1. Create a file named sent_bytes_query.json using the following template. Be sure to change lkc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.server/sent_bytes"
       }
     ],
     "filter": {
       "field": "resource.kafka.id",
       "op": "EQ",
       "value": "lkc-XXXXX"
     },
     "granularity": "PT1M",
     "group_by": [
       "metric.topic"
     ],
     "intervals": [
       "2019-12-19T11:00:00-05:00/2019-12-19T11:05:00-05:00"
     ],
     "limit": 25
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < sent_bytes_query.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @sent_bytes_query.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2019-12-19T16:01:00Z",
         "metric.topic": "test-topic",
         "value": 0.0
       },
       {
         "timestamp": "2019-12-19T16:02:00Z",
         "metric.topic": "test-topic",
         "value": 157.0
       },
       {
         "timestamp": "2019-12-19T16:03:00Z",
         "metric.topic": "test-topic",
         "value": 371.0
       },
       {
         "timestamp": "2019-12-19T16:04:00Z",
         "metric.topic": "test-topic",
         "value": 0.0
       }
     ]
   }
   

   At the topic scope, these metrics emit data points only when there is active production or consumption during the requested time window. If there is no activity for a topic during the window, the dataset is empty for that topic.

Query for max retained bytes per hour over two hours for a cluster lkc-XXXXX

1. Create a file named cluster_retained_bytes_query.json using the following template. Be sure to change lkc-XXXXX and the timestamp values to match your needs:

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.server/retained_bytes"
       }
     ],
     "filter": {
       "field": "resource.kafka.id",
       "op": "EQ",
       "value": "lkc-XXXXX"
     },
     "granularity": "PT1H",
     "intervals": [
       "2019-12-19T11:00:00-05:00/P0Y0M0DT2H0M0S"
     ],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < cluster_retained_bytes_query.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @cluster_retained_bytes_query.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2019-12-19T16:00:00Z",
         "value": 507350.0
       },
       {
         "timestamp": "2019-12-19T17:00:00Z",
         "value": 507350.0
       }
     ]
   }
   

Query for average consumer lag over the last hour grouped by topic and consumer group

1. Create a file named consumer_lag_max_hour.json using the following template. Be sure to change lkc-XXXXX and note the interval is for the last hour with a one-minute granularity.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.server/consumer_lag_offsets"
       }
     ],
     "filter": {
       "field": "resource.kafka.id",
       "op": "EQ",
       "value": "lkc-XXXXX"
     },
     "granularity": "PT1M",
     "group_by": [
       "metric.consumer_group_id",
       "metric.topic"
     ],
     "intervals": [
       "PT1H/now"
     ],
     "limit": 25
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < consumer_lag_max_hour.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @consumer_lag_max_hour.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "metric.consumer_group_id": "group_1",
         "metric.topic": "test_topic_1",
         "timestamp": "2022-03-23T21:00:00Z",
         "value": 0.0
       },
       {
         "metric.consumer_group_id": "group_2",
         "metric.topic": "test_topic_2",
         "timestamp": "2022-03-23T21:00:00Z",
         "value": 6.0
       }
     ]
   }
   

Monitor Schema Registry

The Metrics API exposes Schema Registry metrics through the schema_registry resource type, identified by resource.schema_registry.id (for example, lsrc-XXXXX). The following metrics are available, each prefixed with io.confluent.kafka.schema_registry/:

• schema_count: Number of registered schemas.

• schema_operations_count: Delta count of schema create, delete, and read operations.

• request_count: Delta count of requests received by the Schema Registry server.

• exporter_tasks: Number of schema exporters by status (RUNNING, STARTING, or PAUSED).

• exporter_starting_progress: Progress of a schema exporter in the starting phase, from 0 to 100.

• schema_transfer_success_total: Count of schemas successfully registered by each exporter.

• num_deks: Number of data encryption keys.

• num_keks: Number of key encryption keys.

• num_keks_shared: Number of key encryption keys shared with Confluent.

For a worked example, see Query for the number of schemas in the Schema Registry cluster lsrc-XXXXX.

Query for the number of schemas in the Schema Registry cluster lsrc-XXXXX

1. Create a file named schema_count.json using the following template. Be sure to change lsrc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "time_agg": "MAX",
         "agg": "SUM",
         "metric": "io.confluent.kafka.schema_registry/schema_count"
       }
     ],
     "filter": {
       "field": "resource.schema_registry.id",
       "op": "EQ",
       "value": "lsrc-XXXXX"
     },
     "granularity": "PT1M",
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T10:01:00Z"
     ],
     "group_by": [
       "resource.schema_registry.id"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < schema_count.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @schema_count.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.schema_registry.id": "lsrc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "value": 1.0
       }
     ]
   }
   

Monitor connectors

The following examples query metrics for connectors.

Query for the hourly number of records received by a sink connector lcc-XXXXX

1. Create a file named sink_connector_record_number.json using the following template. Be sure to change lcc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.connect/received_records"
       }
     ],
     "filter": {
       "field": "resource.connector.id",
       "op": "EQ",
       "value": "lcc-XXXXX"
     },
     "granularity": "PT1H",
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ],
     "group_by": [
       "resource.connector.id"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your Confluent Cloud cluster credentials (--resource cloud credentials).

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < sink_connector_record_number.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @sink_connector_record_number.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.connector.id": "lcc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "value": 26455991.0
       }
     ]
   }
   

Query for the total free memory on a custom connector clcc-XXXXX

1. Create a file named custom_connector_free_memory.json using the following template. Be sure to change clcc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.system/memory_free_bytes"
       }
     ],
     "filter": {
       "field": "resource.custom_connector.id",
       "op": "EQ",
       "value": "clcc-XXXXX"
     },
     "granularity": "PT1H",
     "intervals": [
       "2023-05-09T10:00:00Z/2023-05-09T15:00:00Z"
     ],
     "group_by": [
       "resource.custom_connector.id"
     ]
   }
   

2. Submit the query as a POST using the following command.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud-custom/query' --auth '<API_KEY>:<SECRET>' < custom_connector_free_memory.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud-custom/query' -u '<API_KEY>:<SECRET>' -d @custom_connector_free_memory.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
      "data": [
         {
               "resource.custom_connector.id": "clcc-XXXXXX",
               "timestamp": "2023-05-09T10:00:00Z",
               "value": 125229329.06666666
         },
         {
               "resource.custom_connector.id": "clcc-XXXXXX",
               "timestamp": "2023-05-09T11:00:00Z",
               "value": 125193966.93333334
         },
         {
               "resource.custom_connector.id": "clcc-XXXXXX",
               "timestamp": "2023-05-09T12:00:00Z",
               "value": 125140241.06666666
         },
         {
               "resource.custom_connector.id": "clcc-XXXXXX",
               "timestamp": "2023-05-09T13:00:00Z",
               "value": 125099622.4
         },
         {
               "resource.custom_connector.id": "clcc-XXXXXX",
               "timestamp": "2023-05-09T14:00:00Z",
               "value": 124849493.33333333
         }
      ]
   }
   

For Cloud Console metrics for custom connectors, see View metrics.

Query for the total percent CPU used by a custom connector clcc-XXXXX

1. Create a file named custom_connector_percent_cpu.json using the following template. Be sure to change clcc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.system/cpu_load_percent"
       }
     ],
     "filter": {
       "field": "resource.custom_connector.id",
       "op": "EQ",
       "value": "clcc-XXXXX"
     },
     "granularity": "PT1H",
     "intervals": [
       "2023-05-09T10:00:00Z/2023-05-09T15:00:00Z"
     ],
     "group_by": [
       "resource.custom_connector.id"
     ]
   }
   

2. Submit the query as a POST using the following command.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud-custom/query' --auth '<API_KEY>:<SECRET>' < custom_connector_percent_cpu.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud-custom/query' -u '<API_KEY>:<SECRET>' -d @custom_connector_percent_cpu.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
      "data": [
         {
               "resource.custom_connector.id": "clcc-XXXXX",
               "timestamp": "2023-05-09T10:00:00Z",
               "value": 0.021009808092643977
         },
         {
               "resource.custom_connector.id": "clcc-XXXXX",
               "timestamp": "2023-05-09T11:00:00Z",
               "value": 0.01990721858932965
         },
         {
               "resource.custom_connector.id": "clcc-XXXXX",
               "timestamp": "2023-05-09T12:00:00Z",
               "value": 0.020799848444189233
         },
         {
               "resource.custom_connector.id": "clcc-XXXXX",
               "timestamp": "2023-05-09T13:00:00Z",
               "value": 0.019948515028905416
         },
         {
               "resource.custom_connector.id": "clcc-XXXXX",
               "timestamp": "2023-05-09T14:00:00Z",
               "value": 0.020734587261390117
         }
      ]
   }
   

For Cloud Console metrics for custom connectors, see View metrics.

Query for metrics for a specific principal ID

You can use the metric.principal_id label to filter metrics by specific users or service accounts. Metrics such as io.confluent.kafka.server/active_connection_count and io.confluent.kafka.server/request_count support filtering by the metric.principal_id label. This is particularly useful for monitoring and alerting on the activity of specific principals in your cluster. Each active metric.principal_id also provides a metric.principal_name label that contains the human-readable name of the principal, such as the username or service account name, if a name has been defined.

To see all metrics that currently support the metric.principal_id label, see the API Reference.

1. Create a file named principal_query.json using the following template. Be sure to change lkc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.server/active_connection_count"
       }
     ],
     "filter": {
       "field": "resource.kafka.id",
       "op": "EQ",
       "value": "lkc-XXXXX"
     },
     "granularity": "PT1H",
     "group_by": [
       "metric.principal_id"
     ],
     "intervals": [
       "2022-01-01T00:00:00Z/PT1H"
     ],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < principal_query.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @principal_query.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "metric.principal_id": "sa-abc123",
         "metric.principal_name": "Prod Service Account",
         "timestamp": "2022-01-01T00:00:00Z",
         "value": 430.99999999997
       },
       {
         "metric.principal_id": "u-def456",
         "metric.principal_name": "Mary Smith",
         "timestamp": "2022-01-01T00:00:00Z",
         "value": 427.93333333332
       },
       {
         "metric.principal_id": "u-abc123",
         "metric.principal_name": "Tom Nguyen",
         "timestamp": "2022-01-01T00:00:00Z",
         "value": 333.19999999997
       }
     ],
     "meta": {
       "pagination": {
         "next_page_token": "eyJ2ZXJzaW9uIjoiMSIsInJlcXVlc3RI",
         "page_size": 5
       }
     }
   }
   

   Topics without reported metric values during the specified interval aren’t returned.

Query for connection accept count to monitor ECKU scaling

This query measures new connections accepted by the cluster. Each sample is the number of connections accepted since the previous data point, sampled every 60 seconds. This metric corresponds to the connection attempts dimension used to determine ECKU scaling for elastic cluster types, helping you understand and monitor your ECKU usage.

1. Create a file named connection_accept_count_query.json using the following template. Be sure to change lkc-xyz and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.server/connection_accept_count"
       }
     ],
     "filter": {
       "field": "resource.kafka.id",
       "op": "EQ",
       "value": "lkc-xyz"
     },
     "granularity": "PT1M",
     "intervals": [
       "PT1H/now"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < connection_accept_count_query.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @connection_accept_count_query.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2026-04-14T20:30:00Z",
         "value": 12.0
       },
       {
         "timestamp": "2026-04-14T20:31:00Z",
         "value": 10.0
       },
       {
         "timestamp": "2026-04-14T20:32:00Z",
         "value": 13.0
       }
     ]
   }
   

   This metric provides visibility into all connection attempts, including failed ones, which are included in ECKU cost calculations.

Query for throttled clients on a cluster

This query returns the average throttle time applied to principals on a cluster, grouped by principal, violated limit type, and reason. For more information about this metric, see Throttled clients metric.

1. Create a file named throttled_clients_query.json using the following template. Be sure to change lkc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.server/client_limit_milliseconds"
       }
     ],
     "filter": {
       "field": "resource.kafka.id",
       "op": "EQ",
       "value": "lkc-XXXXX"
     },
     "granularity": "PT1M",
     "group_by": [
       "metric.principal_id",
       "metric.violated_limit",
       "metric.reason"
     ],
     "intervals": [
       "PT1H/now"
     ],
     "limit": 25
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < throttled_clients_query.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @throttled_clients_query.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2026-05-04T14:00:00Z",
         "metric.principal_id": "u-23agh7",
         "metric.violated_limit": "produce_throughput_quota",
         "metric.reason": "cluster_quota_violation",
         "value": 150.0
       },
       {
         "timestamp": "2026-05-04T14:00:00Z",
         "metric.principal_id": "sa-abc123",
         "metric.violated_limit": "fetch_throughput_quota",
         "metric.reason": "principal_quota_violation",
         "value": 75.5
       }
     ]
   }
   

Monitor Flink

The following examples query metrics for Flink compute pools and SQL statements.

Query for the total number of records a Flink SQL statement has received

1. Create a file named num_records_in.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), statement name (XXXXXXXX-XXXX-XXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/num_records_in"
       }
     ],
     "filter": {
       "op": "AND",
       "filters": [
         {
           "field": "resource.flink_statement.name",
           "op": "EQ",
           "value": "XXXXXXXX-XXXX-XXXX"
         },
         {
           "field": "resource.compute_pool.id",
           "op": "EQ",
           "value": "lfcp-XXXXXX"
         }
       ]
     },
     "granularity": "PT1M",
     "intervals": [
       "2023-10-23T16:30:00/2023-10-23T16:35:00"
     ],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < num_records_in.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @num_records_in.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2023-10-23T16:30:00Z",
         "value": 115.0
       },
       {
         "timestamp": "2023-10-23T16:31:00Z",
         "value": 116.0
       },
       {
         "timestamp": "2023-10-23T16:32:00Z",
         "value": 116.0
       },
       {
         "timestamp": "2023-10-23T16:33:00Z",
         "value": 131.0
       },
       {
         "timestamp": "2023-10-23T16:34:00Z",
         "value": 127.0
       }
     ]
   }
   

Query for the total number of records a Flink SQL statement has emitted

1. Create a file named num_records_out.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), statement name (XXXXXXXX-XXXX-XXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/num_records_out"
       }
     ],
     "filter": {
       "op": "AND",
       "filters": [
         {
           "field": "resource.flink_statement.name",
           "op": "EQ",
           "value": "XXXXXXXX-XXXX-XXXX"
         },
         {
           "field": "resource.compute_pool.id",
           "op": "EQ",
           "value": "lfcp-XXXXXX"
         }
       ]
     },
     "granularity": "PT1M",
     "intervals": [
       "2023-10-23T16:30:00/2023-10-23T16:35:00"
     ],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < num_records_out.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @num_records_out.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2023-10-23T16:30:00Z",
         "value": 115.0
       },
       {
         "timestamp": "2023-10-23T16:31:00Z",
         "value": 116.0
       },
       {
         "timestamp": "2023-10-23T16:32:00Z",
         "value": 116.0
       },
       {
         "timestamp": "2023-10-23T16:33:00Z",
         "value": 131.0
       },
       {
         "timestamp": "2023-10-23T16:34:00Z",
         "value": 127.0
       }
     ]
   }
   

Query for the backlog of a Flink SQL statement

This metric represents the total number of available records after the consumer offset in a Kafka partition for a Flink SQL statement, across all operators.

1. Create a file named pending_records.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), statement name (XXXXXXXX-XXXX-XXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/pending_records"
       }
     ],
     "filter": {
       "op": "AND",
       "filters": [
         {
           "field": "resource.flink_statement.name",
           "op": "EQ",
           "value": "XXXXXXXX-XXXX-XXXX"
         },
         {
           "field": "resource.compute_pool.id",
           "op": "EQ",
           "value": "lfcp-XXXXXX"
         }
       ]
     },
     "granularity": "PT1M",
     "intervals": [
       "2023-10-23T16:30:00/2023-10-23T16:35:00"
     ],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < pending_records.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @pending_records.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2023-10-23T16:30:00Z",
         "value": 0.0
       },
       {
         "timestamp": "2023-10-23T16:31:00Z",
         "value": 0.0
       },
       {
         "timestamp": "2023-10-23T16:32:00Z",
         "value": 0.0
       },
       {
         "timestamp": "2023-10-23T16:33:00Z",
         "value": 0.0
       },
       {
         "timestamp": "2023-10-23T16:34:00Z",
         "value": 0.0
       }
     ]
   }
   

   The above value might not always be zero. A non-zero value indicates some backlog associated with the Flink statement.

Query for the total number of records all Flink SQL statements using a Flink compute pool have received

1. Create a file named pool_num_records_in.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/num_records_in"
       }
     ],
     "filter": {
       "field": "resource.compute_pool.id",
       "op": "EQ",
       "value": "lfcp-XXXXXX"
     },
     "granularity": "PT1M",
     "intervals": ["2023-10-25T16:30:00/2023-10-25T16:35:00"],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < pool_num_records_in.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @pool_num_records_in.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2023-10-25T16:30:00Z",
         "value": 236.0
       },
       {
         "timestamp": "2023-10-25T16:31:00Z",
         "value": 228.0
       },
       {
         "timestamp": "2023-10-25T16:32:00Z",
         "value": 240.0
       },
       {
         "timestamp": "2023-10-25T16:33:00Z",
         "value": 230.0
       },
       {
         "timestamp": "2023-10-25T16:34:00Z",
         "value": 252.0
       }
     ]
   }
   

Query for the total number of records all Flink SQL statements using a Flink compute pool have emitted

1. Create a file named pool_num_records_out.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/num_records_out"
       }
     ],
     "filter": {
       "field": "resource.compute_pool.id",
       "op": "EQ",
       "value": "lfcp-XXXXXX"
     },
     "granularity": "PT1M",
     "intervals": ["2023-10-25T16:30:00/2023-10-25T16:35:00"],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < pool_num_records_out.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @pool_num_records_out.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2023-10-25T16:30:00Z",
         "value": 236.0
       },
       {
         "timestamp": "2023-10-25T16:31:00Z",
         "value": 228.0
       },
       {
         "timestamp": "2023-10-25T16:32:00Z",
         "value": 240.0
       },
       {
         "timestamp": "2023-10-25T16:33:00Z",
         "value": 230.0
       },
       {
         "timestamp": "2023-10-25T16:34:00Z",
         "value": 252.0
       }
     ]
   }
   

Query for the total backlog of all Flink SQL statements using a Flink compute pool

This metric represents the total number of available records after the consumer offset in a Kafka partition for all Flink SQL statements using a Flink compute pool, across all operators.

1. Create a file named pool_pending_records.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/pending_records"
       }
     ],
     "filter": {
       "field": "resource.compute_pool.id",
       "op": "EQ",
       "value": "lfcp-XXXXXX"
     },
     "granularity": "PT1M",
     "intervals": ["2023-10-25T16:30:00/2023-10-25T16:35:00"],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < pool_pending_records.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @pool_pending_records.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2023-10-25T16:30:00Z",
         "value": 0.0
       },
       {
         "timestamp": "2023-10-25T16:31:00Z",
         "value": 0.0
       },
       {
         "timestamp": "2023-10-25T16:32:00Z",
         "value": 0.0
       },
       {
         "timestamp": "2023-10-25T16:33:00Z",
         "value": 0.0
       },
       {
         "timestamp": "2023-10-25T16:34:00Z",
         "value": 0.0
       }
     ]
   }
   

   The above value might not always be zero. A non-zero value indicates the combined backlog associated with the Flink statements using the Flink compute pool in the query.

Query for the absolute number of CFUs at a given moment in a Flink compute pool

This metric represents the absolute number of CFUs or the current usage at a given moment in a Flink compute pool.

1. Create a file named current_cfus.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/compute_pool_utilization/current_cfus"
       }
     ],
     "filter": {
       "field": "resource.compute_pool.id",
       "op": "EQ",
       "value": "lfcp-XXXXXX"
     },
     "granularity": "PT1M",
     "intervals": ["2024-05-15T14:00:00/2024-05-15T14:05:00"],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < current_cfus.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @current_cfus.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2024-05-15T14:00:00Z",
         "value": 3.0
       },
       {
         "timestamp": "2024-05-15T14:01:00Z",
         "value": 3.0
       },
       {
         "timestamp": "2024-05-15T14:02:00Z",
         "value": 3.0
       },
       {
         "timestamp": "2024-05-15T14:03:00Z",
         "value": 3.0
       },
       {
         "timestamp": "2024-05-15T14:04:00Z",
         "value": 3.0
       }
     ]
   }
   

Query for the maximum number of CFUs assigned to a Flink compute pool

This metric represents the maximum number of CFUs assigned to a Flink compute pool. When Flink statements are running, the compute pool is autoscaled up to this maximum number of CFUs assigned to a Flink compute pool.

1. Create a file named cfu_limit.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/compute_pool_utilization/cfu_limit"
       }
     ],
     "filter": {
       "field": "resource.compute_pool.id",
       "op": "EQ",
       "value": "lfcp-XXXXXX"
     },
     "granularity": "PT1M",
     "intervals": ["2024-05-15T14:00:00/2024-05-15T14:05:00"],
     "limit": 5
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < cfu_limit.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @cfu_limit.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2024-05-15T14:00:00Z",
         "value": 10.0
       },
       {
         "timestamp": "2024-05-15T14:01:00Z",
         "value": 10.0
       },
       {
         "timestamp": "2024-05-15T14:02:00Z",
         "value": 10.0
       },
       {
         "timestamp": "2024-05-15T14:03:00Z",
         "value": 10.0
       },
       {
         "timestamp": "2024-05-15T14:04:00Z",
         "value": 10.0
       }
     ]
   }
   

Query for the state size of a Flink statement

This metric represents the state size of a Flink SQL statement.

1. Create a file named state_size.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/operator/state_size_bytes",
         "agg": "SUM"
       }
     ],
     "filter": {
       "op": "AND",
       "filters": [
         {
           "field": "resource.compute_pool.id",
           "op": "EQ",
           "value": "<compute-pool-id>"
         },
         {
           "field": "resource.flink_statement.name",
           "op": "EQ",
           "value": "<statement-name>"
         }
       ]
     },
     "granularity": "PT5M",
     "intervals": [
       "2026-01-23T00:00:00Z/2026-01-23T01:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < state_size.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @state_size.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "metadata": {
       "name": "<statement-name>",
       "self": "https://flink.<region>.<cloud>.confluent.cloud/sql/v3/organizations/<org-id>/environments/<env-id>/statements/<statement-name>"
     },
     "status": {
       "phase": "RUNNING",
       "stateLimitStatus": {
         "stateLimitState": "APPROACHING_SOFT_LIMIT",
         "message": "Statement state size is approaching the soft state size limit.",
         "lastUpdatedAt": "2026-01-23T09:15:00Z"
       },
       "scalingStatus": {
         "scalingState": "POOL_EXHAUSTED"
       }
     }
   }
   

Query for the statement status for a given Flink SQL statement

This metric represents the status of a Flink SQL statement.

1. Create a file named statement_status.json using the following template. Be sure to change the compute pool ID (lfcp-XXXXXX), and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.flink/statement_status"
       }
     ],
     "filter": {
       "op": "AND",
       "filters": [
         {
           "field": "resource.flink_statement.name",
           "op": "EQ",
           "value": "<statement-name>"
         },
         {
           "field": "resource.compute_pool.id",
           "op": "EQ",
           "value": "lfcp-XXXXXX"
         }
       ]
     },
     "granularity": "PT1M",
     "intervals": ["now-6h/now"],
     "group_by": [
       "resource.flink_statement.uid",
       "metric.status"
     ],
     "limit": 1000
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < statement_status.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @statement_status.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "timestamp": "2025-03-10T09:27:00Z",
         "value": 1.0,
         "resource.flink_statement.uid": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
         "metric.status": "RUNNING"
       },
       {
         "timestamp": "2025-03-10T09:32:00Z",
         "value": 1.0,
         "resource.flink_statement.uid": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
         "metric.status": "RUNNING"
       },
       {
         "timestamp": "2025-03-10T09:34:00Z",
         "value": 1.0,
         "resource.flink_statement.uid": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
         "metric.status": "RUNNING"
       },
       {
         "timestamp": "2025-03-10T09:36:00Z",
         "value": 1.0,
         "resource.flink_statement.uid": "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
         "metric.status": "RUNNING"
       }
     ]
   }
   

Monitor ksqlDB

The following examples query metrics for ksqlDB clusters.

Query for the number of streaming units used per hour for ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_streaming_unit_count.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/streaming_unit_count"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-XXXXX"
     },
     "granularity": "PT1H",
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ],
     "group_by": [
       "resource.ksql.id"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_streaming_unit_count.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_streaming_unit_count.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "value": 4.0
       }
     ]
   }
   

Query for the max % of storage used over all CSUs for a ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_storage_utilization.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/storage_utilization"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_storage_utilization.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_storage_utilization.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "value": 0.85
       }
     ]
   }
   

Query for the bytes of ksqlDB storage used by a query on ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_query_storage.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/task_stored_bytes"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "group_by": [
       "metric.query_id"
     ],
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_query_storage.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_query_storage.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "metric.query_id": "CTAS_PAGEVIEWS_2",
         "timestamp": "2021-02-24T10:00:00Z",
         "value": 7688174488
       }
     ]
   }
   

Query for the bytes of storage used by a task on ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_task_storage.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/task_stored_bytes"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "group_by": [
       "metric.query_id",
       "metric.task_id"
     ],
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_task_storage.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_task_storage.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "metric.task_id": "1_1",
         "metric.query_id": "CTAS_PAGEVIEWS_2",
         "timestamp": "2021-02-24T10:00:00Z",
         "value": 1079295760
       }
     ]
   }
   

Query for the query saturation on ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_query_saturation.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/query_saturation"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_query_saturation.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_query_saturation.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "value": 0.85
       }
     ]
   }
   

Query for the total bytes consumed by ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_bytes_consumed.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/consumed_total_bytes"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_bytes_consumed.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_bytes_consumed.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "value": 1024
       }
     ]
   }
   

Query for the total bytes produced by ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_bytes_produced.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/produced_total_bytes"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_bytes_produced.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_bytes_produced.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "value": 1024
       }
     ]
   }
   

Query for the total topic offsets processed by task on ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_offsets_processed.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/offsets_processed_total"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "group_by": [
         "metric.query_id",
         "metric.task_id"
     ],
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_offsets_processed.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_offsets_processed.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "metric.task_id": "1_1",
         "metric.query_id": "CTAS_PAGEVIEWS_2",
         "value": 123
       }
     ]
   }
   

Query for the total topic offsets processed by all tasks of query on ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_offsets_processed_by_query.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/offsets_processed_total"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "group_by": [
         "metric.query_id"
     ],
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_offsets_processed_by_query.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_offsets_processed_by_query.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "metric.query_id": "CTAS_PAGEVIEWS_2",
         "value": 123
       }
     ]
   }
   

Query for the current committed offset lag by task on ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_offset_lag.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/committed_offset_lag"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "group_by": [
         "metric.query_id",
         "metric.task_id"
     ],
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_offset_lag.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_offset_lag.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "metric.task_id": "1_1",
         "metric.query_id": "CTAS_PAGEVIEWS_2",
         "value": 456
       }
     ]
   }
   

Query for the current total committed offset lag for all tasks in query on ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_offset_lag_by_query.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/committed_offset_lag"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "group_by": [
         "metric.query_id"
     ],
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_offset_lag_by_query.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_offset_lag_by_query.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "metric.query_id": "CTAS_PAGEVIEWS_2",
         "value": 456
       }
     ]
   }
   

Query for the total number of processing errors by query on ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_processing_errors.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/processing_errors_total"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "group_by": [
         "metric.query_id"
     ],
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_processing_errors.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_processing_errors.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "metric.query_id": "CTAS_PAGEVIEWS_2",
         "value": 16
       }
     ]
   }
   

Query for the total number of restarts due to failure by query on ksqlDB cluster lksqlc-XXXXX

1. Create a file named ksql_query_restarts.json using the following template. Be sure to change lksqlc-XXXXX and the timestamp values to match your needs.

   {
     "aggregations": [
       {
         "metric": "io.confluent.kafka.ksql/query_restarts"
       }
     ],
     "filter": {
       "field": "resource.ksql.id",
       "op": "EQ",
       "value": "lksqlc-xxxxx"
     },
     "granularity": "PT1M",
     "group_by": [
         "metric.query_id"
     ],
     "intervals": [
       "2021-02-24T10:00:00Z/2021-02-24T11:00:00Z"
     ]
   }
   

2. Submit the query as a POST using the following command. Be sure to change API_KEY and SECRET to match your environments.

   HTTPie

   http 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' --auth '<API_KEY>:<SECRET>' < ksql_query_restarts.json
   

   cURL

   curl -X POST 'https://api.telemetry.confluent.cloud/v2/metrics/cloud/query' -u '<API_KEY>:<SECRET>' -d @ksql_query_restarts.json -H 'Content-Type: application/json'
   

   Your output should resemble:

   {
     "data": [
       {
         "resource.ksql.id": "lksqlc-XXXXX",
         "timestamp": "2021-02-24T10:00:00Z",
         "metric.query_id": "CTAS_PAGEVIEWS_2",
         "value": 3
       }
     ]
   }