---
title: "Per-topic consumer lag for a destination"
method: GET
path: "/destinations/{destination_id}/lag"
tags: ["Destinations"]
---

# Per-topic consumer lag for a destination

`GET /destinations/{destination_id}/lag`

## Request

- Base URL: `https://api.streamkap.com`
- URL: `https://api.streamkap.com/destinations/{destination_id}/lag`
- Auth: HTTP bearer
- TypeScript: <https://skmtc.dev/streamkap/apis/streamkap-rest-api/docs/destinations/:destination_id/lag/get.ts> (a function with typed inputs; its response is checked against the schema)

Current consumer lag per topic (default), or lag at a point in time when `at` is given. Values are totals: one sample per partition, summed per topic and across topics, so they can be reconciled with alerts. `at` within the last 2 days is served at minute resolution, within 14 days at hour resolution; older or future values return HTTP 400. When no sample exists near `at` the response carries `resolution: null` and a `message` instead of zeros. `latency_ms` is best-effort and only available in current mode. `at` without a timezone is treated as UTC. Destinations list every topic their consumer group reports, including topics without a stored record.

## Path parameters

- `destination_id` string, required

## Query parameters

- `at` string, date-time, nullable: Point in time (ISO 8601). Omit for current lag.
- `topic_name` string, nullable: Case-insensitive substring filter on the topic name
- `sort` 'lag' | 'name' | 'latency'
- `sort_dir` 'asc' | 'desc'
- `page` integer
- `page_size` integer
- `include_partitions` boolean: Return the per-partition breakdown for each topic

## Response `200`

Successful Response

- EntityLagRes
  - `entity_type` string, required
  - `entity_id` string, required
  - `connector_name` string, required
  - `supported` boolean, required: False for connectors that expose no consumer lag (kafkadirect)
  - `mode` 'current' | 'historical', required
  - `requested_at` string, date-time, nullable: Echo of the `at` query parameter
  - `as_of` string, date-time, nullable: Observation time actually used for the values
  - `resolution` 'raw' | 'minute' | 'hour', nullable: raw (current), minute (<= 2d), hour (<= 14d); null when no data was found
  - `stale` boolean, required: Current mode: newest sample is older than the stale threshold or missing
  - `message` string, nullable: Human-readable note when data is missing or unsupported
  - `overall` LagOverallRes, required
    - `lag` integer, required: Total lag across every topic and partition
    - `latency_ms` integer, nullable: Worst partition latency across topics; current mode only
    - `topic_count` integer, required: Topics resolved for the entity (before topic_name filtering)
    - `topics_with_lag` integer, required: Topics whose lag is > 0
  - `topics` LagTopicRes[], required
    - `topic` string, required: Display name: the stored topic_id when known, otherwise the Kafka label
    - `topic_id` string, nullable: Topic id as used by the topic details page (the stored topic name), null when the topic has no record
    - `kafka_topic` string, required: Topic label as written by the JMX scrape (dots replaced by underscores)
    - `producer_type` string, nullable: `sources` | `transforms` | null when unknown
    - `lag` integer, required: Sum of partition lag, one value per partition
    - `share` number, required: lag / overall.lag, 0 when overall is 0
    - `partition_count` integer, required
    - `max_partition_lag` integer, required
    - `latency_ms` integer, nullable: Worst partition latency; current mode only
    - `partitions` LagPartitionRes[], nullable: Only with include_partitions=true
      - `partition` string, required
      - `lag` integer, required: Records behind the log end offset for this partition
      - `latency_ms` integer, nullable: Wall-clock age of the oldest unconsumed record
  - `total` integer, required: Topics matching the filter, before pagination
  - `page` integer, required
  - `page_size` integer, required

## Other responses

- `422`: Validation Error

## Changes

- **2026-10-01** `50e54c2e65f2`: 1 info
  - endpoint added

[Operation history](https://skmtc.dev/streamkap/apis/streamkap-rest-api/changes/destinations/:destination_id/lag/get.md)

---

[API](https://skmtc.dev/streamkap/apis/streamkap-rest-api.md) - [All operations](https://skmtc.dev/streamkap/apis/streamkap-rest-api/llms.txt) - [OpenAPI document](https://skmtc.dev/streamkap/apis/streamkap-rest-api/revisions/e52ab3a4b1f0?raw)