[Unit]
Description=Apache Kafka server (<%= (@service_name.split(/-/)[1] and @service_name.split(/-/)[1].capitalize) or 'broker' -%>)
Documentation=http://kafka.apache.org/documentation.html
<%- if @service_requires_zookeeper -%>
Wants=zookeeper.service
After=zookeeper.service
Requires=zookeeper.service
<%- end -%>

[Service]
Type=simple
User=kafka
Group=kafka
SyslogIdentifier=<%= @service_name %>
<%- case @service_name
  when 'kafka' -%>
Environment='KAFKA_HEAP_OPTS=<%= @heap_opts %>'
Environment='KAFKA_LOG4J_OPTS=<%= @log4j_opts %>'
Environment='KAFKA_JMX_OPTS=<%= @jmx_opts %>'
Environment='KAFKA_OPTS=<%= @opts %>'
Environment='LOG_DIR=<%= @log_dir %>'
ExecStart=/opt/kafka/bin/kafka-server-start.sh <%= @config_dir %>/server.properties
  <%- when 'kafka-consumer' -%>
Environment='KAFKA_LOG4J_OPTS=<%= @consumer_log4j_opts %>'
Environment='KAFKA_JMX_OPTS=<%= @consumer_jmx_opts %>'
ExecStart=/opt/kafka/bin/kafka-console-consumer.sh <% @consumer_service_config.sort.each do |k,v| -%><% unless v.to_s.strip.empty? -%>--<%= k -%>=<%= v.is_a?(Array) ? v.join(',') : v %> <% end -%><% end -%>
  <%- when 'kafka-mirror' -%>
Environment='KAFKA_LOG4J_OPTS=<%= @mirror_log4j_opts %>'
Environment='KAFKA_JMX_OPTS=<%= @mirror_jmx_opts %>'
Environment='KAFKA_HEAP_OPTS=-Xmx<%= @max_heap -%>'
ExecStart=/opt/kafka/bin/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config <%= @consumer_config -%> --num.streams <%= @num_streams -%> --producer.config <%= @producer_config -%><%- if (scope.function_versioncmp([scope.lookupvar('kafka::version'), '0.9.0.0']) < 0) -%> --num.producers <%= @num_producers -%><%- end -%><%- if !@whitelist.eql?('') -%> --whitelist='<%= @whitelist -%>'<%- end %><%- if !@blacklist.eql?('') -%> --blacklist='<%= @blacklist -%>'<%- end -%> <%= @abort_on_send_failure_opt %>
  <%- when 'kafka-producer' -%>
Environment='KAFKA_LOG4J_OPTS=<%= @producer_log4j_opts %>'
Environment='KAFKA_JMX_OPTS=<%= @producer_jmx_opts %>'
ExecStart=/opt/kafka/bin/kafka-console-producer.sh <% @producer_service_config.sort.each do |k,v| -%><% unless v.to_s.strip.empty? -%>--<%= k -%>=<%= v.is_a?(Array) ? v.join(',') : v %> <% end -%><% end -%> <%= @input %>
<%- end %>
LimitNOFILE=<%= @limit_nofile %>
LimitCORE=infinity

[Install]
WantedBy=multi-user.target