Xenial aware

manifests/config.pp

@@ -24,8 +24,15 @@
     notify  => $notify_service,
   }
 
-  file { '/etc/default/kafka':
-    content => template('confluent_kafka/kafka.defaults.erb'),
+  if ($::confluent_kafka::params::initstyle == 'init') {
+    file { '/etc/default/kafka':
+      content => template('confluent_kafka/kafka.defaults.erb'),
+    }
+  }
+  else {
+    file { '/etc/default/kafka':
+      content => template('confluent_kafka/kafka.defaults_systemd.erb'),
+    }
   }
 
   file { '/etc/kafka/server.properties':

manifests/init.pp

@@ -61,6 +61,9 @@
 # [*log4j_opts*]
 #   Override Log4J file
 #
+# [extra_args*]
+#   Override Log4J file
+#
 
 class confluent_kafka (
   $package_name      = $::confluent_kafka::params::package_name,
@@ -82,6 +85,7 @@
   $jvm_perf_opts     = $::confluent_kafka::params::jvm_perf_opts,
   $jmx_opts          = $::confluent_kafka::params::jmx_opts,
   $log4j_opts        = $::confluent_kafka::params::log4j_opts,
+  $extra_args        = $::confluent_kafka::params::extra_args,
   $platform_version  = $::confluent_kafka::params::platform_version,
 
 ) inherits ::confluent_kafka::params {

manifests/install.pp

@@ -35,8 +35,8 @@
 
 
   exec { 'apt-get update':
-    command => "/usr/bin/apt-get update",
-    alias   => "apt-update",
+    command => '/usr/bin/apt-get update',
+    alias   => 'apt-update',
   }
 
   if $::confluent_kafka::install_java {
@@ -46,8 +46,8 @@
   }
 
   package { "${::confluent_kafka::package_name}-${::confluent_kafka::scala_version}":
+    ensure  => $::confluent_kafka::version,
     require => Exec[apt-update],
-    ensure => $::confluent_kafka::version,
   }
 
   group { 'kafka':
@@ -61,11 +61,25 @@
   }
 
   if $::confluent_kafka::install_service {
-    file { "/etc/init.d/${::confluent_kafka::service_name}":
-      mode   => '0755',
-      owner  => 'root',
-      group  => 'root',
-      source => 'puppet:///modules/confluent_kafka/kafka.init',
+    if ($::confluent_kafka::params::initstyle == 'init') {
+      file { "/etc/init.d/${::confluent_kafka::service_name}":
+        mode   => '0755',
+        owner  => 'root',
+        group  => 'root',
+        source => 'puppet:///modules/confluent_kafka/kafka.init',
+      }
+    }
+    else {
+      file { "/lib/systemd/system/${::confluent_kafka::service_name}.service":
+        mode    => '0644',
+        owner   => 'root',
+        group   => 'root',
+        content => template('confluent_kafka/kafka.systemd.erb'),
+      } ~>
+      exec { 'systemctl daemon-reload # for kafka':
+        refreshonly => true,
+        notify      => Service[$::confluent_kafka::service_name]
+      }
     }
   }
 

manifests/params.pp

@@ -4,6 +4,28 @@
 # It sets variables according to platform.
 #
 class confluent_kafka::params {
+  case $::osfamily {
+  'Debian': {
+    case $::operatingsystem {
+      'Debian': {
+        case $::operatingsystemmajrelease {
+          '7': { $initstyle = 'init' }
+          '8': { $initstyle = 'systemd' }
+          default: { $initstyle = undef }
+        }
+      }
+      'Ubuntu': {
+        case $::operatingsystemmajrelease {
+          '14.04': { $initstyle = 'upstart' }
+          '16.04': { $initstyle = 'systemd' }
+          default: { $initstyle = undef }
+        }
+      }
+      default: { $initstyle = 'init' }
+    }
+   }
+  }
+
   $scala_version     = '2.10.4'
   $platform_version  = '1.0'
   $service_name      = 'kafka'
@@ -23,6 +45,7 @@
   $jvm_perf_opts     = '-XX:PermSize=48m -XX:MaxPermSize=48m -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35'
   $jmx_opts          = '-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.net.preferIPv4Stack=true -Dcom.sun.management.jmxremote.port=9999'
   $log4j_opts        = '-Dlog4j.configuration=file:/etc/kafka/log4j.properties'
+  $extra_args        = ''
 
   $brokers           = {
       'localhost' => 0,

templates/kafka.defaults.erb

@@ -7,3 +7,6 @@ export KAFKA_HEAP_OPTS="<%= scope.lookupvar('confluent_kafka::jvm_heap_mem') %>"
 export KAFKA_JMX_OPTS="<%= scope.lookupvar('confluent_kafka::jmx_opts') %>"
 export KAFKA_JVM_PERFORMANCE_OPTS="<%= scope.lookupvar('confluent_kafka::jvm_perf_opts')%>"
 ulimit -n <%= scope.lookupvar('confluent_kafka::params::max_nofiles')%>
+<% if ! scope.lookupvar('confluent_kafka::extra_args').empty? then -%>
+export EXTRA_ARGS="<%= scope.lookupvar('confluent_kafka::extra_args') %>"
+<% end -%>

templates/kafka.defaults_systemd.erb

@@ -0,0 +1,12 @@
+# Have to export these log variables since currently a bug in kafka-server-start
+# that doesnt export these variables
+LOG_DIR="<%= scope.lookupvar('confluent_kafka::app_log_dir') %>"
+KAFKA_LOG4J_OPTS="<%= scope.lookupvar('confluent_kafka::log4j_opts') %>"
+
+KAFKA_HEAP_OPTS="<%= scope.lookupvar('confluent_kafka::jvm_heap_mem') %>"
+KAFKA_JMX_OPTS="<%= scope.lookupvar('confluent_kafka::jmx_opts') %>"
+KAFKA_JVM_PERFORMANCE_OPTS="<%= scope.lookupvar('confluent_kafka::jvm_perf_opts')%>"
+<% if ! scope.lookupvar('confluent_kafka::extra_args').empty? then -%>
+EXTRA_ARGS="<%= scope.lookupvar('confluent_kafka::extra_args') %>"
+<% end -%>
+

templates/kafka.systemd.erb

@@ -0,0 +1,20 @@
+[Unit]
+Description=Kafka server
+After=network.target network-online.target
+
+[Service]
+Type=forking
+User=kafka
+Group=kafka
+EnvironmentFile=-/etc/default/kafka
+ExecStart=/usr/bin/kafka-server-start -daemon /etc/kafka/server.properties
+ExecStop=/usr/bin/kafka-server-stop
+TimeoutStopSec=60s
+Restart=on-failure
+StartLimitInterval=0
+RestartSec=10s
+SuccessExitStatus=143
+LimitNOFILE=<%= scope.lookupvar('confluent_kafka::params::max_nofiles')%>
+
+[Install]
+WantedBy=multi-user.target