Nuxeo folks. In kafka-config.xml.nxftl, the following version supports SASL plaintext, SASL SCRAM-SHA-512, SASL SCRAM-SHA-512 over SSL, and two-way SSL. Please consider:

<#escape x as x?xml>
<?xml version="1.0"?>
<component name="org.nuxeo.kafka.defaultConfig">
<#if "${kafka.enabled}" == "true">
  <require>org.nuxeo.runtime.stream.kafka.service</require>
  <extension point="kafkaConfig" target="org.nuxeo.runtime.stream.kafka.service">
    <kafkaConfig name="default" topicPrefix="${kafka.topicPrefix}">
      <producer>
        <property name="bootstrap.servers">${kafka.bootstrap.servers}</property>
        <property name="default.replication.factor">${kafka.default.replication.factor}</property>
<#if ((kafka.sasl.enabled)!"false") == "true" || ((kafka.ssl)!"false") == "true">
        <property name="security.protocol">${kafka.security.protocol}</property>
</#if>
<#if "${kafka.sasl.enabled}" == "true">
        <property name="sasl.mechanism">${kafka.sasl.mechanism}</property>
        <property name="sasl.jaas.config">${kafka.sasl.jaas.config}</property>
</#if>
<#if "${kafka.ssl}" == "true">
        <property name="ssl.truststore.type">${kafka.truststore.type}</property>
        <property name="ssl.truststore.location">${kafka.truststore.path}</property>
        <property name="ssl.truststore.password">${kafka.truststore.password}</property>
<#if ((kafka.security.protocol)!"") == "SSL">
        <property name="ssl.keystore.type">${kafka.keystore.type}</property>
        <property name="ssl.keystore.location">${kafka.keystore.path}</property>
        <property name="ssl.keystore.password">${kafka.keystore.password}</property>
</#if>
</#if>
      </producer>
      <consumer>
        <property name="bootstrap.servers">${kafka.bootstrap.servers}</property>
        <property name="request.timeout.ms">${kafka.request.timeout.ms}</property>
        <property name="max.poll.interval.ms">${kafka.max.poll.interval.ms}</property>
        <property name="session.timeout.ms">${kafka.session.timeout.ms}</property>
        <property name="heartbeat.interval.ms">${kafka.heartbeat.interval.ms}</property>
        <property name="max.poll.records">${kafka.max.poll.records}</property>
<#if ((kafka.sasl.enabled)!"false") == "true" || ((kafka.ssl)!"false") == "true">
        <property name="security.protocol">${kafka.security.protocol}</property>
</#if>
<#if "${kafka.sasl.enabled}" == "true">
        <property name="sasl.mechanism">${kafka.sasl.mechanism}</property>
        <property name="sasl.jaas.config">${kafka.sasl.jaas.config}</property>
</#if>
<#if "${kafka.ssl}" == "true">
        <property name="ssl.truststore.type">${kafka.truststore.type}</property>
        <property name="ssl.truststore.location">${kafka.truststore.path}</property>
        <property name="ssl.truststore.password">${kafka.truststore.password}</property>
<#if ((kafka.security.protocol)!"") == "SSL">
        <property name="ssl.keystore.type">${kafka.keystore.type}</property>
        <property name="ssl.keystore.location">${kafka.keystore.path}</property>
        <property name="ssl.keystore.password">${kafka.keystore.password}</property>
</#if>
</#if>
      </consumer>
    </kafkaConfig>
    <kafkaConfig name="bulk" topicPrefix="${kafka.topicPrefix}bulk-">
      <producer>
        <property name="bootstrap.servers">${kafka.bootstrap.servers}</property>
        <property name="default.replication.factor">${kafka.default.replication.factor}</property>
<#if ((kafka.sasl.enabled)!"false") == "true" || ((kafka.ssl)!"false") == "true">
        <property name="security.protocol">${kafka.security.protocol}</property>
</#if>
<#if "${kafka.sasl.enabled}" == "true">
        <property name="sasl.mechanism">${kafka.sasl.mechanism}</property>
        <property name="sasl.jaas.config">${kafka.sasl.jaas.config}</property>
</#if>
<#if "${kafka.ssl}" == "true">
        <property name="ssl.truststore.type">${kafka.truststore.type}</property>
        <property name="ssl.truststore.location">${kafka.truststore.path}</property>
        <property name="ssl.truststore.password">${kafka.truststore.password}</property>
<#if ((kafka.security.protocol)!"") == "SSL">
        <property name="ssl.keystore.type">${kafka.keystore.type}</property>
        <property name="ssl.keystore.location">${kafka.keystore.path}</property>
        <property name="ssl.keystore.password">${kafka.keystore.password}</property>
</#if>
</#if>
      </producer>
      <consumer>
        <property name="bootstrap.servers">${kafka.bootstrap.servers}</property>
        <property name="request.timeout.ms">${kafka.request.timeout.ms}</property>
        <property name="max.poll.interval.ms">${kafka.max.poll.interval.ms}</property>
        <property name="session.timeout.ms">${kafka.session.timeout.ms}</property>
        <property name="heartbeat.interval.ms">${kafka.heartbeat.interval.ms}</property>
<#if ((kafka.sasl.enabled)!"false") == "true" || ((kafka.ssl)!"false") == "true">
        <property name="security.protocol">${kafka.security.protocol}</property>
</#if>
<#if "${kafka.sasl.enabled}" == "true">
        <property name="sasl.mechanism">${kafka.sasl.mechanism}</property>
        <property name="sasl.jaas.config">${kafka.sasl.jaas.config}</property>
</#if>
<#if "${kafka.ssl}" == "true">
        <property name="ssl.truststore.type">${kafka.truststore.type}</property>
        <property name="ssl.truststore.location">${kafka.truststore.path}</property>
        <property name="ssl.truststore.password">${kafka.truststore.password}</property>
<#if ((kafka.security.protocol)!"") == "SSL">
        <property name="ssl.keystore.type">${kafka.keystore.type}</property>
        <property name="ssl.keystore.location">${kafka.keystore.path}</property>
        <property name="ssl.keystore.password">${kafka.keystore.password}</property>
</#if>
</#if>
      </consumer>
    </kafkaConfig>
  </extension>
</#if>
</component>
</#escape>