kafka

Kafka REST v3 dataplane for managing topics, partitions, consumer groups, ACLs, configs, records, cluster linking, share groups, and streams groups against a specific Kafka cluster. Targets the Confluent Kafka REST v3 surface as implemented by Confluent Cloud; the same surface is also implemented by Confluent Platform, Karapace, and (partially) Redpanda's HTTP Proxy.

Related provider

This provider covers Kafka dataplane operations against a specific Kafka cluster. For Confluent Cloud control-plane operations (org, environments, IAM, managed cluster lifecycle, schema registry control, connectors, Flink, networking, billing), see the confluent provider.

Provider Summary

total services: 3
total resources: 31
source project: stackql-provider-confluent

See also: [SHOW] [DESCRIBE] [REGISTRY]

---

Installation

To pull the latest version of the kafka provider, run the following command:

REGISTRY PULL kafka;

To view previous provider versions or to pull a specific provider version, see here.

Authentication

The kafka provider talks to a per-cluster Kafka REST v3 endpoint, not the Confluent Cloud control plane. It requires a cluster-scoped Resource API key (created in the Confluent UI under Cluster -> API Keys), which is distinct from the org-level Cloud API key used by the confluent provider.

The following system environment variables are used for authentication by default:

• KAFKA_API_KEY - Cluster-scoped Resource API key (see Confluent Cloud API Keys)
• KAFKA_API_SECRET - Cluster-scoped Resource API secret

These variables are sourced at runtime (from the local machine or as CI variables/secrets).

The per-cluster host is supplied at query time as three WHERE-clause bindings (kafka_endpoint_id, region, cloud_provider). These can be read out of the cluster's spec in the confluent provider:

SELECT
  spec.http_endpoint,
  spec.region,
  spec.cloud
FROM confluent.managed_kafka_clusters.clusters
WHERE environment = 'env-xxxxx';

then passed through to a kafka query:

SELECT cluster_id, topic_name, partitions_count
FROM kafka.kafka.topics
WHERE cluster_id = 'lkc-xxxxx'
  AND kafka_endpoint_id = 'pkc-xxxxx'
  AND region = 'ap-southeast-2'
  AND cloud_provider = 'aws';

Using different environment variables

To use different environment variables (instead of the defaults), use the --auth flag of the stackql program. For example:

AUTH='{ "kafka": { "type": "basic", "username_var": "MY_KAFKA_API_KEY_VAR", "password_var": "MY_KAFKA_API_SECRET_VAR" }}'
stackql shell --auth="${AUTH}"

or using PowerShell:

$Auth = "{ 'kafka': { 'type': 'basic', 'username_var': 'MY_KAFKA_API_KEY_VAR', 'password_var': 'MY_KAFKA_API_SECRET_VAR' }}"
stackql.exe shell --auth=$Auth

Example Queries

Try the following queries using stackql shell, or run them from a script or CI pipeline with stackql exec.

Topics in a cluster

Every topic in the cluster with its partition count and replication factor:

SELECT topic_name, partitions_count, replication_factor, is_internal
FROM kafka.kafka.topics
WHERE cluster_id = 'lkc-xxxxx'
  AND kafka_endpoint_id = 'pkc-xxxxx'
  AND region = 'ap-southeast-2'
  AND cloud_provider = 'aws'
ORDER BY topic_name;

Topic configuration overrides

Configuration values set explicitly on a topic rather than inherited from the cluster defaults, across every topic in the cluster:

SELECT topic_name, name, value
FROM kafka.kafka.topic_configs
WHERE cluster_id = 'lkc-xxxxx'
  AND kafka_endpoint_id = 'pkc-xxxxx'
  AND region = 'ap-southeast-2'
  AND cloud_provider = 'aws'
  AND source = 'DYNAMIC_TOPIC_CONFIG'
ORDER BY topic_name, name;

Partitions

Partitions of a topic, with the REST links to each partition's leader and replica set:

SELECT topic_name, partition_id,
       json_extract(leader, '$.related') AS leader,
       json_extract(replicas, '$.related') AS replicas
FROM kafka.kafka.topic_partitions
WHERE cluster_id = 'lkc-xxxxx'
  AND topic_name = 'orders'
  AND kafka_endpoint_id = 'pkc-xxxxx'
  AND region = 'ap-southeast-2'
  AND cloud_provider = 'aws'
ORDER BY partition_id;

Consumer groups

Consumer groups on the cluster with their state, group type and assignment strategy:

SELECT consumer_group_id, state, type, partition_assignor, is_simple
FROM kafka.kafka.consumer_groups
WHERE cluster_id = 'lkc-xxxxx'
  AND kafka_endpoint_id = 'pkc-xxxxx'
  AND region = 'ap-southeast-2'
  AND cloud_provider = 'aws'
ORDER BY consumer_group_id;

Consumer lag

Total and maximum lag for a consumer group, with the partition and consumer that are furthest behind:

SELECT consumer_group_id, total_lag, max_lag,
       max_lag_topic_name, max_lag_partition_id, max_lag_consumer_id
FROM kafka.kafka.consumers_lag_summary
WHERE cluster_id = 'lkc-xxxxx'
  AND consumer_group_id = 'orders-processor'
  AND kafka_endpoint_id = 'pkc-xxxxx'
  AND region = 'ap-southeast-2'
  AND cloud_provider = 'aws';

Per-partition lag for the same group, largest first:

SELECT topic_name, partition_id, consumer_id, current_offset, log_end_offset, lag
FROM kafka.kafka.consumers_lags
WHERE cluster_id = 'lkc-xxxxx'
  AND consumer_group_id = 'orders-processor'
  AND kafka_endpoint_id = 'pkc-xxxxx'
  AND region = 'ap-southeast-2'
  AND cloud_provider = 'aws'
ORDER BY lag DESC;

ACLs

ACL bindings on the cluster, one row per binding, expanding the data array with json_each (the UserV2:* principal filter returns service accounts in sa-xxxxx form rather than numeric IDs):

SELECT json_extract(a.value, '$.principal') AS principal,
       json_extract(a.value, '$.resource_type') AS resource_type,
       json_extract(a.value, '$.resource_name') AS resource_name,
       json_extract(a.value, '$.pattern_type') AS pattern_type,
       json_extract(a.value, '$.operation') AS operation,
       json_extract(a.value, '$.permission') AS permission
FROM kafka.kafka.acls l, json_each(l.data) a
WHERE l.cluster_id = 'lkc-xxxxx'
  AND l.kafka_endpoint_id = 'pkc-xxxxx'
  AND l.region = 'ap-southeast-2'
  AND l.cloud_provider = 'aws'
  AND l.principal = 'UserV2:*'
ORDER BY principal, resource_type, resource_name;

Topic lifecycle

Create a topic with a retention override (Confluent Cloud requires a replication factor of 3), then delete it:

INSERT INTO kafka.kafka.topics (
  topic_name, partitions_count, replication_factor, configs,
  cluster_id, kafka_endpoint_id, region, cloud_provider
)
SELECT 'orders', 6, 3,
       '[{"name": "retention.ms", "value": "604800000"}]',
       'lkc-xxxxx', 'pkc-xxxxx', 'ap-southeast-2', 'aws';

DELETE FROM kafka.kafka.topics
WHERE cluster_id = 'lkc-xxxxx'
  AND topic_name = 'orders'
  AND kafka_endpoint_id = 'pkc-xxxxx'
  AND region = 'ap-southeast-2'
  AND cloud_provider = 'aws';

Services

kafka
share_group

streams_group