SASL_SSL security.protocol for kafka source

[jamhamblin]

We're trying to export a kafka topic to datadog for log retention. In the kafka documentation the settings required for the topic are:

bootstrap.servers={{ BOOTSTRAP SERVER }}
security.protocol=SASL_SSL
sasl.mechanisms=PLAIN
sasl.username={{ CLUSTER_API_KEY }}
sasl.password={{ CLUSTER_API_SECRET }}

# Best practice for higher availability in librdkafka clients prior to 1.7
session.timeout.ms=45000

I have attempted to configure this in vector as follows:

      kafka:
        type: kafka
        bootstrap_servers: "<bootstrap-server>"
        group_id: <group-id>
        librdkafka_options."security.protocol": SASL_SSL
        topics:
          - <topic-name>
        sasl:
          enabled: true
          mechanism: PLAIN
          username: "${CLUSTER_API_KEY}"
          password: "${CLUSTER_API_SECRET}"

Setting librdkafka_options."security.protocol": gives the error:

Configuration error. error=unknown field `librdkafka_options."security.protocol"` in `sources.kafka`

Whereas without librdkafka_options."security.protocol set there is an error broker transport failure indicating that the wrong security.protocol is being used.

I believe that this issue is caused by SASL_SSL not being included in the active listeners here: https://github.com/vectordotdev/vector/blob/master/scripts/integration/kafka/compose.yaml#L16.
Hoping however that there is something wrong with the configuration and would really appreciate any help from the community 🙏

[jamhamblin]

Coming back to say that enabling setting tls.enabled: true solved our issue