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.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 Allowed values: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL	PLAINTEXT	—
optional.connection.ssl_ca	string	No	PEM-encoded CA certificate for TLS verification (optional for client-certificate auth; validates the broker's certificate)	—	—
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",
    "bootstrap_servers": "broker1:9092,broker2: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",
    "bootstrap_servers": "broker: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 a broker-issued CA certificate for TLS verification

Config Payload

{
  "type": "KAFKA",
  "required": {
    "auth_mode": "SASL",
    "bootstrap_servers": "kafka-example-1234.example-cloud.com:12345"
  },
  "masked": {
    "sasl_username": "avnadmin",
    "sasl_password": "your-service-password"
  },
  "optional": {
    "connection": {
      "security_protocol": "SASL_SSL",
      "sasl_mechanism": "PLAIN",
      "ssl_ca": "-----BEGIN CERTIFICATE-----\nMIIDazCCAlOgAwIBAgIUfakefakefakefakefakefakefakefakewDQYJKoZIhvcNAQEL\nBQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM\nGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDAeFw0yNjAxMDEwMDAwMDBaFw0zNjAx\nMDEwMDAwMDBaMEUxCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEw\nHwYDVQQKDBhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQwggEiMA0GCSqGSIb3DQEB\nAQUAA4IBDwAwggEKAoIBAQDFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEF\nAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEFAKEF\n-----END CERTIFICATE-----\n"
    },
    "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 using a client certificate and private key instead of SASL — no username/password required

Config Payload

{
  "type": "KAFKA",
  "required": {
    "auth_mode": "CLIENT_CERT",
    "bootstrap_servers": "broker.internal: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"
  },
  "optional": {
    "connection": {
      "security_protocol": "SSL",
      "ssl_ca": "-----BEGIN CERTIFICATE-----\nMIIDazCCAlOgAwIBAgIUfakefakefakefakefakefakefakefakewDQYJKoZIhvcNAQEL\nBQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM\n-----END CERTIFICATE-----\n"
    }
  },
  "sampling": {
    "strategy": "AUTOMATIC",
    "rows_per_page": 50
  },
  "detectors": [
    {
      "type": "PII",
      "enabled": true
    }
  ]
}

Schedule

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