Apache Kafka

Required

Fields required for a valid configuration.

Path	Type	Required	Description	Default	Constraints
required	object	Yes	—	—	—

Masked

Sensitive fields under(secrets/credentials).

Path	Type	Required	Description	Default	Constraints
masked	object	No	—	—	—

Optional

Optional configuration fields.

Path	Type	Required	Description	Default	Constraints
optional	object	No	—	—	no extra properties
optional.connection	object	No	Kafka client connection and security options.	—	no extra properties
optional.connection.request_timeout_ms	integer	No	Client request timeout in milliseconds	30000	min 1000
optional.connection.rest_use_tls	boolean	No	REST proxy only: call the proxy over HTTPS. Turn off for a plain-HTTP proxy (e.g. a local Karapace).	true	—
optional.connection.sasl_mechanism	enum	No	SASL mechanism used when security_protocol is SASL_* Allowed values: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512	PLAIN	—
optional.connection.security_protocol	enum	No	Kafka client security protocol. Leave unset to derive it from the authentication mode (SASL → SASL_SSL, Client Certificate → SSL, No Authentication → PLAINTEXT); set it only to override, e.g. SASL_PLAINTEXT for a SASL broker without TLS. Allowed values: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL	—	—
optional.scope	object	No	Kafka topic selection scope.	—	no extra properties
optional.scope.exclude_topics	array	No	Topic denylist	—	—
optional.scope.exclude_topics[]	string	No	—	—	—
optional.scope.include_internal	boolean	No	Include internal topics (names starting with __)	false	—
optional.scope.include_topics	array	No	Optional topic allowlist	—	—
optional.scope.include_topics[]	string	No	—	—	—
optional.scope.topic_limit	integer	No	Optional cap on number of topic assets	—	min 1

Extracted Metadata

Topic

Field	Type	Required	Description
partition_count	integer	Yes	Number of partitions
replication_factor	integer	No	Replication factor
earliest_offset	integer	No	Sum of earliest offsets across partitions
latest_offset	integer	No	Sum of latest offsets across partitions
retention_ms	integer	No	Topic retention in milliseconds
cleanup_policy	string	No	Topic cleanup policy (delete/compact)

Examples

Topic discovery scan

Discover Kafka topics and sample recent messages on a local, unauthenticated broker

Config Payload

{
  "type": "KAFKA",
  "required": {
    "auth_mode": "NONE",
    "host": "localhost",
    "port": 9092
  },
  "optional": {
    "connection": {
      "security_protocol": "PLAINTEXT"
    }
  },
  "sampling": {
    "strategy": "LATEST",
    "rows_per_page": 50
  }
}

Schedule

{
  "enabled": true,
  "preset": "weekday_business",
  "cron": "13 6 * * 1-5",
  "timezone": "UTC"
}

SASL_SSL secrets detection

Sample messages from selected topics over SASL_SSL and run secrets detection

Config Payload

{
  "type": "KAFKA",
  "required": {
    "auth_mode": "SASL",
    "host": "broker.internal.example.com",
    "port": 9093
  },
  "masked": {
    "sasl_username": "your-username",
    "sasl_password": "your-password"
  },
  "optional": {
    "connection": {
      "security_protocol": "SASL_SSL",
      "sasl_mechanism": "SCRAM-SHA-256"
    },
    "scope": {
      "include_topics": [
        "payments",
        "users"
      ]
    }
  },
  "sampling": {
    "strategy": "RANDOM",
    "rows_per_page": 100
  },
  "detectors": [
    {
      "type": "SECRETS",
      "enabled": true
    },
    {
      "type": "PII",
      "enabled": true
    }
  ]
}

Schedule

{
  "enabled": true,
  "preset": "weekday_business",
  "cron": "37 8 * * 1-5",
  "timezone": "UTC"
}

Managed Kafka (Aiven/Confluent Cloud) with CA cert

