Confluent Cloud Optimisation Scripts

A collection of Python scripts for analysing and optimising Confluent Cloud Kafka clusters.

Table of Contents

• Overview
• Prerequisites
• Authentication & Environment Variables
• Service Account Permissions (Confluent Cloud)
• Scripts
  • find_orphaned_topics.py
  • tag_topics_for_deletion.py
  • tag_schemas_for_deletion.py
  • apply_deny_acls_to_topics.py
  • find_unused_schemas.py
  • find_unused_schema_versions.py
  • find_overlapping_schema_ids.py
  • remove_permissions_for_topics_to_delete.py
  • delete_topics.py
  • find_idle_flink_statements.py
  • delete_flink_statements.py
  • idle_connectors.py
  • dump_connector_configs.py
  • consumer_details.py
  • create_test_topics.py
  • consume_sample_messages.py
  • migration/find_migration_candidates.py
• Recommended Workflow
• API Reference
• Notes

Overview

Script	Purpose	Output
find_orphaned_topics.py	Identify inactive / orphaned topics	orphaned_topics.csv
tag_topics_for_deletion.py	Tag topics in Stream Catalog for decommissioning	(API side-effect)
apply_deny_acls_to_topics.py	Create explicit DENY READ/WRITE Kafka ACLs for topics	(API side-effect)
find_unused_schemas.py	Detect Schema Registry schema versions that could be deleted, and optionally hard delete them from CSV input	unused_schemas.csv
find_unused_schema_versions.py	Detect Schema Registry schema versions older than their topic's retention window, and optionally hard delete them from CSV input	unused_schema_versions.csv
find_overlapping_schema_ids.py	Detect global schema IDs referenced by more than one subject/version across all schema contexts	overlapping_schema_ids.csv
tag_schemas_for_deletion.py	Tag schema subject versions in Stream Catalog for decommissioning	(API side-effect)
remove_permissions_for_topics_to_delete.py	Revoke Kafka ACLs and/or RBAC role bindings for topics marked for deletion	(API side-effect)
delete_topics.py	Permanently delete confirmed orphan topics	(API side-effect)
find_idle_flink_statements.py	Identify idle or stuck Flink SQL statements wasting CFUs	idle_flink_statements.csv
delete_flink_statements.py	Stop and/or delete idle Flink SQL statements	(API side-effect)
idle_connectors.py	Find connectors that are not processing data	idle_connectors.csv
dump_connector_configs.py	Export all connector configs to JSON	connector_configs/
consumer_details.py	Analyse consumer group health and classify zombie/idle groups	consumer_groups.csv, consumer_lag_by_group_topic.csv, idle_consumers.csv
create_test_topics.py	Create a batch of test topics, optionally producing dummy messages to each	(API side-effect)
consume_sample_messages.py	Consume the earliest message from every topic to verify data exists	(console output)
migration/find_migration_candidates.py	Build a per-principal workload inventory for Dedicated → Enterprise migration assessment	migration_candidates.csv

---

Prerequisites

Python 3.10+ recommended.

pip install -r requirements.txt

The confluent-kafka package is optional — it is only needed when using the CHECK_LAST_MESSAGE=true flag in find_orphaned_topics.py.

---

Authentication & Environment Variables

All scripts read credentials from environment variables. Never hard-code credentials in source files.

Common variables

Variable	Required by	Description
CONFLUENT_REST_ENDPOINT	find_orphaned_topics, consumer_details, delete_topics, apply_deny_acls_to_topics, remove_permissions_for_topics_to_delete (ACLs)	Kafka REST Proxy base URL, e.g. https://pkc-xxx.us-east-1.aws.confluent.cloud:443
CONFLUENT_CLUSTER_ID	find_orphaned_topics, tag_topics_for_deletion, consumer_details, delete_topics, remove_permissions_for_topics_to_delete	Cluster ID, e.g. lkc-xxxxx
CONFLUENT_API_KEY	find_orphaned_topics, consumer_details, delete_topics, apply_deny_acls_to_topics, find_unused_schemas (detect), remove_permissions_for_topics_to_delete (ACLs)	Kafka cluster API key
CONFLUENT_API_SECRET	find_orphaned_topics, consumer_details, delete_topics, apply_deny_acls_to_topics, find_unused_schemas (detect), remove_permissions_for_topics_to_delete (ACLs)	Kafka cluster API secret
CONFLUENT_CLOUD_API_KEY	find_orphaned_topics, consumer_details, find_idle_flink_statements (Metrics API), idle_connectors (single-cluster), dump_connector_configs (single-cluster), remove_permissions_for_topics_to_delete (RBAC), migration/find_migration_candidates	Confluent Cloud API key
CONFLUENT_CLOUD_API_SECRET	same as above	Confluent Cloud API secret
CONFLUENT_ENVIRONMENT_ID	find_idle_flink_statements, idle_connectors (single-cluster), dump_connector_configs (single-cluster), remove_permissions_for_topics_to_delete (RBAC), migration/find_migration_candidates	Environment ID, e.g. env-xxxxx
CONFLUENT_ORG_ID	find_idle_flink_statements, remove_permissions_for_topics_to_delete (RBAC), migration/find_migration_candidates (optional)	Organisation ID, e.g. abc-123. Required by find_idle_flink_statements.py; auto-fetched for RBAC mode if not set.
CONFLUENT_FLINK_API_KEY	find_idle_flink_statements	Flink API key (scoped to environment/region/cloud provider)
CONFLUENT_FLINK_API_SECRET	find_idle_flink_statements	Flink API secret
CONFLUENT_FLINK_REGION	find_idle_flink_statements	Cloud region where the Flink compute pool runs, e.g. us-east-1, europe-west1
CONFLUENT_FLINK_CLOUD	find_idle_flink_statements	Cloud provider for Flink, e.g. aws, gcp, azure
CONFLUENT_SCHEMA_REGISTRY_URL	tag_topics_for_deletion, tag_schemas_for_deletion, find_unused_schemas	Schema Registry / Stream Catalog base URL
CONFLUENT_SCHEMA_REGISTRY_API_KEY	tag_topics_for_deletion, tag_schemas_for_deletion, find_unused_schemas	Schema Registry API key
CONFLUENT_SCHEMA_REGISTRY_API_SECRET	tag_topics_for_deletion, tag_schemas_for_deletion, find_unused_schemas	Schema Registry API secret

A .env file pattern (use export or a tool like direnv):

# Kafka cluster credentials
export CONFLUENT_REST_ENDPOINT="https://pkc-xxxxx.us-east-1.aws.confluent.cloud:443"
export CONFLUENT_CLUSTER_ID="lkc-xxxxx"
export CONFLUENT_API_KEY="your-kafka-api-key"
export CONFLUENT_API_SECRET="your-kafka-api-secret"

# Confluent Cloud credentials (for Metrics API, Connect API)
export CONFLUENT_CLOUD_API_KEY="your-cloud-api-key"
export CONFLUENT_CLOUD_API_SECRET="your-cloud-api-secret"
export CONFLUENT_ENVIRONMENT_ID="env-xxxxx"
export CONFLUENT_ORG_ID=""              # required for find_idle_flink_statements; auto-fetched for RBAC mode

# Flink SQL REST API credentials (for find_idle_flink_statements.py)
# The Flink API uses a regional endpoint and its own API key, separate from the Cloud API key.
export CONFLUENT_FLINK_API_KEY="your-flink-api-key"
export CONFLUENT_FLINK_API_SECRET="your-flink-api-secret"
export CONFLUENT_FLINK_REGION="us-east-1"   # region where your Flink compute pool runs
export CONFLUENT_FLINK_CLOUD="aws"           # cloud provider: aws, gcp, or azure

# Schema Registry / Stream Catalog (for tag_topics_for_deletion.py)
export CONFLUENT_SCHEMA_REGISTRY_URL="https://psrc-xxxxx.us-east-1.aws.confluent.cloud"
export CONFLUENT_SCHEMA_REGISTRY_API_KEY="your-sr-api-key"
export CONFLUENT_SCHEMA_REGISTRY_API_SECRET="your-sr-api-secret"

---

Service Account Permissions (Confluent Cloud)

Each script authenticates with one or more API key types. The table below shows the minimum Confluent Cloud RBAC roles required for each service account so that the scripts function correctly without over-provisioning access.

Kafka Cluster API key (CONFLUENT_API_KEY / CONFLUENT_API_SECRET)

Used to authenticate against the Kafka REST v3 API for topic listing and consumer group operations.

Script	Required role	Scope
find_orphaned_topics.py	DeveloperRead	Topic * and Group * on the target cluster
consumer_details.py	DeveloperRead	Topic * and Group * on the target cluster
apply_deny_acls_to_topics.py	CloudClusterAdmin	Target cluster
find_unused_schemas.py (detect)	Schema Registry read access; optionally DeveloperRead on the target cluster for topic existence checks	Environment scope + optional cluster scope
delete_topics.py	CloudClusterAdmin	Target cluster
remove_permissions_for_topics_to_delete.py (ACLs)	CloudClusterAdmin	Target cluster
migration/find_migration_candidates.py	DeveloperRead	Topic * and Group * on the target cluster (read-only; the script only lists topics/configs/ACLs, it makes no ACL/RBAC changes)

In Confluent Cloud RBAC, assign DeveloperRead to the service account at the Kafka cluster resource level, covering both Topic and Group resource types.

CloudClusterAdmin is required by delete_topics.py and remove_permissions_for_topics_to_delete.py (when revoking ACLs) because deleting topics and managing ACL entries via the REST v3 API requires elevated cluster access. Consider using a separate, dedicated service account for these steps rather than upgrading the read-only audit account.

Confluent Cloud API key (CONFLUENT_CLOUD_API_KEY / CONFLUENT_CLOUD_API_SECRET)

Used for the Confluent Cloud Metrics API and the Connect REST API.

Script	Required role	Scope
find_orphaned_topics.py	MetricsViewer	Organisation level
consumer_details.py	MetricsViewer	Organisation level
find_idle_flink_statements.py	MetricsViewer (Metrics API only)	Organisation level
idle_connectors.py	Operator (listing/status only) or CloudClusterAdmin (offsets)	Target cluster(s)
dump_connector_configs.py	CloudClusterAdmin	Target cluster(s)
remove_permissions_for_topics_to_delete.py (RBAC)	OrganizationAdmin or EnvironmentAdmin	Organisation or Environment level
migration/find_migration_candidates.py	MetricsViewer (Metrics API) + read access to /iam/v2/role-bindings (e.g. EnvironmentAdmin or OrganizationAdmin, read-only usage) + read access to the Connect API (e.g. Operator at cluster scope) + optionally BillingAdmin (Billing API — see below)	Organisation level (Metrics, Billing), Organisation/Environment level (role-binding lookups), target cluster(s) (Connect)

