kafka_consumer: collect Kafka Connect connector metrics and configurations

[piochelepiotr]

What does this PR do?

Adds Kafka Connect monitoring to the `kafka_consumer` integration via a new `connectors.py` module (`KafkaConnectCollector`). When `kafka_connect_url` is configured and `enable_cluster_monitoring` is enabled, the agent queries the Kafka Connect REST API and emits connector health metrics and configuration events.

New metrics (all tagged `kafka_cluster_id`, `connect_url`):

Metric	Description
`kafka.connector.count`	Total connectors per Connect worker
`kafka.connector.running`	1 if RUNNING, 0 otherwise — tagged with `connector`, `connector_type`, `connector_class`, `connector_state`
`kafka.connector.task.count`	Configured task count per connector
`kafka.connector.tasks`	Task count by state — tagged with `task_state` (running/failed/paused/unassigned)
`kafka.connector.task.running`	Per-task 0/1, gated by `kafka_connect_collect_task_metrics: true` (default off)

Configuration events (`data-streams-message`, `config_type: connector`): emitted when connector config changes or the 1-hour TTL expires, using the same hash-based dedup pattern as broker/topic configs. Sensitive config values (passwords, secrets, JAAS configs) are redacted before emission.

Plugin inventory events (`config_type: connector_plugins`): emitted on the `kafka_configs_refresh_interval` cadence.

Connectivity status in heartbeat: instead of a service check, whether each Connect API endpoint was reachable is reported in the cluster monitoring heartbeat event (`connect_api_status` field).

Authentication methods supported:

• No auth (default)
• Basic auth (`kafka_connect_username` / `kafka_connect_password`)
• TLS client certificates (`kafka_connect_tls_cert` / `kafka_connect_tls_key`)
• Custom CA verification (`kafka_connect_tls_ca_cert`)
• OAuth/OIDC (`kafka_connect_oauth_token_provider` — identical shape to `schema_registry_oauth_token_provider`)
• Confluent Cloud Connect (`kafka_connect_confluent_cloud_environment_id` / `kafka_connect_confluent_cloud_cluster_id`)

Dashboard: a new integration dashboard asset is included — Kafka Connect - Connector Monitoring (DSM).

Implementation notes:

• Collection is gated by `enable_cluster_monitoring` (same as schema registry and other DSM-only metrics)
• Uses a dedicated `requests.Session` (not the shared check HTTP client) so Connect auth doesn't conflict with schema registry auth
• Single `GET /connectors?expand=info&expand=status` call per worker — no N+1 requests
• Confluent Cloud Connect uses the same REST collection logic via `https://api.confluent.cloud/connect/v1/environments/{env}/clusters/{cluster}\`

Motivation

Kafka Connect is widely deployed alongside Kafka for CDC, streaming ETL, and data replication. Currently there is no native Datadog integration for monitoring connector health — users have to rely on JMX or custom scripts. The primary pain point is knowing when a connector or task has failed so data flow is interrupted.

Review checklist (to be filled by reviewers)

• Feature or bugfix MUST have appropriate tests (unit, integration, e2e)
• Add `qa/required` if this PR needs QA validation, or `qa/skip-qa` if it does not. Exactly one of the two is required.
• If you need to backport this PR to another branch, you can add the `backport/` label to the PR and it will automatically open a backport PR once this one is merged

[dd-octo-sts]

Validation Report

All 21 validations passed.

Show details

Validation	Description	Status
agent-reqs	Verify check versions match the Agent requirements file	✅
ci	Validate CI configuration and code coverage settings	✅
codeowners	Validate every integration has a CODEOWNERS entry	✅
config	Validate default configuration files against spec.yaml	✅
dep	Verify dependency pins are consistent and Agent-compatible	✅
http	Validate integrations use the HTTP wrapper correctly	✅
imports	Validate check imports do not use deprecated modules	✅
integration-style	Validate check code style conventions	✅
jmx-metrics	Validate JMX metrics definition files and config	✅
labeler	Validate PR labeler config matches integration directories	✅
legacy-signature	Validate no integration uses the legacy Agent check signature	✅
license-headers	Validate Python files have proper license headers	✅
licenses	Validate third-party license attribution list	✅
metadata	Validate metadata.csv metric definitions	✅
models	Validate configuration data models match spec.yaml	✅
openmetrics	Validate OpenMetrics integrations disable the metric limit	✅
package	Validate Python package metadata and naming	✅
qa-label	Validate the pull request declares whether it needs QA for the next Agent release	✅
readmes	Validate README files have required sections	✅
saved-views	Validate saved view JSON file structure and fields	✅
version	Validate version consistency between package and changelog	✅

View full run