Connect to a managed SASL_SSL Kafka service (e.g. Aiven, Confluent Cloud) using the service's CA certificate for TLS verification

Config Payload

{
  "type": "KAFKA",
  "required": {
    "auth_mode": "SASL",
    "host": "kafka-example-1234.example-cloud.com",
    "port": 12345
  },
  "masked": {
    "sasl_username": "avnadmin",
    "sasl_password": "your-service-password",
    "ssl_ca": "-----BEGIN CERTIFICATE-----\nMIIDazCCAlOgAwIBAgIUfakefakefakefakefakefakefakefakewDQYJKoZIhvcNAQEL\nBQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM\nGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDAeFw0yNjAxMDEwMDAwMDBaFw0zNjAx\n-----END CERTIFICATE-----\n"
  },
  "optional": {
    "connection": {
      "security_protocol": "SASL_SSL",
      "sasl_mechanism": "PLAIN"
    },
    "scope": {
      "include_internal": false
    }
  },
  "sampling": {
    "strategy": "AUTOMATIC",
    "rows_per_page": 100
  }
}

Schedule

{
  "enabled": true,
  "preset": "nightly",
  "cron": "5 2 * * *",
  "timezone": "UTC"
}

Mutual TLS (client certificate) authentication

Authenticate to a Kafka broker with a client access certificate and access key instead of SASL — no username/password required

Config Payload

{
  "type": "KAFKA",
  "required": {
    "auth_mode": "CLIENT_CERT",
    "host": "broker.internal",
    "port": 9094
  },
  "masked": {
    "ssl_certfile": "-----BEGIN CERTIFICATE-----\nMIIDCTCCAfGgAwIBAgIUfakefakefakefakefakefakefakefakewDQYJKoZIhvcNAQEL\nBQAwFDESMBAGA1UEAwwJY2xpZW50LWNhMB4XDTI2MDEwMTAwMDAwMFoXDTI3MDEw\nMTAwMDAwMFowFDESMBAGA1UEAwwJY2xpZW50LWNhMIIBIjANBgkqhkiG9w0BAQEF\n-----END CERTIFICATE-----\n",
    "ssl_keyfile": "-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDFAKEFAKEFAKEF\nAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEF\n-----END PRIVATE KEY-----\n",
    "ssl_ca": "-----BEGIN CERTIFICATE-----\nMIIDazCCAlOgAwIBAgIUfakefakefakefakefakefakefakefakewDQYJKoZIhvcNAQEL\nBQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM\nGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDAeFw0yNjAxMDEwMDAwMDBaFw0zNjAx\n-----END CERTIFICATE-----\n"
  },
  "optional": {
    "connection": {
      "security_protocol": "SSL"
    }
  },
  "sampling": {
    "strategy": "AUTOMATIC",
    "rows_per_page": 50
  },
  "detectors": [
    {
      "type": "PII",
      "enabled": true
    }
  ]
}

Schedule

{
  "enabled": true,
  "preset": "nightly",
  "cron": "45 1 * * *",
  "timezone": "UTC"
}

Kafka REST Proxy scan

Scan topics through a Kafka REST Proxy (Confluent REST Proxy, Karapace) with HTTP basic auth — no broker protocol access required

Config Payload

{
  "type": "KAFKA",
  "required": {
    "auth_mode": "REST",
    "host": "kafka-rest.example.com",
    "port": 8082
  },
  "masked": {
    "username": "your-username",
    "password": "your-password"
  },
  "optional": {
    "connection": {
      "rest_use_tls": true
    },
    "scope": {
      "include_internal": false
    }
  },
  "sampling": {
    "strategy": "AUTOMATIC",
    "rows_per_page": 100
  },
  "detectors": [
    {
      "type": "PII",
      "enabled": true
    }
  ]
}

Schedule

{
  "enabled": true,
  "preset": "daily",
  "cron": "25 7 * * *",
  "timezone": "UTC"
}