Kafka Connect configuration parameters

{% set connect_distributed_log4j_properties = services.kafka_connect.config.connect_distributed_log4j_properties_content %}

log4j.rootLogger={{ connect_distributed_log4j_properties['rootLogger'] }}, connectDRFA, connectRFA, stdout

# Send the logs to the console.
#
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

#
# The `%X{connector.context}` parameter in the layout includes connector-specific and task-specific information
# in the log message, where appropriate. This makes it easier to identify those log messages that apply to a
# specific connector. Simply add this parameter to the log layout configuration below to include the contextual information.
#
#log4j.appender.stdout.layout.ConversionPattern=[%d] %p %X{connector.context}%m (%c:%L)%n
log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n
connect.log.pattern=[%d] %p %m (%c:%L)%n

{% for key, value in connect_distributed_log4j_properties['loggers'] | dictsort -%}
log4j.logger.{{ key }}={{ value }}
{% endfor -%}

log4j.appender.connectDRFA=org.apache.log4j.DailyRollingFileAppender
log4j.appender.connectDRFA.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.connectDRFA.File=${kafka.logs.dir}/connect-distributed.log
log4j.appender.connectDRFA.layout=org.apache.log4j.PatternLayout
log4j.appender.connectDRFA.layout.ConversionPattern=${connect.log.pattern}
log4j.appender.connectDRFA.MaxBackupIndex={{ connect_distributed_log4j_properties['MaxBackupIndex'] }}

log4j.appender.connectRFA=org.apache.log4j.RollingFileAppender
log4j.appender.connectRFA.File=${kafka.logs.dir}/connect-distributed.log
log4j.appender.connectRFA.layout=org.apache.log4j.PatternLayout
log4j.appender.connectRFA.layout.ConversionPattern=[%d] %p %m (%c)%n
log4j.appender.connectRFA.Append=true
log4j.appender.connectRFA.MaxBackupIndex={{ connect_distributed_log4j_properties['MaxBackupIndex'] }}
log4j.appender.connectRFA.MaxFileSize={{ connect_distributed_log4j_properties['MaxFileSize'] }}

{% if cluster.config.kerberos_client and cluster.config.kerberos_client.enable_kerberos %}
KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    useTicketCache=false
    serviceName="kafka"
    keyTab="{{ cluster.config.kerberos_client.keytab_dir }}/kafka-connect.service.keytab"
    principal="kafka-connect/{{ ansible_fqdn }}@{{ cluster.config.kerberos_client.realm }}";
};
{%- elif cluster.config.sasl_plain_auth_default_config is not none %}
    {%- set credential = cluster.config.sasl_plain_auth_default_config.sasl_plain_users_data %}
KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="kafka-connect"
  password="{{ credential['kafka-connect'] }}";
};
{%- endif %}