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 

Monitoring Kafka clusters and clients shows you how much data is flowing through a cluster, whether consumer groups are keeping up, and which principals are consuming cluster resources. 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.

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
}

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.

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
}

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 for a cluster `lkc-XXXXX`

This query measures the maximum number of bytes retained by a cluster, sampled hourly over a two-hour window.

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
}

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 by topic and consumer group

This query measures average consumer lag over the last hour, grouped by topic and consumer group, so you can identify which consumers are falling behind.

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
}

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/:

Metric	Description
`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`

This query measures the number of registered schemas in a Schema Registry cluster at a given time.

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"
  ]
}

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
    }
  ]
}

Run Example Queries for the Confluent Cloud Metrics API

Monitor Kafka clusters and clients 

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

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

Query for max retained bytes for a cluster `lkc-XXXXX`

Query for average consumer lag by topic and consumer group

Monitor Schema Registry 

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

Monitor connectors 

Query for hourly records received by a sink connector `lcc-XXXXX`

Query for free memory on a custom connector `clcc-XXXXX`

Query for percent CPU used by a custom connector `clcc-XXXXX`

Query for metrics for a specific principal ID

Query for connection accept count to monitor eCKU scaling

Query for throttled clients on a cluster

Monitor Flink 

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

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

Query for the backlog of a Flink SQL statement

Query for records received by all Flink SQL statements in a Flink compute pool

Query for records emitted by all Flink SQL statements in a Flink compute pool

Query for the backlog of all Flink SQL statements in a Flink compute pool

Query for the current number of CFUs in a Flink compute pool

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

Query for the state size of a Flink statement

Query for the statement status for a given Flink SQL statement

Monitor ksqlDB 

Query for hourly streaming units for ksqlDB cluster `lksqlc-XXXXX`

Query for max storage percent used by a ksqlDB cluster `lksqlc-XXXXX`

Query for storage bytes used by a query on ksqlDB cluster `lksqlc-XXXXX`

Query for storage bytes used by a task on ksqlDB cluster `lksqlc-XXXXX`

Query for the query saturation on ksqlDB cluster `lksqlc-XXXXX`

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

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

Query for topic offsets processed by a task on ksqlDB cluster `lksqlc-XXXXX`

Query for total offsets by all tasks of a query on ksqlDB cluster `lksqlc-XXXXX`

Query for committed offset lag by task on ksqlDB cluster `lksqlc-XXXXX`

Query for total committed offset lag for a query on ksqlDB cluster `lksqlc-XXXXX`

Query for processing errors by query on ksqlDB cluster `lksqlc-XXXXX`

Query for restarts due to failure by query on ksqlDB cluster `lksqlc-XXXXX`

Run Example Queries for the Confluent Cloud Metrics API

Monitor Kafka clusters and clients

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

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

Query for max retained bytes for a cluster lkc-XXXXX

Query for average consumer lag by topic and consumer group

Monitor Schema Registry

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

Monitor connectors

Query for hourly records received by a sink connector lcc-XXXXX

Query for free memory on a custom connector clcc-XXXXX

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

Query for metrics for a specific principal ID

Query for connection accept count to monitor eCKU scaling

Query for throttled clients on a cluster

Monitor Flink

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

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

Query for the backlog of a Flink SQL statement

Query for records received by all Flink SQL statements in a Flink compute pool

Query for records emitted by all Flink SQL statements in a Flink compute pool

Query for the backlog of all Flink SQL statements in a Flink compute pool

Query for the current number of CFUs in a Flink compute pool

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

Query for the state size of a Flink statement

Query for the statement status for a given Flink SQL statement

Monitor ksqlDB

Query for hourly streaming units for ksqlDB cluster lksqlc-XXXXX

Query for max storage percent used by a ksqlDB cluster lksqlc-XXXXX

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

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

Query for the query saturation on ksqlDB cluster lksqlc-XXXXX

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

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

Query for topic offsets processed by a task on ksqlDB cluster lksqlc-XXXXX

Query for total offsets by all tasks of a query on ksqlDB cluster lksqlc-XXXXX

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

Query for total committed offset lag for a query on ksqlDB cluster lksqlc-XXXXX

Query for processing errors by query on ksqlDB cluster lksqlc-XXXXX

Query for restarts due to failure by query on ksqlDB cluster lksqlc-XXXXX

Monitor Kafka clusters and clients 

Query for max retained bytes for a cluster `lkc-XXXXX`

Monitor Schema Registry 

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

Monitor connectors 

Query for hourly records received by a sink connector `lcc-XXXXX`

Query for free memory on a custom connector `clcc-XXXXX`

Query for percent CPU used by a custom connector `clcc-XXXXX`

Monitor Flink 

Monitor ksqlDB 

Query for hourly streaming units for ksqlDB cluster `lksqlc-XXXXX`

Query for max storage percent used by a ksqlDB cluster `lksqlc-XXXXX`

Query for storage bytes used by a query on ksqlDB cluster `lksqlc-XXXXX`

Query for storage bytes used by a task on ksqlDB cluster `lksqlc-XXXXX`

Query for the query saturation on ksqlDB cluster `lksqlc-XXXXX`

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

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

Query for topic offsets processed by a task on ksqlDB cluster `lksqlc-XXXXX`

Query for total offsets by all tasks of a query on ksqlDB cluster `lksqlc-XXXXX`

Query for committed offset lag by task on ksqlDB cluster `lksqlc-XXXXX`

Query for total committed offset lag for a query on ksqlDB cluster `lksqlc-XXXXX`

Query for processing errors by query on ksqlDB cluster `lksqlc-XXXXX`

Query for restarts due to failure by query on ksqlDB cluster `lksqlc-XXXXX`