Operator is sufficient to list connectors and read their status (used by idle_connectors.py). However, reading connector offsets (which is how idle_connectors.py detects idle connectors) requires CloudClusterAdmin. With only Operator, the offsets endpoint returns 403 and connectors are classified as NO_OFFSETS rather than ACTIVE/IDLE — assign CloudClusterAdmin for accurate results. Note that some fully-managed connector classes (e.g. Datagen Source) are not on Confluent Cloud's supported offsets list at all — for these, the offsets endpoint returns 403 regardless of role, and idle_connectors.py classifies them as NO_OFFSETS rather than treating it as a permissions problem. CloudClusterAdmin is also required by dump_connector_configs.py because reading full connector configurations — including redacted sensitive fields — requires elevated cluster access.

remove_permissions_for_topics_to_delete.py requires OrganizationAdmin or EnvironmentAdmin when revoking RBAC role bindings, because only these roles can list and delete IAM v2 role bindings. Consider using a separate, dedicated service account for this step.

Schema Registry API key (CONFLUENT_SCHEMA_REGISTRY_API_KEY / CONFLUENT_SCHEMA_REGISTRY_API_SECRET)

Used by tag_topics_for_deletion.py and tag_schemas_for_deletion.py to apply and remove Stream Catalog tags and business metadata.

Script	Required role	Scope
tag_topics_for_deletion.py	DataSteward	Environment scope (env-xxxxx)
tag_schemas_for_deletion.py	DataSteward	Environment scope (env-xxxxx)
find_unused_schemas.py	read access to Schema Registry subjects and delete permission for hard delete mode	Environment scope

DataSteward is required — it grants create/manage/read/write/delete access to Stream Catalog tags and business metadata, including creating new tag type definitions via /catalog/v1/types/tagdefs. ResourceOwner is not sufficient for this and will result in a 403 error.

Grant it with:

confluent iam rbac role-binding create \
  --role DataSteward \
  --principal User:<SERVICE_ACCOUNT_ID> \
  --environment <ENVIRONMENT_ID>

Minimum recommended service account setup

Read-only audit (orphan detection + consumer analysis):

1. Create a service account in Confluent Cloud (IAM → Service Accounts).
2. Generate a Kafka cluster API key scoped to that service account. Assign DeveloperRead on the cluster for both Topic * and Group *.
3. Generate a Cloud API key for the service account. Assign MetricsViewer at the organisation level.

Add connector inspection (idle_connectors.py):

4. Assign Operator on each target cluster to the Cloud API key's service account for listing and status. To also read connector offsets (required for IDLE vs ACTIVE classification), assign CloudClusterAdmin instead.

Add connector config dump (dump_connector_configs.py):

5. Upgrade the cluster-level role to CloudClusterAdmin (replaces Operator).

Add topic tagging (tag_topics_for_deletion.py) or schema tagging (tag_schemas_for_deletion.py):

6. Generate a Schema Registry API key for the service account. Assign DataSteward at the environment scope (not cluster scope).

Revoke access for decommissioned topics (remove_permissions_for_topics_to_delete.py) and delete topics (delete_topics.py):

Use a separate, dedicated service account for these steps — the roles required here are significantly broader than those needed for read-only auditing.

7. For ACL revocation: assign CloudClusterAdmin on the target cluster to the Kafka cluster API key's service account.
8. For RBAC revocation: assign OrganizationAdmin or EnvironmentAdmin to the Cloud API key's service account.
9. For topic deletion (delete_topics.py): assign CloudClusterAdmin on the target cluster to the Kafka cluster API key's service account (same role as ACL revocation — a single service account can cover both steps).

# Grant CloudClusterAdmin for ACL removal
confluent iam rbac role-binding create \
  --role CloudClusterAdmin \
  --principal User:<SERVICE_ACCOUNT_ID> \
  --cloud-cluster <CLUSTER_ID> \
  --environment <ENVIRONMENT_ID>

# Grant EnvironmentAdmin for RBAC role-binding removal
confluent iam rbac role-binding create \
  --role EnvironmentAdmin \
  --principal User:<SERVICE_ACCOUNT_ID> \
  --environment <ENVIRONMENT_ID>

---

Scripts

find_orphaned_topics.py

Identifies topics that are inactive based on throughput metrics, and optionally consumer lag and last message age.

Detection logic

A topic is flagged as orphan_candidate=true when all of the following conditions are met:

total_bytes (received_bytes + sent_bytes)
    over the observation window  <  ORPHAN_BYTES_THRESHOLD  (default 1 MB)

AND (if CHECK_CONSUMER_LAG=true)
    no consumer group has positive lag on the topic

AND (if CHECK_LAST_MESSAGE=true, used as additional signal)
    the most recent message timestamp across all partitions
    is older than LAST_MESSAGE_THRESHOLD_DAYS  (default 30 days)

Why each signal matters:

Signal	Why it is used	Limitation
received_bytes (produced to topic)	Direct measure of producer activity	Near-zero for compacted topics with infrequent updates
sent_bytes (consumed from topic)	Direct measure of consumer delivery	A paused consumer still gets zero sent_bytes even if data exists. Conversely, any message read from a topic — via CHECK_LAST_MESSAGE=true, the Confluent Cloud Message Browser, kcat, or any other consumer — counts as real sent_bytes egress and can make a genuinely orphaned topic look active for the rest of the observation window
Consumer lag	Non-zero lag means a group is still interested in the topic	Stale groups (ZOMBIE) can hold lag indefinitely without actually consuming
Last message timestamp	Confirms no data has been written recently, regardless of throughput	Requires confluent-kafka binary protocol connection

Why throughput alone is not enough: A topic with a compacted cleanup policy accumulates very little byte traffic (only new/changed keys are transmitted) but may still be actively read by applications. This is why consumer lag is checked as a second gate — if any group is behind, the topic is kept off the orphan list regardless of bytes.

Why consumer lag alone is not enough: Zombie consumer groups (stalled, broken, or orphaned groups) can hold positive lag on dead topics indefinitely. Relying solely on lag would keep those topics alive forever. The bytes check catches topics where both producting and consuming have genuinely stopped.

The combination: Only topics that show both near-zero data movement and no active consumer interest are flagged. This two-gate approach significantly reduces false positives.

How it works (steps):

1. Lists all topics via Kafka REST v3 API (internal topics excluded by default).
2. Queries received_bytes and sent_bytes per topic over the last N days via the Confluent Cloud Metrics API, fetched concurrently. Each query requests a single aggregated total for the whole window (granularity: ALL) rather than a daily breakdown, since only the total is used — this keeps pagination proportional to topic count instead of topic count × days.
3. Iterates all consumer groups via REST v3, fetches per-partition lag, and aggregates total lag per topic across all groups.
4. Optionally establishes a direct Kafka consumer connection to seek to the last offset of every partition and read back the message timestamp (CHECK_LAST_MESSAGE=true).
5. Applies the decision logic above and writes orphaned_topics.csv.

Orphan examples:

# TRUE orphan — no bytes, no consumer interest
topic: legacy-migration-2022
  received_bytes: 0
  sent_bytes:     0
  consumer_lag:   0           → orphan_candidate: TRUE

# FALSE positive avoided — zero bytes but active consumer
topic: config-changelog
  received_bytes: 410         (compact topic, only key updates)
  sent_bytes:     820
  consumer_lag:   5           → orphan_candidate: FALSE  (lag present)

# FALSE positive avoided — zombie group holding lag
topic: old-events
  received_bytes: 0
  sent_bytes:     0
  consumer_lag:   14200       (stale zombie group)
                              → orphan_candidate: FALSE  (lag present)
                              (the zombie group should be investigated via
                               consumer_details.py instead)

Note: After obtaining the orphan candidate list, always validate with the owning application team before proceeding to deletion. Shadow consumers or batch jobs that run infrequently (e.g. monthly) may not appear active within a 7-day window.

How CHECK_LAST_MESSAGE works:

For each topic, per partition:

1. get_watermark_offsets() — a metadata-only call, no message fetch — returns the partition's low (earliest) and high (past-the-end) offsets.
2. The consumer is assigned directly to offset high - 1, i.e. the last message in that partition (no scan from the beginning).
3. One poll() reads exactly that message and its timestamp.

The latest timestamp across all of a topic's partitions becomes its last_message_timestamp. Topics are processed across a pool of worker threads (LAST_MESSAGE_WORKERS), each with its own consumer.

Warning — CHECK_LAST_MESSAGE=true affects future runs: step 3 above reads (consumes) one real message per partition (not per topic — a 6-partition topic triggers 6 reads). The broker records each of these as real sent_bytes traffic, the same metric used for orphan detection. Any find_orphaned_topics.py run within the next OBSERVATION_DAYS will see these reads reflected in received_bytes/sent_bytes totals, which can mask a genuinely orphaned topic. This cannot be avoided: Kafka has no metadata-only way to read a message's timestamp without a real fetch, and the Metrics API's sent_bytes/received_bytes metrics carry no consumer-group dimension to filter this diagnostic traffic back out. Avoid running with CHECK_LAST_MESSAGE=true if you plan to trust sent_bytes/received_bytes results from a run within the same observation window.

General limitation — any message read inflates sent_bytes, not just this script: this isn't specific to CHECK_LAST_MESSAGE. Confluent Cloud's Message Browser (in the web console), kcat, another team's ad-hoc consumer, or any other tool that reads messages from a topic all generate the same real sent_bytes egress at the broker. The Metrics API has no way to distinguish "a diagnostic/manual peek" from "genuine application traffic" — there is no consumer-group or client-id dimension on sent_bytes/received_bytes to filter by. If someone opens the Message Browser on a candidate topic during the observation window, that topic can drop off the orphan list even though no application is actually using it. Treat the orphan candidate list as a starting point for investigation, not a final answer, especially for topics near the threshold.

Optional environment variables:

Variable	Default	Description
CONFLUENT_BOOTSTRAP_SERVER	derived from REST endpoint	Bootstrap server for last-message inspection, e.g. pkc-xxx:9092
ORPHAN_BYTES_THRESHOLD	1048576 (1 MB)	Bytes threshold; topics below this are candidates. Since bytes can never be negative, setting this to 0 disables the bytes check entirely (total < 0 is never true) — 1 flags only topics with truly zero produce/consume, but if you also use CHECK_LAST_MESSAGE=true a threshold of 1 is fragile: a single diagnostic read (~500-2000 bytes, depending on partition count) is enough to push a dead topic's total above 1 and keep it off the orphan list on a later run. Recommended: 10240 (10 KB) — comfortably above that diagnostic noise floor, so a CHECK_LAST_MESSAGE read or a stray Message Browser peek can't by itself save a genuinely dead topic, while still catching truly near-zero-traffic topics
OBSERVATION_DAYS	7	Days of metrics to analyse
CHECK_CONSUMER_LAG	true	Require zero lag for orphan flag
CHECK_LAST_MESSAGE	false	Inspect last message timestamp (requires confluent-kafka). Side effect: this consumes one message per partition, which counts as real sent_bytes traffic for up to OBSERVATION_DAYS afterward — see warning above
LAST_MESSAGE_THRESHOLD_DAYS	30	Flag as old if last message is older than N days
LAST_MESSAGE_WORKERS	10	Parallel worker threads for last-message inspection (CHECK_LAST_MESSAGE=true); each worker uses its own Kafka consumer, since topics are processed in parallel rather than one at a time
EXCLUDE_INTERNAL_TOPICS	true	Skip __*, _confluent*, connect-* topics
METRICS_BACKEND	api	Source for received_bytes/sent_bytes: api (Confluent Metrics API, ~7 day query limit) or prometheus (query a Prometheus instance that scrapes the Metrics API /export endpoint — e.g. the ccloud-prometheus-grafana stack — for history beyond 7 days, limited only by that Prometheus instance's retention). When set to prometheus, CONFLUENT_CLOUD_API_KEY/SECRET are not required. Queries are scoped to CONFLUENT_CLUSTER_ID via a kafka_id PromQL label matcher, so the Prometheus instance must expose that label (true by default for the ccloud-prometheus-grafana stack) — if your setup relabels it to something else, the filter will silently match no series
PROMETHEUS_URL	http://localhost:9090	Base URL of the Prometheus HTTP API. Only used when METRICS_BACKEND=prometheus
INCLUDE_COST_ESTIMATE	false	When true, fetches cleanup.policy, retention_ms, and retained_bytes per topic and adds cost columns to the CSV output
STORAGE_COST_PER_GB_MONTH	0.08	Storage cost rate (USD per GB per month) used when INCLUDE_COST_ESTIMATE=true. Default is the Confluent Cloud general-purpose rate; override for specialty clusters or negotiated pricing. To find your actual rate: Confluent Cloud Console → Billing & payment → Usage → find the KafkaStorage line item's $/GB-hour unit price, then convert to monthly with rate_per_hour * 24 * 30.44 (e.g. $0.00012603/GB-hour → ≈ 0.0921).
OUTPUT_FILE	orphaned_topics.csv	Output CSV path. All log output is also written to a .log file with the same base name and timestamp
LOG_LEVEL	INFO	Set to DEBUG for per-topic and per-page progress logs (topic discovery, config/lag fetch progress, metrics API pagination) — useful for diagnosing slow runs

Usage:

python find_orphaned_topics.py

Output columns:

Column	Description
topic_name	Kafka topic name
partition_count	Number of partitions
received_bytes	Bytes produced to topic in window
sent_bytes	Bytes consumed from topic in window
total_bytes	Sum of received + sent
consumer_lag	Total lag across all consumer groups
cleanup_policy	(only when INCLUDE_COST_ESTIMATE=true) Topic cleanup policy (delete, compact, or compact,delete)
retention_ms	(only when INCLUDE_COST_ESTIMATE=true) Configured retention in milliseconds (-1 = infinite)
retained_bytes	(only when INCLUDE_COST_ESTIMATE=true) Bytes currently stored for the topic
estimated_monthly_storage_cost_usd	(only when INCLUDE_COST_ESTIMATE=true) Estimated monthly storage cost at $0.08/GB-month. Most meaningful for compact or infinite-retention topics — delete topics will self-purge as retention expires.
last_message_timestamp	ISO timestamp of last message (if CHECK_LAST_MESSAGE=true)
last_message_age_days	Days since last message
last_message_old	True if older than LAST_MESSAGE_THRESHOLD_DAYS
orphan_candidate	True if topic meets orphan criteria

---

tag_topics_for_deletion.py

Tags topics in the Confluent Cloud Stream Catalog to mark them for decommissioning. Implements step 2 of the recommended topic decommissioning lifecycle.

Per topic, the script applies:

• Tag: ScheduledForDeletion — classification label visible in the Stream Catalog.

• Business metadata (DecommissionInfo type) — structured fields visible under the Business Metadata panel of each topic in the Confluent Cloud UI:

  Field	Required	Description
  decommissionDate	Yes	Planned deletion date (YYYY-MM-DD)
  teamOwner	No	Owning team name
  contactEmail	No	Email address other teams can use to raise questions before the topic is removed

The script can also remove tags and business metadata (to unblock a topic that was incorrectly scheduled) and list current tags and metadata on topics.

Optional environment variables:

Variable	Default	Description
DECOMMISSION_DATE	today + 30 days	Planned deletion date YYYY-MM-DD
TEAM_OWNER	(empty)	Owning team name; set to populate teamOwner business metadata field
CONTACT_EMAIL	(empty)	Contact email; set to populate contactEmail business metadata field
DRY_RUN	false	Print actions without making any API calls

Usage:

# Tag all orphan candidates from find_orphaned_topics.py output
python tag_topics_for_deletion.py --action apply --input-file orphaned_topics.csv

# Tag a specific list of topics
python tag_topics_for_deletion.py --action apply --topics "topic-a,topic-b"

# Preview what would be tagged (no changes made)
DRY_RUN=true python tag_topics_for_deletion.py --action apply --topics "topic-a"

# Remove tags from a topic (unblock / cancel deletion)
python tag_topics_for_deletion.py --action remove --topics "topic-a"

# List current tags on topics
python tag_topics_for_deletion.py --action list --input-file topics.txt

Input file formats:

• Plain text — one topic name per line (lines starting with # are ignored).
• CSV — the output format of find_orphaned_topics.py; only rows where orphan_candidate is true are processed.

---

tag_schemas_for_deletion.py

Tags schema subject versions in the Confluent Cloud Stream Catalog to mark them for decommissioning. This mirrors the topic-tagging workflow, but targets sr_subject_version entities so it lines up with the find_unused_schemas.py detect and delete flow.

Per subject version, the script applies:

• Tag: SchemaScheduledForDeletion — classification label visible in the Stream Catalog.

• Business metadata (SchemaDecommissionInfo type) — structured fields visible under the schema version entity:

  Field	Required	Description
  decommissionDate	Yes	Planned deletion date (YYYY-MM-DD)
  subject	Yes	Schema Registry subject name
  version	Yes	Subject version scheduled for deletion
  schemaId	No	Global schema ID from Schema Registry
  candidateReason	No	Reason the version was flagged by find_unused_schemas.py
  teamOwner	No	Owning team name
  contactEmail	No	Email address other teams can use to raise questions before deletion

The script can also remove tags and business metadata and list current tags and metadata on schema subject versions.

Optional environment variables:

Variable	Default	Description
DECOMMISSION_DATE	today + 30 days	Planned deletion date YYYY-MM-DD
TEAM_OWNER	(empty)	Owning team name; set to populate teamOwner business metadata field
CONTACT_EMAIL	(empty)	Contact email; set to populate contactEmail business metadata field
DRY_RUN	false	Print actions without making any API calls

Usage:

# Tag all schema versions listed in the detect CSV
python tag_schemas_for_deletion.py --action apply --input-file unused_schemas.csv

# Tag specific schema subject versions directly
python tag_schemas_for_deletion.py --action apply --schemas "orders-value:3,orders-key:2"

# Preview what would be tagged (no changes made)
DRY_RUN=true python tag_schemas_for_deletion.py --action apply --input-file unused_schemas.csv

# Remove tags from a schema subject version
python tag_schemas_for_deletion.py --action remove --schemas "orders-value:3"

# List current tags on schema subject versions
python tag_schemas_for_deletion.py --action list --input-file unused_schemas.csv

Input file formats:

• Plain text — one subject:version entry per line (lines starting with # are ignored).
• CSV — the output format of find_unused_schemas.py; if a delete_candidate column is present, only rows where it is true are processed.

---

apply_deny_acls_to_topics.py

Creates explicit topic-scoped Kafka ACL entries with permission=DENY for the READ and WRITE operations. This is useful when you want to actively block a specific principal from interacting with orphaned topics before full permission cleanup or deletion.

Important: In Kafka ACL evaluation, DENY takes precedence over ALLOW. Always test with DRY_RUN=true first. To deny every principal, use the explicit --all-principals switch, which creates ACLs for User:*.

Required service account role

Mode	Role	Scope
create topic ACLs	CloudClusterAdmin	Target cluster

Optional environment variables

Variable	Default	Description
DRY_RUN	false	Print planned ACLs without creating them

Usage

# Preview DENY READ/WRITE ACLs for all orphan candidates in the CSV
DRY_RUN=true python apply_deny_acls_to_topics.py \
  --input-file orphaned_topics.csv \
  --principal "User:sa-123456"

# Apply DENY READ/WRITE ACLs for a specific principal
python apply_deny_acls_to_topics.py \
  --input-file orphaned_topics.csv \
  --principal "User:sa-123456"

# Apply DENY ACLs for multiple principals and skip the prompt
python apply_deny_acls_to_topics.py \
  --topics "topic-a,topic-b" \
  --principal "User:sa-123456" \
  --principal "User:sa-999999" \
  --yes

# Apply DENY ACLs for all principals (Kafka wildcard principal User:*)
python apply_deny_acls_to_topics.py \
  --input-file orphaned_topics.csv \
  --all-principals

Input file formats:

• Plain text — one topic name per line (lines starting with # are ignored).
• CSV — the output format of find_orphaned_topics.py; only rows where orphan_candidate is true are processed.

---

find_unused_schemas.py

Detects Schema Registry schema versions that could be deleted using a metadata-only heuristic:

1. Identify eligible TopicNameStrategy subjects (*-key, *-value)
2. Derive the Kafka topic name from each subject
3. Retrieve all registered versions under each subject
4. Check whether each version is referenced by another schema
5. Optionally check whether the derived Kafka topic currently exists via Kafka REST
6. Optionally read a topic list / orphaned-topics CSV for topics already deleted or confirmed for deletion
7. Preserve the latest version per subject unless the topic is marked deleted or no corresponding topic exists
8. Flag unreferenced candidate versions and write only those candidate rows to CSV

The script also supports --action delete, which reads the generated CSV and performs the required soft delete followed by hard delete for each selected subject/version pair.

Important: This script no longer scans Kafka payloads. It uses Schema Registry metadata, plus an optional deleted-topics input file, so it is fast but heuristic-based. The detect CSV contains only rows already flagged as delete candidates, so review it before using delete mode.

If CONFLUENT_REST_ENDPOINT, CONFLUENT_CLUSTER_ID, CONFLUENT_API_KEY, and CONFLUENT_API_SECRET are set, the script also checks whether the derived Kafka topic exists. When a topic does not exist, the corresponding schema versions can be flagged even if they are the latest version under the subject.

Schema contexts: GET /subjects returns subjects across all schema contexts, not just the default context, so --all already covers every context without extra configuration. However, subjects in different contexts can derive the same bare Kafka topic name (e.g. :.DEV:orders-value and :.TEST:orders-value both derive topic orders) — these are typically unrelated schemas, not versions of the same history. The context output column disambiguates these; do not assume rows sharing a topic_name are related unless their context also matches. Topic-existence checks are per bare topic name (Kafka has no notion of contexts), so subjects from different contexts that truly do point at the same physical topic will correctly get the same result.

Required service account roles

Mode	Role	Scope
detect	Schema Registry read access	Environment
delete	Schema Registry delete permission	Environment

Optional environment variables

Variable	Default	Description
OUTPUT_FILE	unused_schemas_<cluster>_<timestamp>.csv	Output CSV path for detect mode
DRY_RUN	false	Print planned changes without deleting schema versions
INCLUDE_QUOTA_ESTIMATE	false	Report delete-candidate schema versions against the Stream Governance package's schema quota (requires CONFLUENT_CLOUD_API_KEY/CONFLUENT_CLOUD_API_SECRET/CONFLUENT_ENVIRONMENT_ID)
MAX_WORKERS	10	Concurrent subjects evaluated in detect mode

Performance on large Schema Registries

Detect mode evaluates each subject with several sequential Schema Registry calls (list versions, fetch each version's details, check references), so a registry with hundreds of subjects can take a while to fully scan. To keep this fast and show progress on long runs:

• Concurrency — subjects are evaluated in parallel using a thread pool sized by MAX_WORKERS (default 10). Since these are network-bound calls, raising MAX_WORKERS can meaningfully cut runtime on large registries; lower it if you see 429/throttling responses from Schema Registry.
• Shared topic-existence cache — topic_exists() results are cached and shared (behind a lock) across subjects, so a topic with both -key and -value subjects is only checked once even when evaluated concurrently.
• Progress logging — logs a running count (... X/N subjects evaluated ...) roughly every 5% of subjects processed, so a long detect run doesn't look frozen.

Usage

# Detect candidates for all eligible TopicNameStrategy subjects
python find_unused_schemas.py --action detect --all

# Detect candidates for one or more specific subjects
python find_unused_schemas.py --action detect --subject orders-value --subject orders-key

# Detect candidates and treat topics from the orphan/deletion CSV as deleted,
# which allows latest subject versions for those topics to be flagged too
python find_unused_schemas.py \
  --action detect \
  --all \
  --deleted-topics-file orphaned_topics_lkc-9ny7qv_delete.csv

# If Kafka REST env vars are configured, subjects whose derived topics no longer
# exist are also eligible, even for latest versions
python find_unused_schemas.py --action detect --all

# Report how many delete candidates count against the Stream Governance schema quota
INCLUDE_QUOTA_ESTIMATE=true python find_unused_schemas.py --action detect --all

# Preview hard delete actions for rows flagged as delete_candidate=true
DRY_RUN=true python find_unused_schemas.py \
  --action delete \
  --input-file unused_schemas.csv

# Hard delete all candidate rows from the CSV
python find_unused_schemas.py \
  --action delete \
  --input-file unused_schemas.csv

Detect output columns:

Column	Description
subject	Schema Registry subject
context	Schema context the subject belongs to (. for the default context)
topic_name	Derived Kafka topic name
subject_kind	key or value
version	Subject version
latest_version	Latest version currently registered under the subject
is_latest_version	True when this row is the latest version under the subject
topic_marked_deleted	True when the derived topic name appears in --deleted-topics-file
topic_exists	True / False when Kafka REST topic existence checks are enabled; blank otherwise
schema_id	Global schema ID
schema_type	AVRO, PROTOBUF, or JSON
referenced_by_count	Number of schema IDs that reference this version
delete_candidate	True when the schema version is unreferenced and either non-latest, belongs to a topic marked deleted, or has no corresponding topic
candidate_reason	Why the row was or was not marked as a delete candidate

Delete input format:

• CSV — the output format of find_unused_schemas.py; by default only rows where delete_candidate is true are processed.

---

find_unused_schema_versions.py

Detects Schema Registry schema versions that could be deleted using a retention-age heuristic, for TopicNameStrategy subjects. This is a separate, independent heuristic from find_unused_schemas.py — it does not check referencedby or cross-schema references, so a version could be flagged here even if another schema still references it.

1. Identify eligible TopicNameStrategy subjects (*-key, *-value)
2. Derive the Kafka topic name from each subject
3. Retrieve all registered versions under each subject, including each version's registration timestamp (ts, epoch milliseconds)
4. Look up the topic's retention.ms via the Kafka REST v3 API
5. Flag a non-latest version as a delete candidate when now - ts > retention.ms, since any messages produced under that schema would have already aged out of the topic
6. Always preserve the latest version of a subject, and skip topics with infinite retention (retention.ms == -1)
7. Write deletion candidates to a CSV report

The script also supports --action delete, which reads the generated CSV and performs the required soft delete followed by hard delete for each selected subject/version pair — same delete flow as find_unused_schemas.py.

Important: This heuristic assumes schema versions are registered by a producer before being used, and that a version is "superseded" once a later version exists. It does not inspect Kafka topic contents. Retention lookup requires the Kafka REST v3 API, which is a required dependency here (unlike find_unused_schemas.py, where it's optional).

Schema Registry response quirk: the ts (registration timestamp) field is only returned by Schema Registry when the request sends the Confluent-Accept-Unknown-Properties: true header — without it, ts is silently omitted and every version falls into candidate_reason= no_registration_timestamp, never becoming a candidate. The script always sends this header; if you see this reason code, the topic likely has no ts for another cause (very old schema, or a registry that predates the field).

Single-cluster limitation: Schema Registry is shared across every Kafka cluster in an environment, but this script's Kafka REST credentials (CONFLUENT_REST_ENDPOINT / CONFLUENT_CLUSTER_ID / CONFLUENT_API_KEY/ SECRET) point at exactly one cluster. Any subject whose topic lives on a different cluster in the same environment cannot be resolved — it is reported as candidate_reason=topic_not_found_on_configured_cluster and is not treated as a delete candidate. To cover an environment with multiple clusters, run this script once per cluster with that cluster's REST endpoint/credentials.

Schema contexts: GET /subjects returns subjects across all schema contexts, not just the default context, so --all already covers every context without extra configuration. However, subjects in different contexts can derive the same bare Kafka topic name (e.g. :.DEV:orders-value and :.TEST:orders-value both derive topic orders) — these are typically unrelated schemas, not versions of the same history. The context output column disambiguates these; do not assume rows sharing a topic_name are related unless their context also matches. Retention lookups themselves are per bare topic name (Kafka has no notion of contexts), so subjects from different contexts that truly do point at the same physical topic will correctly get the same retention.ms.

Required service account roles

Mode	Role	Scope
detect	Schema Registry read access + DeveloperRead on the target cluster (Topic *, for retention.ms lookups)	Environment + target cluster
delete	Schema Registry delete permission	Environment

Required environment variables

Detect mode requires all of CONFLUENT_SCHEMA_REGISTRY_URL, CONFLUENT_SCHEMA_REGISTRY_API_KEY, CONFLUENT_SCHEMA_REGISTRY_API_SECRET, CONFLUENT_REST_ENDPOINT, CONFLUENT_CLUSTER_ID, CONFLUENT_API_KEY, and CONFLUENT_API_SECRET (all defined in Common variables above). Delete mode requires only the three Schema Registry variables.

Optional environment variables

Variable	Default	Description
OUTPUT_FILE	unused_schema_versions_<cluster>_<timestamp>.csv	Output CSV path for detect mode
LOG_FILE	find_unused_schema_versions_<cluster>_<timestamp>.log	Log file path (logs are always written to both stdout and this file)
DRY_RUN	false	Print planned changes without deleting schema versions
MAX_WORKERS	10	Concurrent subjects evaluated in detect mode
NOW_EPOCH_MS	current time	Override "now" (epoch ms) used for age comparisons — mainly for testing

Usage

# Detect candidates for all eligible TopicNameStrategy subjects
python find_unused_schema_versions.py --action detect --all

# Detect candidates for one or more specific subjects
python find_unused_schema_versions.py --action detect --subject orders-value --subject orders-key

# Preview hard delete actions for rows flagged as delete_candidate=true
DRY_RUN=true python find_unused_schema_versions.py \
  --action delete \
  --input-file unused_schema_versions.csv

# Hard delete all candidate rows from the CSV
python find_unused_schema_versions.py \
  --action delete \
  --input-file unused_schema_versions.csv

Detect output columns:

Column	Description
subject	Schema Registry subject
context	Schema context the subject belongs to (. for the default context)
topic_name	Derived Kafka topic name
subject_kind	key or value
version	Subject version
latest_version	Latest version currently registered under the subject
is_latest_version	True when this row is the latest version under the subject
schema_id	Global schema ID
schema_type	AVRO, PROTOBUF, or JSON
schema_deleted	True if the version is already soft-deleted in Schema Registry
registration_ts_ms	Version's registration timestamp (epoch ms) from Schema Registry's ts field
registration_age_days	Age of the registration timestamp in days, relative to now (or NOW_EPOCH_MS)
topic_retention_ms	The derived topic's retention.ms config value
delete_candidate	True when the version is non-latest and its registration is older than the topic's retention window
candidate_reason	Why the row was or was not marked as a delete candidate — see below

candidate_reason values:

Reason	Meaning
registration_older_than_retention	Delete candidate — non-latest version registered longer ago than the topic's retention window
within_retention_window	Not a candidate — version is newer than the retention window
latest_version_preserved	Not a candidate — always preserved regardless of age
infinite_retention	Not a candidate — topic has retention.ms == -1, so nothing ever ages out
topic_not_found_on_configured_cluster	Not a candidate — topic isn't visible via the configured cluster's REST API (see single-cluster limitation above)
no_registration_timestamp	Not a candidate — Schema Registry returned no ts for this version
schema_version_already_deleted	Not a candidate — version is already soft-deleted

Delete input format:

• CSV — the output format of find_unused_schema_versions.py; by default only rows where delete_candidate is true are processed.

---

find_overlapping_schema_ids.py

Detects Schema Registry global schema IDs that are referenced by more than one subject/version across all schema contexts. This is a read-only, detect-only audit script — there is nothing to delete, since a schema ID cannot be selectively unregistered from just one of several subjects that reference it.

Schema Registry's global schema ID is not always partitioned per context — the same numeric ID can be assigned to subjects in different contexts, and those subjects are not guaranteed to hold the same schema content. This script surfaces every such overlap so you can distinguish:

• Expected reuse — identical schema content shared across subjects (e.g. a -key and -value subject registering the same record, or the same schema intentionally reused across contexts).
• A true collision — the same schema ID maps to genuinely different content in different subjects/contexts. Worth investigating: any tooling that assumes a schema ID uniquely identifies one schema will behave incorrectly here.

1. List every subject across all schema contexts (GET /subjects already returns subjects from every context, not just the default one)
2. For each subject, retrieve every registered version's global schema ID and raw schema content
3. Group all subject/version references by global schema ID
4. Flag any schema ID referenced by more than one subject/version
5. For each flagged ID, compare the raw schema string (and schemaType) across all references: identical content is content_matches_other_ references=true (expected reuse); any difference is false (a true collision)
6. Write every overlapping ID's references to a CSV report

Comparison method: content is compared as an exact string match of the raw schema field (plus schemaType) returned by Schema Registry — not a structurally normalized comparison. Cosmetic differences (field ordering, whitespace) in an otherwise-equivalent schema will be reported as content_matches_other_references=false.

GET /schemas/ids/{id} cannot be used to discover this: that endpoint only returns a single subject/version for a given ID (the first one Schema Registry associates it with), even when multiple subjects reference the same ID. Discovering overlaps requires walking every subject and version individually, which is what this script does.

Required service account roles

Mode	Role	Scope
detect	Schema Registry read access	Environment

Required environment variables

Variable	Description
CONFLUENT_SCHEMA_REGISTRY_URL	Schema Registry base URL
CONFLUENT_SCHEMA_REGISTRY_API_KEY	Schema Registry API key
CONFLUENT_SCHEMA_REGISTRY_API_SECRET	Schema Registry API secret

Optional environment variables

Variable	Default	Description
OUTPUT_FILE	overlapping_schema_ids_<timestamp>.csv	Output CSV path
LOG_FILE	find_overlapping_schema_ids_<timestamp>.log	Log file path (logs are always written to both stdout and this file)
MAX_WORKERS	10	Concurrent subjects evaluated

Usage

python find_overlapping_schema_ids.py

Output columns:

Column	Description
schema_id	Global schema ID shared by more than one subject/version
subject	Schema Registry subject referencing this ID
context	Schema context the subject belongs to (. for the default context)
version	Subject version referencing this ID
subject_kind	key or value for TopicNameStrategy subjects; blank for other naming strategies
schema_type	AVRO, PROTOBUF, or JSON
schema_deleted	True if this version is already soft-deleted
content_matches_other_references	True if every reference to this schema ID has identical content (expected reuse); False if content differs across references (a true collision)
schema_preview	First 200 characters of the schema definition

Every row for a given schema_id shares the same content_matches_other_references value — it reflects whether all references to that ID agree, not a pairwise comparison.

---

remove_permissions_for_topics_to_delete.py

Revokes Kafka ACLs and/or Confluent Cloud RBAC role bindings for topics that have been marked for decommissioning. This is the access-lockout step in the decommission workflow — run it on or close to the decommissionDate recorded by tag_topics_for_deletion.py, after all consuming teams have confirmed they no longer need the topic.

This operation is irreversible. Revoking access cuts off producers and consumers immediately. Always run with --dry-run first to review what will be removed.

What is removed

Mode	What is revoked	API used
acls	Explicit, topic-specific Kafka ACL entries (LITERAL pattern type)	Kafka REST v3 /acls
rbac	Confluent Cloud IAM role bindings scoped to the specific topic CRN	Cloud IAM v2 /role-bindings
both (default)	Both of the above	Both APIs

Wildcard ACLs are not removed. ACL entries with resource_name=* (which happen to cover the topic) are not touched — removing them would affect every other topic on the cluster.

Required service account roles

Mode	Role	Scope
--revoke acls	CloudClusterAdmin	Target cluster
--revoke rbac	OrganizationAdmin or EnvironmentAdmin	Organisation or Environment

Optional environment variables

Variable	Default	Description
DRY_RUN	false	Print planned actions without making any API calls
CONFLUENT_ORG_ID	(auto-fetched)	Confluent organisation ID. If not set, the script calls GET /org/v2/organizations once and caches the result. Set this explicitly if the Cloud API key's service account lacks permission to list organisations.

Usage

# Preview what would be removed — no changes made
DRY_RUN=true python remove_permissions_for_topics_to_delete.py \
  --revoke both \
  --input-file orphaned_topics.csv

# Revoke only Kafka ACLs for a specific topic (interactive confirmation)
python remove_permissions_for_topics_to_delete.py \
  --revoke acls \
  --topics "topic-a"

# Revoke only RBAC role bindings, skipping confirmation (CI/CD use)
python remove_permissions_for_topics_to_delete.py \
  --revoke rbac \
  --topics "topic-a,topic-b" \
  --yes

# Revoke both ACLs and RBAC for all orphan candidates from find_orphaned_topics.py
python remove_permissions_for_topics_to_delete.py \
  --revoke both \
  --input-file orphaned_topics.csv

Input file formats:

• Plain text — one topic name per line (lines starting with # are ignored).
• CSV — the output format of find_orphaned_topics.py; only rows where orphan_candidate is true are processed.

---

delete_topics.py

Permanently deletes Kafka topics from the cluster via the Kafka REST v3 API.

This is the final, irreversible step in the topic decommissioning lifecycle and should only be run after topics have been identified, tagged, and had their permissions revoked.

WARNING: Topic deletion cannot be undone. All messages stored in the topic are permanently lost. Confluent Cloud does not provide a way to recover deleted topics or their data. Always run with DRY_RUN=true first to confirm the topic list before proceeding.

Required service account role: CloudClusterAdmin on the target cluster (same as remove_permissions_for_topics_to_delete.py ACL mode — a single service account covers both steps).

Optional environment variables:

Variable	Default	Description
DRY_RUN	false	Print what would be deleted without making any API calls

Usage:

# Preview — print the topics that would be deleted, no changes made
DRY_RUN=true python delete_topics.py --input-file orphaned_topics.csv

# Delete from CSV (only rows where orphan_candidate=true are processed)
# Prompts for confirmation before proceeding
python delete_topics.py --input-file orphaned_topics.csv

# Delete a specific list of topics
python delete_topics.py --topics "topic-a,topic-b,topic-c"

# Skip confirmation prompt (CI/CD pipelines)
python delete_topics.py --input-file orphaned_topics.csv --yes

# Delete from a plain-text list
python delete_topics.py --input-file topics_to_delete.txt

Confirmation behaviour:

In interactive mode (no --yes flag, DRY_RUN=false) the script prints the full list of topics to be deleted, displays a warning, and requires you to type yes to proceed. Any other input aborts without making changes. In dry-run mode no prompt is shown.

Input file formats:

• Plain text — one topic name per line (lines starting with # are ignored).
• CSV — the output format of find_orphaned_topics.py; only rows where orphan_candidate is true are processed.

---

find_idle_flink_statements.py

Identifies Apache Flink SQL statements in a Confluent Cloud environment that are consuming CFUs without processing data — idle or stuck statements that are candidates for stopping.

Detection logic

Confluent Cloud exposes per-statement metrics via the Metrics API, scoped to a compute pool resource. The script aggregates record throughput over a configurable observation window and classifies each RUNNING statement:

For each RUNNING statement in the environment:

  records_in  = SUM(io.confluent.flink/num_records_in)  over observation window
  records_out = SUM(io.confluent.flink/num_records_out) over observation window
  pending     = MAX(io.confluent.flink/pending_records) over observation window

  if records_in ≤ threshold AND records_out ≤ threshold AND pending ≤ 0:
      → IDLE   (burning CFUs with zero throughput, no backlog)

  if records_in ≤ threshold AND records_out ≤ threshold AND pending > 0:
      → STUCK  (data exists in input topic(s) but statement is not consuming it;
                investigate for errors, watermark issues, or resource starvation)

  if records_in > threshold AND records_out ≤ threshold:
      → SINK_ONLY  (receiving records but emitting none — valid for some INSERT INTO
                    patterns; review manually before stopping)

  otherwise:
      → ACTIVE  (data is flowing; no action needed)

Statements with a phase other than RUNNING (e.g. COMPLETED, FAILED, STOPPED) are classified as STOPPED and excluded from the output by default (they are not burning CFUs). Set INCLUDE_STOPPED=true to include them.

Key differences vs find_orphaned_topics.py:

Aspect	Kafka topics	Flink statements
Resource dimension in Metrics API	resource.kafka.id	resource.compute_pool.id
Record metrics	Byte-based (received_bytes, sent_bytes)	Count-based (num_records_in, num_records_out)
Metric type	Counters → SUM	Counters (records) + Gauges (pending, CFUs) → SUM / MAX
Observation window default	7 days	24 hours (Flink metrics retention is shorter)
Cost signal	retained_bytes (storage)	cfu_minutes_consumed (compute)
Phase source	Kafka topic existence	Flink REST API status.phase

Authentication & API endpoints:

This script uses two separate APIs, each with its own credentials:

API	Base URL	Credentials
Flink SQL REST API (list statements)	https://flink.{region}.{cloud}.confluent.cloud	CONFLUENT_FLINK_API_KEY / CONFLUENT_FLINK_API_SECRET
Confluent Cloud Metrics API (throughput metrics)	https://api.telemetry.confluent.cloud	CONFLUENT_CLOUD_API_KEY / CONFLUENT_CLOUD_API_SECRET

Important: The Flink SQL REST API uses a regional base URL and requires a Flink API key scoped to the environment/region/cloud provider — this is different from the general Cloud API key used for the Metrics API.

Create a Flink API key with:

confluent api-key create --resource flink \
  --environment <ENVIRONMENT_ID> \
  --cloud <CLOUD_PROVIDER> \
  --region <REGION>

Required service account roles:

API	Role	Scope
Confluent Cloud Metrics API	MetricsViewer	Organisation level
Flink SQL REST API	FlinkDeveloper	Environment level

The Cloud API key (for Metrics) and the Flink API key (for listing statements) typically belong to the same service account, but the keys themselves are separate.

# MetricsViewer at org level (for the Cloud API key)
confluent iam rbac role-binding create \
  --role MetricsViewer \
  --principal User:<SERVICE_ACCOUNT_ID>

# FlinkDeveloper at environment level (for the Flink API key)
confluent iam rbac role-binding create \
  --role FlinkDeveloper \
  --principal User:<SERVICE_ACCOUNT_ID> \
  --environment <ENVIRONMENT_ID>

Required environment variables:

Variable	Description
CONFLUENT_CLOUD_API_KEY	Cloud API key (Metrics API)
CONFLUENT_CLOUD_API_SECRET	Cloud API secret
CONFLUENT_ENVIRONMENT_ID	Environment ID, e.g. env-xxxxx
CONFLUENT_ORG_ID	Organisation ID, e.g. b0b421724-xxxx-…
CONFLUENT_FLINK_API_KEY	Flink API key (scoped to environment/region/cloud)
CONFLUENT_FLINK_API_SECRET	Flink API secret
CONFLUENT_FLINK_REGION	Cloud region, e.g. us-east-1, europe-west1
CONFLUENT_FLINK_CLOUD	Cloud provider: aws, gcp, or azure

Optional environment variables:

Variable	Default	Description
CONFLUENT_FLINK_COMPUTE_POOL_ID	(all pools)	Filter analysis to a single compute pool, e.g. lfcp-xxxxx. When unset, all pools found in the statement list are queried.
OBSERVATION_HOURS	24	Hours of metrics to analyse
IDLE_RECORDS_THRESHOLD	0	records_in and records_out at or below this value is considered idle. Default 0 means strictly zero throughput.
INCLUDE_STOPPED	false	When true, includes non-RUNNING statements (COMPLETED, FAILED, STOPPED) in the output
OUTPUT_FILE	idle_flink_statements_{env_id}_{ts}.csv	Output CSV path

Usage:

# Set required environment variables
export CONFLUENT_CLOUD_API_KEY="your-cloud-api-key"
export CONFLUENT_CLOUD_API_SECRET="your-cloud-api-secret"
export CONFLUENT_ENVIRONMENT_ID="env-xxxxx"
export CONFLUENT_ORG_ID="your-org-id"
export CONFLUENT_FLINK_API_KEY="your-flink-api-key"
export CONFLUENT_FLINK_API_SECRET="your-flink-api-secret"
export CONFLUENT_FLINK_REGION="us-east-1"
export CONFLUENT_FLINK_CLOUD="aws"

# Analyse all Flink statements in the environment (all compute pools)
python find_idle_flink_statements.py

# Analyse a specific compute pool only
CONFLUENT_FLINK_COMPUTE_POOL_ID=lfcp-xxxxx python find_idle_flink_statements.py

# Widen the observation window to 48 hours
OBSERVATION_HOURS=48 python find_idle_flink_statements.py

# Include stopped / failed statements in the output
INCLUDE_STOPPED=true python find_idle_flink_statements.py

Output columns:

Column	Description
statement_name	Flink statement name (human-readable, e.g. my-aggregation)
compute_pool_id	Compute pool the statement runs on
phase	Current phase from the Flink REST API (RUNNING, STOPPED, FAILED, etc.)
classification	IDLE, STUCK, ACTIVE, SINK_ONLY, or STOPPED
records_in	Records received by the statement over the observation window (SUM)
records_out	Records emitted by the statement over the observation window (SUM)
pending_records	Maximum pending-record backlog seen in the window (MAX gauge)
cfu_minutes_consumed	CFU-minutes consumed in the observation window (SUM) — primary cost indicator
max_cfus	Peak CFU allocation observed in the window (MAX gauge)
sql_preview	First 200 characters of the SQL statement
created_at	Statement creation timestamp
idle_candidate	True when classification is IDLE or STUCK

---

delete_flink_statements.py

Stops and deletes Apache Flink SQL statements in a Confluent Cloud environment. This is the final step after identifying idle or unwanted statements with find_idle_flink_statements.py.

Flink statement deletion is a two-phase operation: Phase 1 STOP — sets the statement phase to STOPPED (drains in-flight work). Phase 2 DELETE — permanently removes the statement definition and all state. Use --stop-only (or STOP_ONLY=true) to perform only the stop phase.

WARNING: Deletion is irreversible — the statement definition and all associated state are permanently lost. Always run with --dry-run first.

Required environment variables

Variable	Description
CONFLUENT_ORG_ID	Organisation ID, e.g. b0b421724-xxxx-…
CONFLUENT_ENVIRONMENT_ID	Environment ID, e.g. env-xxxxx
CONFLUENT_FLINK_API_KEY	Flink API key (scoped to environment/region/cloud)
CONFLUENT_FLINK_API_SECRET	Flink API secret
CONFLUENT_FLINK_REGION	Cloud region for Flink, e.g. us-east-1, europe-west1
CONFLUENT_FLINK_CLOUD	Cloud provider for Flink: aws, gcp, or azure

Required service account role

Same as find_idle_flink_statements.py — FlinkDeveloper at the environment level for the Flink API key's service account.

Optional environment variables

Variable	Default	Description
DRY_RUN	false	Print planned actions without making any API calls
STOP_ONLY	false	Stop statements but do not delete them

Usage

# Preview stop+delete for all IDLE/STUCK statements from the detect CSV
DRY_RUN=true python delete_flink_statements.py \
  --input-file idle_flink_statements.csv

# Stop and delete all IDLE/STUCK statements from the detect CSV
python delete_flink_statements.py --input-file idle_flink_statements.csv

# Only stop (do not delete) statements, skipping the confirmation prompt
python delete_flink_statements.py \
  --input-file idle_flink_statements.csv \
  --stop-only --yes

# Target specific statements directly, including only STUCK ones from a CSV
python delete_flink_statements.py \
  --input-file idle_flink_statements.csv \
  --classification STUCK

# Delete an explicit list of statement names (bypasses classification filter)
python delete_flink_statements.py --statements "my-old-job,another-job" --yes

Input file formats:

• Plain text — one statement name per line (lines starting with # are ignored).
• CSV — the output format of find_idle_flink_statements.py; only rows whose classification matches --classification (default IDLE,STUCK; use ALL for every classification) are processed.

---

idle_connectors.py

Detects Kafka Connect connectors that are not processing data by comparing connector offsets between runs.

Detection logic

Connector status (RUNNING, PAUSED, etc.) only tells you whether the connector process is alive — it does not tell you whether it is actually moving data. A connector can be RUNNING while ingesting zero records indefinitely.

The reliable signal is the connector's offsets: these represent the last processed position in the source system (file byte offset, database CDC LSN, S3 object key, etc.). If the offsets do not advance between runs, no new data has been processed.

For each connector:

  current_offsets = GET /connectors/{name}/offsets

  if no previous state exists:
      classify as BASELINE  (first run, establish a snapshot)

  elif connector STATUS == FAILED:
      classify as FAILED

  elif current_offsets is None (API returned no offsets):
      classify as NO_OFFSETS  (connector never committed — likely never ran)

  elif current_offsets != previous_offsets:
      classify as ACTIVE      (offsets advanced — data is moving)

  else:
      idle_run_count += 1
      if idle_run_count >= IDLE_THRESHOLD_RUNS:
          classify as IDLE    (offsets static for N consecutive runs)
      else:
          classify as ACTIVE  (not yet at threshold)

  save current_offsets to state file for next comparison

Why this is more reliable than STATUS=RUNNING:

• A source connector polling an empty table is RUNNING but has zero throughput.
• A sink connector writing to a paused external system may appear RUNNING but backs up silently.
• Offset comparison detects both cases regardless of the reported status.

Limitations:

• Confluent Cloud only supports offset management for a specific list of fully-managed source connectors (plus all sink connectors); custom connectors, earlier source connector versions, and unlisted connector classes (e.g. Datagen Source) always return 403 on the offsets endpoint, regardless of role. These are classified as NO_OFFSETS, and idle_connectors.py logs this at info level for known-unsupported classes (see OFFSETS_UNSUPPORTED_CONNECTOR_CLASSES) rather than warning about a permissions issue.
• Connectors that process data in large infrequent batches may appear IDLE between batch windows. Use IDLE_THRESHOLD_RUNS to require multiple consecutive idle observations before flagging (e.g. run daily and set IDLE_THRESHOLD_RUNS=5 to require idle for 5 days).

Classifications:

Status	Meaning	Typical action
ACTIVE	Offsets changed since last run — data is moving	No action needed
IDLE	Offsets unchanged for N consecutive runs	Review with team; consider pausing on lower environments
BASELINE	First run — no previous offsets to compare	Re-run tomorrow
FAILED	Connector is in FAILED state	Investigate error logs immediately
NO_OFFSETS	Connector has never committed offsets (likely never ran)	Verify if it was ever needed

Multi-cluster mode (YAML config):

cp config.yaml.example config.yaml
# Edit config.yaml with your environment/cluster IDs
# Set each cluster's API key/secret via environment variables referenced in the YAML
export PROD_API_KEY="..."
export PROD_API_SECRET="..."
CONFIG_FILE=config.yaml python idle_connectors.py

Single-cluster mode (environment variables only):

export CONFLUENT_CLOUD_API_KEY="..."
export CONFLUENT_CLOUD_API_SECRET="..."
export CONFLUENT_ENVIRONMENT_ID="env-xxxxx"
export CONFLUENT_CLUSTER_ID="lkc-xxxxx"
python idle_connectors.py

Optional environment variables:

Variable	Default	Description
CONFIG_FILE	config.yaml	YAML config for multi-cluster mode
STATE_FILE	connector_offsets_state.json	Persisted offset state
OUTPUT_FILE	idle_connectors.csv	Output CSV path
IDLE_THRESHOLD_RUNS	1	Consecutive idle runs before flagging as IDLE
MAX_WORKERS	10	Concurrent connector checks per cluster
RATE_LIMIT_MAX_RETRIES	5	Retries on a 429 rate-limit response before giving up
RATE_LIMIT_BACKOFF_SECONDS	1	Base backoff seconds on a 429; doubles per retry (or uses Retry-After if present)

Tip: Schedule this script to run daily (e.g. via cron or a CI pipeline). After 5+ days of consecutive IDLE classification a connector is a strong decommissioning candidate.

Performance on large clusters

Each connector requires up to 3 API calls (status, config, offsets), so a cluster with hundreds of connectors can take a while to fully check. To keep this fast and give visibility into progress:

• Concurrency — connectors within a cluster are checked in parallel using a thread pool sized by MAX_WORKERS (default 10). Since these calls are network-bound, raising MAX_WORKERS (e.g. to 20) can meaningfully cut runtime on large clusters. If you see 429 Too Many Requests from the Connect API, lower it instead.
• Connector class caching — a connector's connector.class never changes, so after the first run it's cached in STATE_FILE and no longer re-fetched, saving one API call per connector per run.
• Progress logging — the script logs a running count (... X/N connectors checked in 'cluster' ...) roughly every 5% of connectors processed per cluster, so a long run doesn't look frozen.

---

dump_connector_configs.py

Dumps the full configuration of every Kafka Connect connector to individual JSON files, organised by environment and cluster.

This is particularly useful before a cluster migration (e.g. TGW → PrivateLink) so connector configurations can be inspected and recreated on the destination cluster with updated bootstrap servers and API keys.

Sensitive fields (password, api.key, api.secret, etc.) are redacted by default to prevent credential leakage. Set INCLUDE_SENSITIVE=true only in controlled environments.

Output structure:

connector_configs/
  prod/
    prod-main/
      my-s3-sink.json
      my-jdbc-source.json
  dev/
    dev-main/
      ...

Usage:

# Multi-cluster (using config.yaml)
python dump_connector_configs.py

# Single-cluster
export CONFLUENT_CLOUD_API_KEY="..."
export CONFLUENT_CLOUD_API_SECRET="..."
export CONFLUENT_ENVIRONMENT_ID="env-xxxxx"
export CONFLUENT_CLUSTER_ID="lkc-xxxxx"
python dump_connector_configs.py

Optional environment variables:

Variable	Default	Description
CONFIG_FILE	config.yaml	YAML config for multi-cluster mode
OUTPUT_DIR	connector_configs	Root output directory
INCLUDE_SENSITIVE	false	Disable redaction of secret fields

---

consumer_details.py

Performs a comprehensive consumer group health analysis across the cluster.

Steps performed:

1. Fetches all consumer groups → consumer_groups.csv
2. Fetches per-partition lag for every group → consumer_lag_by_group_topic.csv
3. Queries sent_bytes per consumer group from the Metrics API (last 1h by default)
4. Cross-references with orphaned_topics.csv to determine if a group's topics are themselves active
5. Classifies each group → idle_consumers.csv

Classification logic

The classification uses two independent signals:

• Consumer lag — whether the group is behind the latest offset. Non-zero lag means the group knows it has unread messages.
• sent_bytes — the bytes actually delivered by the broker to that consumer group in the recent window. Zero means the broker has not sent anything to that group, regardless of lag.

if group.state == "Dead":
    → DEAD
      (no members, no committed offsets — safe to remove)

elif total_lag > 0  AND  sent_bytes == 0:
    → ZOMBIE
      (group is behind but broker delivered nothing — consumer is
       stalled, dead, or has an authorization/network problem)

elif total_lag == 0  AND  sent_bytes == 0:

    if ALL topics in the group are orphan candidates:
        → IDLE_NO_DATA
          (caught up on dead topics — nothing to read, nothing consumed)

    else:
        → IDLE_CAUGHT_UP
          (caught up on at least one active topic — could be a
           scheduled/batch consumer, or genuinely keeping up with
           low-volume traffic)

else:
    → HEALTHY
      (either sent_bytes > 0, or no anomaly in lag)

Why sent_bytes is used instead of just consumer group status: A consumer group can be in Stable state (members connected) while still processing zero records — for example, consumers that are connected but filtering all messages, or consumers that poll very slowly. sent_bytes from the broker's perspective is the only reliable signal that data actually moved.

Why lag alone is not enough:

• Zero lag on an inactive topic looks the same as zero lag on a healthy low-volume topic. Cross-referencing with orphaned topic data resolves the ambiguity (IDLE_NO_DATA vs IDLE_CAUGHT_UP).
• Zombie groups maintain positive lag forever even though nothing is consuming. Lag alone would not distinguish "behind and stuck" from "behind and catching up."

Consumer group classifications:

Classification	Lag	sent_bytes	Topics active	Meaning	Action
ZOMBIE	> 0	0	any	Behind but not receiving data	Investigate immediately — broken deployment, auth issue, dead consumer
IDLE_NO_DATA	0	0	none	Caught up on inactive topics	Safe to decommission with topics
IDLE_CAUGHT_UP	0	0	≥ 1	Caught up; topic has traffic	Verify — may be a batch/scheduled job
DEAD	n/a	n/a	n/a	No members, no offsets	Safe to delete
HEALTHY	any	> 0	any	Active consumption	No action needed

Usage:

# Run find_orphaned_topics.py first for best cross-reference accuracy
python find_orphaned_topics.py
python consumer_details.py

Optional environment variables:

Variable	Default	Description
ORPHANED_TOPICS_CSV	orphaned_topics.csv	Cross-reference file from find_orphaned_topics.py
SENT_BYTES_WINDOW_HOURS	1	Metrics look-back window in hours
OUTPUT_DIR	.	Directory for output CSV files

---

create_test_topics.py

Creates a batch of test topics via the Kafka admin protocol (not REST v3), and optionally produces a fixed number of dummy JSON messages to each topic after creation. Intended for generating test data — e.g. a known set of low-traffic topics to exercise find_orphaned_topics.py against.

Topics are named <prefix><index> starting at 0, e.g. with --prefix topic_ and --num-topics 5: topic_0, topic_1, topic_2, topic_3, topic_4.

A topic that already exists is skipped (logged as a warning) rather than failing the run — the script is safe to re-run, e.g. to top up messages on an existing set of topics.

Each dummy message is a small JSON payload:

{"index": 0, "timestamp": "2026-07-20T12:00:00+00:00", "topic": "topic_0"}

Required environment variables:

Reuses CONFLUENT_API_KEY / CONFLUENT_API_SECRET (defined above) plus:

Variable	Description
CONFLUENT_BOOTSTRAP_SERVER	Kafka bootstrap server, e.g. pkc-xxxxx.us-east-1.aws.confluent.cloud:9092

Usage:

# Create 10 topics (topic_0 .. topic_9), 3 partitions each, no messages
python create_test_topics.py --num-topics 10 --num-partitions 3

# Create 5 topics with a custom prefix and 6-way replication
python create_test_topics.py --num-topics 5 --num-partitions 6 \
    --prefix loadtest_ --replication-factor 6

# Create topics and produce 100 dummy messages to each
python create_test_topics.py --num-topics 10 --num-partitions 3 \
    --include-dummy-messages --num-dummy-messages 100

Flags:

Flag	Required	Default	Description
--num-topics	Yes	—	Number of topics to create
--num-partitions	Yes	—	Number of partitions per topic
--prefix	No	topic_	Topic name prefix
--replication-factor	No	3	Replication factor for created topics
--include-dummy-messages	No	off	Produce dummy messages to each topic after creation
--num-dummy-messages	No	0	Messages to produce per topic (requires --include-dummy-messages)

Set LOG_LEVEL=DEBUG for per-topic and per-message progress logs.

---

consume_sample_messages.py

Consumes the earliest available message from every topic on the cluster via the Kafka binary protocol, and reports what was found per topic (consumed, empty, timed out, not found, or errored). Useful as a lightweight sanity check that topics actually contain data — e.g. after running create_test_topics.py with --include-dummy-messages.

Topic discovery uses the Kafka REST v3 API (internal topics excluded by default via EXCLUDE_INTERNAL_TOPICS); consumption itself uses the Kafka protocol directly. Work is split across a bounded pool of worker threads, each with its own Consumer instance (confluent-kafka clients are not thread-safe to share), so topics are processed concurrently rather than one at a time.

Required environment variables:

Reuses CONFLUENT_REST_ENDPOINT, CONFLUENT_CLUSTER_ID, CONFLUENT_API_KEY, CONFLUENT_API_SECRET (defined above) plus:

Variable	Description
CONFLUENT_BOOTSTRAP_SERVER	Kafka bootstrap server, e.g. pkc-xxxxx.us-east-1.aws.confluent.cloud:9092

Usage:

# Default: 10 parallel workers, 10s poll timeout per topic
python consume_sample_messages.py

# More workers for a large topic count, shorter per-topic timeout
python consume_sample_messages.py --workers 20 --timeout 5

Flags:

Flag	Default	Description
--workers	10	Number of parallel consumer workers
--timeout	10.0	Per-topic poll timeout in seconds

Set LOG_LEVEL=DEBUG for per-topic progress logs.

---

migration/find_migration_candidates.py

Builds a per-principal (service account) workload inventory for a Dedicated cluster, to support the assessment described in migration/Workload Rightsizing and Migration Assessment from Dedicated to Enterprise.md. Output columns match A–Z of the Workloads sheet in migration/dedicated_to_enterprise_candidate_model.xlsx — columns AA–AK (sizing/cost formulas) are left to that spreadsheet's own formulas, since they depend on client-specific rates in its Price_Inputs sheet.

This is a read-only audit script — it makes no cluster mutations and no ACL/RBAC changes.

Running against a non-Dedicated cluster first

The script runs cleanly against any cluster type — topic/ACL/RBAC discovery and per-principal throughput/connection/request metrics work identically regardless of SKU. This makes it safe to run as a first pass against a client's Basic/Standard/Enterprise/Freight staging cluster before running it against their live Dedicated cluster (e.g. in a maintenance window).

Only two columns are Dedicated-specific, and both degrade to blank rather than a silently wrong value on a non-Dedicated cluster:

• current_dedicated_ckus_context (column Q) — the underlying dedicated_cku_count metric is Dedicated-only; on other SKUs it returns no data, so the column is blank, not 0.
• current_dedicated_monthly_cost_share (column G) — depends on a CKU count and/or a KAFKA_NUM_CKU billing line item, neither of which exists on non-Dedicated clusters (they're billed on different line items). The Billing API check and the manual-rate fallback both detect this and leave the column blank rather than reporting $0.00.

Why principal, not topic

The migration doc — following Confluent's own guidance for tracking usage by team on Dedicated clusters — recommends principal (service account) as the workload unit, since topics are often shared resources that don't map cleanly to one owning application.

What the Metrics API can and cannot provide

No Confluent Cloud Metrics API metric has both a principal_id and a topic/partition dimension. This script combines two sources:

Column(s)	Source	Notes
avg/peak_ingress_mbps, avg/peak_egress_mbps	request_bytes / response_bytes, grouped by principal_id	These are network bytes including protocol overhead — Confluent's documented per-principal showback proxy — not the raw payload bytes that received_bytes/sent_bytes measure (those have no principal dimension at all)
client_connections	active_connection_count, grouped by principal_id	Mean over the observation window
requests_per_sec	request_count, grouped by principal_id	Summed then divided by window seconds
current_dedicated_ckus_context	dedicated_cku_count	Cluster-wide gauge — same value on every row; blank on non-Dedicated clusters
partitions, compacted_partitions, monthly_storage_gb	Kafka REST v3 topic list + configs, joined to principals via ACLs ∪ RBAC role bindings ∪ Kafka Connect connector configs (ALLOW + WRITE/IDEMPOTENT_WRITE/ALL ACLs, ownership RBAC roles, or a connector's kafka.service.account.id → target topic). Topics with no owner discoverable via any of the three are reported in a separate (unattributed) summary row, not distributed across workload rows	Consumer-only principals get their own connection/request/byte metrics but zero directly-owned partitions. Wildcard/prefixed ACLs, and connectors using topics.regex/topic.prefix or targeting a topic that no longer exists, are not expanded/attributed — flagged in notes_blockers instead
current_dedicated_monthly_cost_share	Tries the Billing API (GET /billing/v1/costs, actual billed KAFKA_NUM_CKU cost) first, then dedicated_cku_count × DEDICATED_CKU_HOURLY_RATE × 730 — split proportionally by each principal's share of total request_bytes + response_bytes	Blank if neither source is available (including on non-Dedicated clusters, where neither a CKU count nor a CKU billing line item exists) — an approximation that assumes cost scales with throughput, not connection/request load

owner_team, target_sla, needs_ksqldb, needs_dedicated_broker_config, needs_public_networking, and private_networking_pattern have no reliable API signal and are always left blank for manual completion. frequent_long_peaks is a heuristic (peak_ingress_mbps > 2 × avg_ingress_mbps), not an authoritative measurement.

Required service account roles

Credential	Role	Scope
CONFLUENT_API_KEY (Kafka cluster)	DeveloperRead	Topic * and Group * on the target cluster
CONFLUENT_CLOUD_API_KEY (Cloud)	MetricsViewer	Organisation level
CONFLUENT_CLOUD_API_KEY (Cloud)	Read access to /iam/v2/role-bindings (e.g. EnvironmentAdmin/OrganizationAdmin, read-only usage)	Organisation or Environment level
CONFLUENT_CLOUD_API_KEY (Cloud)	Read access to the Connect API (e.g. Operator at cluster scope)	Target cluster(s)
CONFLUENT_CLOUD_API_KEY (Cloud) — optional	BillingAdmin	Organisation level. Only needed for the Billing API cost source (see column G above); without it, the script logs a 403 warning and falls back to DEDICATED_CKU_HOURLY_RATE automatically — no error, no missing rows

Optional environment variables

Variable	Default	Description
OBSERVATION_DAYS	7	Days of metrics history to analyse. Not the migration doc's recommended 90-day baseline — see Limitations below
DEDICATED_CKU_HOURLY_RATE	(unset)	USD/CKU-hour; fallback for current_dedicated_monthly_cost_share if the Billing API is unavailable. Leave unset (with no Billing API access) to leave that column blank (a warning is logged once)
EXCLUDE_INTERNAL_TOPICS	true	Skip __*, _confluent*, connect-* topics when attributing partitions/storage
INCLUDE_USER_PRINCIPALS	false	Include human users (principal_id starting with u-) in the workload inventory — off by default since an admin/consultant's own diagnostic activity isn't a real workload
EXCLUDE_PRINCIPAL_IDS	(unset)	Comma-separated principal_id values to exclude regardless of metrics activity, e.g. this repo's own tooling service account
CONFLUENT_ORG_ID	(auto-fetched)	Confluent organisation ID, used to build topic CRNs for RBAC role-binding lookups
OUTPUT_FILE	migration_candidates_{cluster_id}_{timestamp}.csv	Output CSV path
LOG_LEVEL	INFO	Set to DEBUG for per-topic/per-page progress logs

Usage

# Basic run — column G (cost share) will be blank
python migration/find_migration_candidates.py

# Include cost-share estimation
DEDICATED_CKU_HOURLY_RATE=4.50 python migration/find_migration_candidates.py

# Wider observation window (requires Metrics API retention to support it)
OBSERVATION_DAYS=30 python migration/find_migration_candidates.py

Output columns

Column	Description
workload_id	Generated WLK-{n:03d}, sorted by combined avg ingress+egress descending
workload_name	Metrics API principal_name label if available, else principal_id
owner_team	(blank — manual)
principal_id	Service account / principal ID
current_cluster_name	Target cluster ID
allocation_method	Constant: principal_id (Metrics API)
current_dedicated_monthly_cost_share	See cost-share formula above; blank if DEDICATED_CKU_HOURLY_RATE unset
target_sla	(blank — manual)
avg_ingress_mbps / peak_ingress_mbps	From request_bytes
avg_egress_mbps / peak_egress_mbps	From response_bytes
partitions	Summed over topics where principal has producer/owner ACL or RBAC role
compacted_partitions	Same, filtered to cleanup.policy=compact topics
client_connections	Mean active_connection_count
requests_per_sec	From request_count
current_dedicated_ckus_context	Cluster-wide dedicated_cku_count
needs_ksqldb / needs_dedicated_broker_config / needs_public_networking	(blank — manual)
frequent_long_peaks	True if peak_ingress_mbps > 2 × avg_ingress_mbps
private_networking_pattern	(blank — manual)
notes_blockers	Generated notes, e.g. missing ownership attribution or wildcard-ACL caveats
monthly_write_gb / monthly_read_gb	Derived from avg ingress/egress, scaled to a 30-day month
monthly_storage_gb	Summed retained_bytes over the principal's owned topics

Limitations

• request_bytes/response_bytes measure network bytes including protocol overhead, not raw payload bytes — expect these to run higher than find_orphaned_topics.py's topic-level received_bytes/sent_bytes.
• Default 7-day observation window, not the migration doc's recommended 90-day baseline. Increase OBSERVATION_DAYS if your account's metrics retention supports it.
• Partition/storage attribution covers literal (non-wildcard) ACLs, directly-scoped RBAC role bindings, and Kafka Connect connector configs; wildcard/prefixed ACLs and connectors using topics.regex/topic.prefix are flagged in notes_blockers rather than expanded.
• Topics with no discoverable owner appear in a single (unattributed) summary row, not a real workload — a large value there means part of the cluster needs manual ownership investigation before finalizing a migration plan.
• Feature-blocker columns (ksqlDB, broker config, public networking) have no API-derivable signal on a per-principal basis and are always manual.

---

Recommended Workflow

Health check and cleanup

# 1. Identify orphaned topics
python find_orphaned_topics.py

# 2. Analyse consumer groups (cross-references orphaned_topics.csv)
python consumer_details.py

# 3. Review idle_consumers.csv — investigate ZOMBIE groups first
#    Review orphaned_topics.csv — validate orphan candidates with teams

# 4. Tag confirmed orphans for deletion
TEAM_OWNER=platform DECOMMISSION_DATE=2026-05-01 \
  python tag_topics_for_deletion.py --action apply --input-file orphaned_topics.csv

# 5. Revoke permissions for those topics
DRY_RUN=true python remove_permissions_for_topics_to_delete.py \
  --revoke both --input-file orphaned_topics.csv
python remove_permissions_for_topics_to_delete.py \
  --revoke both --input-file orphaned_topics.csv

# 6. (Optional) Dry-run deletion to confirm the topic list
DRY_RUN=true python delete_topics.py --input-file orphaned_topics.csv

# 7. Delete confirmed orphan topics (prompts for confirmation)
python delete_topics.py --input-file orphaned_topics.csv

# 8. Check connector activity (run on a schedule for idle detection)
python idle_connectors.py

# 9. After a few days, re-run idle_connectors.py to confirm IDLE status
python idle_connectors.py

# 10. Identify idle Flink statements burning CFUs unnecessarily
python find_idle_flink_statements.py

# 11. Review idle_flink_statements_*.csv, then dry-run stop+delete for the
#     IDLE/STUCK statements found
DRY_RUN=true python delete_flink_statements.py \
  --input-file idle_flink_statements_*.csv

# 12. Stop and delete the confirmed idle/stuck Flink statements
#     (prompts for confirmation)
python delete_flink_statements.py --input-file idle_flink_statements_*.csv

Pre-migration connector audit (TGW → PrivateLink)

# Dump all connector configs for review and recreation on new cluster
python dump_connector_configs.py

# Review connector_configs/ directory
# Update bootstrap servers / API keys per connector JSON file before re-deploying

---

API Reference

API	Documentation
Confluent Cloud Metrics API	https://docs.confluent.io/cloud/current/monitoring/metrics-api.html
Kafka REST v3 API	https://docs.confluent.io/cloud/current/api.html
Confluent Connect REST API	https://docs.confluent.io/cloud/current/connectors/connect-api-section.html
Stream Catalog (tagging)	https://docs.confluent.io/cloud/current/stream-governance/stream-catalog-rest-apis.html
Flink SQL REST API	https://docs.confluent.io/cloud/current/api.html (SQL v1 — Statements)

---

Notes

• Metrics retention: Confluent Cloud retains detailed metrics for only 7 days. For longer-term trend analysis, forward metrics to Prometheus, Grafana, Datadog, or a time-series database. find_orphaned_topics.py can query such a Prometheus instance directly — see METRICS_BACKEND=prometheus above — to run OBSERVATION_DAYS beyond 7, as long as the Prometheus instance retains data that far back.
• Internal topics: find_orphaned_topics.py skips topics matching __*, _confluent*, connect-* prefixes by default (EXCLUDE_INTERNAL_TOPICS=true).
• Compact topics: The find_orphaned_topics.py throughput check may under-report activity for compacted topics with very infrequent updates. Use CHECK_LAST_MESSAGE=true for a more accurate signal on those topics.
• Connector offsets API: The offsets endpoint is not available on all connector types. Connectors with no available offsets are classified as NO_OFFSETS.