diff --git a/manifests/broker/config.pp b/manifests/broker/config.pp
index eab8f36..9a59aaa 100644
--- a/manifests/broker/config.pp
+++ b/manifests/broker/config.pp
@@ -1,40 +1,41 @@
 # Author::    Liam Bennett  (mailto:lbennett@opentable.com)
 # Copyright:: Copyright (c) 2013 OpenTable Inc
 # License::   MIT
 
 # == Class: kafka::broker::config
 #
 # This private class is meant to be called from `kafka::broker`.
 # It manages the broker config files
 #
 class kafka::broker::config(
   Stdlib::Absolutepath $config_dir = $kafka::broker::config_dir,
   String $service_name             = $kafka::broker::service_name,
   Boolean $service_install         = $kafka::broker::service_install,
   Boolean $service_restart         = $kafka::broker::service_restart,
   Hash $config                     = $kafka::broker::config,
   Stdlib::Filemode $config_mode    = $kafka::broker::config_mode,
+  String $user                     = $kafka::broker::user,
   String $group                    = $kafka::broker::group,
 ) {
 
   if ($caller_module_name != $module_name) {
     fail("Use of private class ${name} by ${caller_module_name}")
   }
 
   if ($service_install and $service_restart) {
     $config_notify = Service[$service_name]
   } else {
     $config_notify = undef
   }
 
   $doctag = 'brokerconfigs'
   file { "${config_dir}/server.properties":
     ensure  => present,
-    owner   => 'root',
+    owner   => $user,
     group   => $group,
     mode    => $config_mode,
     content => template('kafka/properties.erb'),
     notify  => $config_notify,
     require => File[$config_dir],
   }
 }
diff --git a/manifests/consumer/config.pp b/manifests/consumer/config.pp
index e02a3b6..555834f 100644
--- a/manifests/consumer/config.pp
+++ b/manifests/consumer/config.pp
@@ -1,36 +1,37 @@
 # Author::    Liam Bennett  (mailto:lbennett@opentable.com)
 # Copyright:: Copyright (c) 2013 OpenTable Inc
 # License::   MIT
 
 # == Class: kafka::consumer::config
 #
 # This private class is meant to be called from `kafka::consumer`.
 # It manages the consumer config files
 #
 class kafka::consumer::config(
   Stdlib::Absolutepath $config_dir = $kafka::consumer::config_dir,
   String $service_name             = $kafka::consumer::service_name,
   Boolean $service_install         = $kafka::consumer::service_install,
   Boolean $service_restart         = $kafka::consumer::service_restart,
   Hash $config                     = $kafka::consumer::config,
   Stdlib::Filemode $config_mode    = $kafka::consumer::config_mode,
+  String $user                     = $kafka::consumer::user,
   String $group                    = $kafka::consumer::group,
 ) {
 
   if ($service_install and $service_restart) {
     $config_notify = Service[$service_name]
   } else {
     $config_notify = undef
   }
 
   $doctag = 'consumerconfigs'
   file { "${config_dir}/consumer.properties":
     ensure  => present,
-    owner   => 'root',
+    owner   => $user,
     group   => $group,
     mode    => $config_mode,
     content => template('kafka/properties.erb'),
     notify  => $config_notify,
     require => File[$config_dir],
   }
 }
diff --git a/manifests/init.pp b/manifests/init.pp
index 7a70952..fe44a1c 100644
--- a/manifests/init.pp
+++ b/manifests/init.pp
@@ -1,230 +1,230 @@
 # Author::    Liam Bennett  (mailto:lbennett@opentable.com)
 # Copyright:: Copyright (c) 2013 OpenTable Inc
 # License::   MIT
 
 # == Class: kafka
 #
 # This class will install kafka binaries
 #
 # === Requirements/Dependencies
 #
 # Currently requires the puppetlabs/stdlib module on the Puppet Forge in
 # order to validate much of the the provided configuration.
 #
 # === Parameters
 #
 # [*version*]
 # The version of kafka that should be installed.
 #
 # [*scala_version*]
 # The scala version what kafka was built with.
 #
 # [*install_dir*]
 # The directory to install kafka to.
 #
 # [*mirror_url*]
 # The url where the kafka is downloaded from.
 #
 # [*install_java*]
 # Install java if it's not already installed.
 #
 # [*package_dir*]
 # The directory to install kafka.
 #
 # [*package_name*]
 # Package name, when installing kafka from a package.
 #
 # [*package_ensure*]
 # Package version (or 'present', 'absent', 'latest'), when installing kafka from a package.
 #
 # [*user*]
 # User to run kafka as.
 #
 # [*group*]
 # Group to run kafka as.
 #
 # [*user_id*]
 # Create the kafka user with this ID.
 #
 # [*system_user*]
 # Whether the kafka user is a system user or not.
 #
 # [*group_id*]
 # Create the kafka group with this ID.
 #
 # [*system_group*]
 # Whether the kafka group is a system group or not.
 #
 # [*manage_user*]
 # Create the kafka user if it's not already present.
 #
 # [*manage_group*]
 # Create the kafka group if it's not already present.
 #
 # [*config_dir*]
 # The directory to create the kafka config files to.
 #
 # [*log_dir*]
 # The directory for kafka log files.
 #
 # === Examples
 #
 #
 class kafka (
   String $version                   = $kafka::params::version,
   String $scala_version             = $kafka::params::scala_version,
   Stdlib::Absolutepath $install_dir = $kafka::params::install_dir,
   Stdlib::HTTPUrl $mirror_url       = $kafka::params::mirror_url,
   Boolean $install_java             = $kafka::params::install_java,
   Stdlib::Absolutepath $package_dir = $kafka::params::package_dir,
   Optional[String] $package_name    = $kafka::params::package_name,
   Optional[String] $mirror_subpath  = $kafka::params::mirror_subpath,
   Optional[String] $proxy_server    = $kafka::params::proxy_server,
   Optional[String] $proxy_port      = $kafka::params::proxy_port,
   Optional[String] $proxy_host      = $kafka::params::proxy_host,
   Optional[String] $proxy_type      = $kafka::params::proxy_type,
   String $package_ensure            = $kafka::params::package_ensure,
   String $user                      = $kafka::params::user,
   String $group                     = $kafka::params::group,
   Boolean $system_user              = $kafka::params::system_user,
   Boolean $system_group             = $kafka::params::system_group,
   Optional[Integer] $user_id        = $kafka::params::user_id,
   Optional[Integer] $group_id       = $kafka::params::group_id,
   Boolean $manage_user              = $kafka::params::manage_user,
   Boolean $manage_group             = $kafka::params::manage_group,
   Stdlib::Absolutepath $config_dir  = $kafka::params::config_dir,
   Stdlib::Absolutepath $log_dir     = $kafka::params::log_dir,
   Optional[String] $install_mode    = $kafka::params::install_mode,
 ) inherits kafka::params {
 
   if $install_java {
     class { 'java':
       distribution => 'jdk',
     }
   }
 
   if $manage_group {
     group { $group:
       ensure => present,
       gid    => $group_id,
       system => $system_group,
     }
   }
 
   if $manage_user {
     user { $user:
       ensure  => present,
       shell   => '/bin/bash',
       require => Group[$group],
       uid     => $user_id,
       system  => $system_user,
     }
   }
 
   file { $config_dir:
     ensure => directory,
-    owner  => 'root',
-    group  => 'root',
+    owner  => $user,
+    group  => $group,
   }
 
   file { $log_dir:
     ensure  => directory,
     owner   => $user,
     group   => $group,
     require => [
       Group[$group],
       User[$user],
     ],
   }
 
   if $package_name == undef {
 
     include archive
 
     $mirror_path = $mirror_subpath ? {
       # if mirror_subpath was not changed,
       # we adapt it for the version
       $kafka::params::mirror_subpath => "kafka/${version}",
       # else, we just take whatever was supplied:
       default                        => $mirror_subpath,
     }
 
     $basefilename = "kafka_${scala_version}-${version}.tgz"
     $package_url = "${mirror_url}${mirror_path}/${basefilename}"
 
     $source = $mirror_url ?{
       /tgz$/ => $mirror_url,
       default  => $package_url,
     }
 
     $install_directory = $install_dir ? {
       # if install_dir was not changed,
       # we adapt it for the scala_version and the version
       $kafka::params::install_dir => "/opt/kafka-${scala_version}-${version}",
       # else, we just take whatever was supplied:
       default                     => $install_dir,
     }
 
     file { $package_dir:
       ensure  => directory,
       owner   => $user,
       group   => $group,
       require => [
         Group[$group],
         User[$user],
       ],
     }
 
     file { $install_directory:
       ensure  => directory,
       owner   => $user,
       group   => $group,
       mode    => $install_mode,
       require => [
         Group[$group],
         User[$user],
       ],
     }
 
     file { '/opt/kafka':
       ensure  => link,
       target  => $install_directory,
       require => File[$install_directory],
     }
 
     if $proxy_server == undef and $proxy_host != undef and $proxy_port != undef {
       $final_proxy_server = "${proxy_host}:${proxy_port}"
     } else {
       $final_proxy_server = $proxy_server
     }
 
     archive { "${package_dir}/${basefilename}":
       ensure          => present,
       extract         => true,
       extract_command => 'tar xfz %s --strip-components=1',
       extract_path    => $install_directory,
       source          => $source,
       creates         => "${install_directory}/config",
       cleanup         => true,
       proxy_server    => $final_proxy_server,
       proxy_type      => $proxy_type,
       user            => $user,
       group           => $group,
       require         => [
         File[$package_dir],
         File[$install_directory],
         Group[$group],
         User[$user],
       ],
       before          => File[$config_dir],
     }
 
   } else {
 
     package { $package_name:
       ensure => $package_ensure,
       before => File[$config_dir],
     }
 
   }
 }
diff --git a/manifests/mirror/config.pp b/manifests/mirror/config.pp
index 91a4406..01cb182 100644
--- a/manifests/mirror/config.pp
+++ b/manifests/mirror/config.pp
@@ -1,54 +1,57 @@
 # Author::    Liam Bennett  (mailto:lbennett@opentable.com)
 # Copyright:: Copyright (c) 2013 OpenTable Inc
 # License::   MIT
 
 # == Class: kafka::mirror::config
 #
 # This private class is meant to be called from `kafka::mirror`.
 # It manages the mirror-maker config files
 #
 class kafka::mirror::config(
   Stdlib::Absolutepath $config_dir = $kafka::mirror::config_dir,
   String $service_name             = $kafka::mirror::service_name,
   Boolean $service_install         = $kafka::mirror::service_install,
   Boolean $service_restart         = $kafka::mirror::service_restart,
   Hash $consumer_config            = $kafka::mirror::consumer_config,
   Hash $producer_config            = $kafka::mirror::producer_config,
   Stdlib::Filemode $config_mode    = $kafka::mirror::config_mode,
+  String $user                     = $kafka::mirror::user,
   String $group                    = $kafka::mirror::group,
 ) {
 
   if $caller_module_name != $module_name {
     fail("Use of private class ${name} by ${caller_module_name}")
   }
 
   if $consumer_config['group.id'] == '' {
     fail('[Consumer] You need to specify a value for group.id')
   }
   if $consumer_config['zookeeper.connect'] == '' {
     fail('[Consumer] You need to specify a value for zookeeper.connect')
   }
   if $producer_config['bootstrap.servers'] == '' {
     fail('[Producer] You need to specify a value for bootstrap.servers')
   }
 
   class { 'kafka::consumer::config':
     config_dir      => $config_dir,
     config_mode     => $config_mode,
     service_name    => $service_name,
     service_install => $service_install,
     service_restart => $service_restart,
     config          => $consumer_config,
+    user            => $user,
     group           => $group,
   }
 
   class { 'kafka::producer::config':
     config_dir      => $config_dir,
     config_mode     => $config_mode,
     service_name    => $service_name,
     service_install => $service_install,
     service_restart => $service_restart,
     config          => $producer_config,
+    user            => $user,
     group           => $group,
   }
 }
diff --git a/manifests/producer/config.pp b/manifests/producer/config.pp
index 03d8753..fad8113 100644
--- a/manifests/producer/config.pp
+++ b/manifests/producer/config.pp
@@ -1,36 +1,37 @@
 # Author::    Liam Bennett  (mailto:lbennett@opentable.com)
 # Copyright:: Copyright (c) 2013 OpenTable Inc
 # License::   MIT
 
 # == Class: kafka::producer::config
 #
 # This private class is meant to be called from `kafka::producer`.
 # It manages the producer config files
 #
 class kafka::producer::config(
   Stdlib::Absolutepath $config_dir = $kafka::producer::config_dir,
   String $service_name             = $kafka::producer::service_name,
   Boolean $service_install         = $kafka::producer::service_install,
   Boolean $service_restart         = $kafka::producer::service_restart,
   Hash $config                     = $kafka::producer::config,
   Stdlib::Filemode $config_mode    = $kafka::producer::config_mode,
+  String $user                     = $kafka::producer::user,
   String $group                    = $kafka::producer::group,
 ) {
 
   if ($service_install and $service_restart) {
     $config_notify = Service[$service_name]
   } else {
     $config_notify = undef
   }
 
   $doctag = 'producerconfigs'
   file { "${config_dir}/producer.properties":
     ensure  => present,
-    owner   => 'root',
+    owner   => $user,
     group   => $group,
     mode    => $config_mode,
     content => template('kafka/properties.erb'),
     notify  => $config_notify,
     require => File[$config_dir],
   }
 }
diff --git a/spec/acceptance/broker_spec.rb b/spec/acceptance/broker_spec.rb
index 4cfd990..8f9973c 100644
--- a/spec/acceptance/broker_spec.rb
+++ b/spec/acceptance/broker_spec.rb
@@ -1,224 +1,224 @@
 require 'spec_helper_acceptance'
 
 describe 'kafka::broker' do
   it 'works with no errors' do
     pp = <<-EOS
       class { 'kafka::broker':
         config => {
           'zookeeper.connect' => 'localhost:2181',
         },
       } ->
       kafka::topic { 'demo':
         ensure    => present,
         zookeeper => 'localhost:2181',
       }
     EOS
 
     apply_manifest(pp, catch_failures: true)
     apply_manifest(pp, catch_changes: true)
   end
 
   describe 'kafka::broker::install' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::broker':
             config => {
               'zookeeper.connect' => 'localhost:2181',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe group('kafka') do
         it { is_expected.to exist }
       end
 
       describe user('kafka') do
         it { is_expected.to exist }
         it { is_expected.to belong_to_group 'kafka' }
         it { is_expected.to have_login_shell '/bin/bash' }
       end
 
       describe file('/var/tmp/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka-2.12-2.4.1') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka') do
         it { is_expected.to be_linked_to('/opt/kafka-2.12-2.4.1') }
       end
 
       describe file('/opt/kafka/config') do
         it { is_expected.to be_directory }
-        it { is_expected.to be_owned_by 'root' }
-        it { is_expected.to be_grouped_into 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
+        it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/var/log/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 
   describe 'kafka::broker::config' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::broker':
             config => {
               'zookeeper.connect' => 'localhost:2181',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe file('/opt/kafka/config/server.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
         it { is_expected.to contain 'zookeeper.connect=localhost:2181' }
       end
     end
 
     context 'with custom config dir' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::broker':
             config => {
               'zookeeper.connect' => 'localhost:2181',
             },
             config_dir => '/opt/kafka/custom_config'
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe file('/opt/kafka/custom_config/server.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
         it { is_expected.to contain 'zookeeper.connect=localhost:2181' }
       end
     end
 
     context 'with specific version' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::broker':
             version => '2.4.0',
             config  => {
               'zookeeper.connect' => 'localhost:2181',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe file('/opt/kafka/config/server.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 
   describe 'kafka::broker::service' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::broker':
             config => {
               'zookeeper.connect' => 'localhost:2181',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe file('/etc/init.d/kafka'), if: (fact('operatingsystemmajrelease') =~ %r{(5|6)} && fact('osfamily') == 'RedHat') do
         it { is_expected.to be_file }
         it { is_expected.to be_owned_by 'root' }
         it { is_expected.to be_grouped_into 'root' }
       end
 
       describe file('/etc/systemd/system/kafka.service'), if: (fact('operatingsystemmajrelease') == '7' && fact('osfamily') == 'RedHat') do
         it { is_expected.to be_file }
         it { is_expected.to be_owned_by 'root' }
         it { is_expected.to be_grouped_into 'root' }
       end
 
       describe service('kafka') do
         it { is_expected.to be_running }
         it { is_expected.to be_enabled }
       end
     end
   end
 
   describe 'kafka::broker::service' do
     context 'with log4j/jmx parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           exec { 'create log dir':
             command => '/bin/mkdir -p /some/path/to/logs',
             creates => '/some/path/to/logs',
           } ->
           class { 'kafka::broker':
             config => {
               'zookeeper.connect' => 'localhost:2181',
             },
             heap_opts  => '-Xmx512M -Xmx512M',
             log4j_opts => '-Dlog4j.configuration=file:/tmp/log4j.properties',
             jmx_opts   => '-Dcom.sun.management.jmxremote',
             opts       => '-Djava.security.policy=/some/path/my.policy',
             log_dir    => '/some/path/to/logs'
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
         apply_manifest(pp, catch_changes: true)
       end
 
       describe file('/etc/init.d/kafka'), if: (fact('operatingsystemmajrelease') =~ %r{(5|6)} && fact('osfamily') == 'RedHat') do
         it { is_expected.to be_file }
         it { is_expected.to be_owned_by 'root' }
         it { is_expected.to be_grouped_into 'root' }
         it { is_expected.to contain 'export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote"' }
         it { is_expected.to contain 'export KAFKA_HEAP_OPTS="-Xmx512M -Xmx512M"' }
         it { is_expected.to contain 'export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:/tmp/log4j.properties"' }
       end
 
       describe file('/etc/systemd/system/kafka.service'), if: (fact('operatingsystemmajrelease') == '7' && fact('osfamily') == 'RedHat') do
         it { is_expected.to be_file }
         it { is_expected.to be_owned_by 'root' }
         it { is_expected.to be_grouped_into 'root' }
         it { is_expected.to contain "Environment='KAFKA_JMX_OPTS=-Dcom.sun.management.jmxremote'" }
         it { is_expected.to contain "Environment='KAFKA_HEAP_OPTS=-Xmx512M -Xmx512M'" }
         it { is_expected.to contain "Environment='KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:/tmp/log4j.properties'" }
         it { is_expected.to contain "Environment='KAFKA_OPTS=-Djava.security.policy=/some/path/my.policy'" }
         it { is_expected.to contain "Environment='LOG_DIR=/some/path/to/logs'" }
       end
 
       describe service('kafka') do
         it { is_expected.to be_running }
         it { is_expected.to be_enabled }
       end
     end
   end
 end
diff --git a/spec/acceptance/consumer_spec.rb b/spec/acceptance/consumer_spec.rb
index f346f2c..31c4526 100644
--- a/spec/acceptance/consumer_spec.rb
+++ b/spec/acceptance/consumer_spec.rb
@@ -1,157 +1,157 @@
 require 'spec_helper_acceptance'
 
 describe 'kafka::consumer' do
   it 'works with no errors' do
     pp = <<-EOS
       class { 'kafka::consumer':
         service_config => {
           topic            => 'demo',
           bootstrap-server => 'localhost:9092',
         },
       }
     EOS
 
     apply_manifest(pp, catch_failures: true)
     apply_manifest(pp, catch_changes: true)
   end
 
   describe 'kafka::consumer::install' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::consumer':
             service_config => {
               topic            => 'demo',
               bootstrap-server => 'localhost:9092',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe group('kafka') do
         it { is_expected.to exist }
       end
 
       describe user('kafka') do
         it { is_expected.to exist }
         it { is_expected.to belong_to_group 'kafka' }
         it { is_expected.to have_login_shell '/bin/bash' }
       end
 
       describe file('/var/tmp/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka-2.12-2.4.1') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka') do
         it { is_expected.to be_linked_to('/opt/kafka-2.12-2.4.1') }
       end
 
       describe file('/opt/kafka/config') do
         it { is_expected.to be_directory }
-        it { is_expected.to be_owned_by 'root' }
-        it { is_expected.to be_grouped_into 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
+        it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/var/log/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 
   describe 'kafka::consumer::config' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::consumer':
             service_config => {
               topic            => 'demo',
               bootstrap-server => 'localhost:9092',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe file('/opt/kafka/config/consumer.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 
   describe 'kafka::consumer::config' do
     context 'with custom config_dir' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::consumer':
             service_config => {
               topic            => 'demo',
               bootstrap-server => 'localhost:9092',
             },
             config_dir => '/opt/kafka/custom_config',
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe file('/opt/kafka/custom_config/consumer.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 
   describe 'kafka::consumer::service' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::consumer':
             service_config => {
               topic            => 'demo',
               bootstrap-server => 'localhost:9092',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe file('/etc/init.d/kafka-consumer'), if: (fact('operatingsystemmajrelease') =~ %r{(5|6)} && fact('osfamily') == 'RedHat') do
         it { is_expected.to be_file }
         it { is_expected.to be_owned_by 'root' }
         it { is_expected.to be_grouped_into 'root' }
         it { is_expected.to contain 'export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=9993"' }
         it { is_expected.to contain 'export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:/opt/kafka/config/log4j.properties"' }
       end
 
       describe file('/etc/systemd/system/kafka-consumer.service'), if: (fact('operatingsystemmajrelease') == '7' && fact('osfamily') == 'RedHat') do
         it { is_expected.to be_file }
         it { is_expected.to be_owned_by 'root' }
         it { is_expected.to be_grouped_into 'root' }
         it { is_expected.to contain 'Environment=\'KAFKA_JMX_OPTS=-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=9993\'' }
         it { is_expected.to contain 'Environment=\'KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:/opt/kafka/config/log4j.properties\'' }
       end
 
       describe service('kafka-consumer') do
         it { is_expected.to be_running }
         it { is_expected.to be_enabled }
       end
     end
   end
 end
diff --git a/spec/acceptance/init_spec.rb b/spec/acceptance/init_spec.rb
index 5b31550..fdd3b92 100644
--- a/spec/acceptance/init_spec.rb
+++ b/spec/acceptance/init_spec.rb
@@ -1,212 +1,212 @@
 require 'spec_helper_acceptance'
 
 describe 'kafka' do
   it 'works with no errors' do
     pp = <<-EOS
       class { 'kafka': }
     EOS
 
     apply_manifest(pp, catch_failures: true)
     apply_manifest(pp, catch_changes: true)
   end
 
   describe 'kafka::init' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka': }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe group('kafka') do
         it { is_expected.to exist }
       end
 
       describe user('kafka') do
         it { is_expected.to exist }
         it { is_expected.to belong_to_group 'kafka' }
         it { is_expected.to have_login_shell '/bin/bash' }
       end
 
       describe file('/var/tmp/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka-2.12-2.4.1') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka') do
         it { is_expected.to be_linked_to('/opt/kafka-2.12-2.4.1') }
       end
 
       describe file('/opt/kafka/config') do
         it { is_expected.to be_directory }
-        it { is_expected.to be_owned_by 'root' }
-        it { is_expected.to be_grouped_into 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
+        it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/var/log/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
 
     context 'with specific kafka version' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka':
             version => '2.4.0',
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe group('kafka') do
         it { is_expected.to exist }
       end
 
       describe user('kafka') do
         it { is_expected.to exist }
         it { is_expected.to belong_to_group 'kafka' }
         it { is_expected.to have_login_shell '/bin/bash' }
       end
 
       describe file('/var/tmp/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka-2.12-2.4.0') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka') do
         it { is_expected.to be_linked_to('/opt/kafka-2.12-2.4.0') }
       end
 
       describe file('/opt/kafka/config') do
         it { is_expected.to be_directory }
-        it { is_expected.to be_owned_by 'root' }
-        it { is_expected.to be_grouped_into 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
+        it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/var/log/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
 
     context 'with specific scala version' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka':
             scala_version => '2.13',
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe group('kafka') do
         it { is_expected.to exist }
       end
 
       describe user('kafka') do
         it { is_expected.to exist }
         it { is_expected.to belong_to_group 'kafka' }
         it { is_expected.to have_login_shell '/bin/bash' }
       end
 
       describe file('/var/tmp/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka-2.13-2.4.1') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka') do
         it { is_expected.to be_linked_to('/opt/kafka-2.13-2.4.1') }
       end
 
       describe file('/opt/kafka/config') do
         it { is_expected.to be_directory }
-        it { is_expected.to be_owned_by 'root' }
-        it { is_expected.to be_grouped_into 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
+        it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/var/log/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
 
     context 'with specific config dir' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka':
             config_dir => '/opt/kafka/custom_config',
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe group('kafka') do
         it { is_expected.to exist }
       end
 
       describe user('kafka') do
         it { is_expected.to exist }
         it { is_expected.to belong_to_group 'kafka' }
         it { is_expected.to have_login_shell '/bin/bash' }
       end
 
       describe file('/var/tmp/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka-2.12-2.4.1') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka') do
         it { is_expected.to be_linked_to('/opt/kafka-2.12-2.4.1') }
       end
 
       describe file('/opt/kafka/custom_config') do
         it { is_expected.to be_directory }
-        it { is_expected.to be_owned_by 'root' }
-        it { is_expected.to be_grouped_into 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
+        it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/var/log/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 end
diff --git a/spec/acceptance/mirror_spec.rb b/spec/acceptance/mirror_spec.rb
index aa44f84..e61102f 100644
--- a/spec/acceptance/mirror_spec.rb
+++ b/spec/acceptance/mirror_spec.rb
@@ -1,235 +1,235 @@
 require 'spec_helper_acceptance'
 
 describe 'kafka::mirror' do
   it 'works with no errors' do
     pp = <<-EOS
       class { 'kafka::mirror':
         consumer_config => {
           'group.id'          => 'kafka-mirror',
           'bootstrap.servers' => 'localhost:9092',
         },
         producer_config => {
           'bootstrap.servers' => 'localhost:9092',
         },
         service_config => {
           'whitelist' => '.*',
         },
       }
     EOS
 
     apply_manifest(pp, catch_failures: true)
     apply_manifest(pp, catch_changes: true)
   end
 
   describe 'kafka::mirror::install' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::mirror':
             consumer_config => {
               'group.id'          => 'kafka-mirror',
               'bootstrap.servers' => 'localhost:9092',
             },
             producer_config => {
               'bootstrap.servers' => 'localhost:9092',
             },
             service_config => {
               'whitelist' => '.*',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
       end
 
       describe group('kafka') do
         it { is_expected.to exist }
       end
 
       describe user('kafka') do
         it { is_expected.to exist }
         it { is_expected.to belong_to_group 'kafka' }
         it { is_expected.to have_login_shell '/bin/bash' }
       end
 
       describe file('/var/tmp/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka-2.12-2.4.1') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka') do
         it { is_expected.to be_linked_to('/opt/kafka-2.12-2.4.1') }
       end
 
       describe file('/opt/kafka/config') do
         it { is_expected.to be_directory }
-        it { is_expected.to be_owned_by 'root' }
-        it { is_expected.to be_grouped_into 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
+        it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/var/log/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 
   describe 'kafka::mirror::config' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::mirror':
             consumer_config => {
               'group.id'          => 'kafka-mirror',
               'bootstrap.servers' => 'localhost:9092',
             },
             producer_config => {
               'bootstrap.servers' => 'localhost:9092',
             },
             service_config => {
               'whitelist' => '.*',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
         apply_manifest(pp, catch_changes: true)
       end
 
       describe file('/opt/kafka/config/consumer.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka/config/producer.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
 
     context 'with custom config_dir' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::mirror':
             consumer_config => {
               'group.id'          => 'kafka-mirror',
               'bootstrap.servers' => 'localhost:9092',
             },
             producer_config => {
               'bootstrap.servers' => 'localhost:9092',
             },
             service_config => {
               'whitelist' => '.*',
             },
             config_dir => '/opt/kafka/custom_config',
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
         apply_manifest(pp, catch_changes: true)
       end
 
       describe file('/opt/kafka/custom_config/consumer.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka/custom_config/producer.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
 
     context 'with specific version' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::mirror':
             version         => '2.4.0',
             consumer_config => {
               'group.id'          => 'kafka-mirror',
               'bootstrap.servers' => 'localhost:9092',
             },
             producer_config => {
               'bootstrap.servers' => 'localhost:9092',
             },
             service_config => {
               'whitelist' => '.*',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
         apply_manifest(pp, catch_changes: true)
       end
 
       describe file('/opt/kafka/config/consumer.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka/config/producer.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 
   describe 'kafka::mirror::service' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::mirror':
             consumer_config => {
               'group.id'          => 'kafka-mirror',
               'bootstrap.servers' => 'localhost:9092',
             },
             producer_config => {
               'bootstrap.servers' => 'localhost:9092',
             },
             service_config => {
               'whitelist' => '.*',
             },
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
         apply_manifest(pp, catch_changes: true)
       end
 
       describe file('/etc/init.d/kafka-mirror'), if: (fact('operatingsystemmajrelease') =~ %r{(5|6)} && fact('osfamily') == 'RedHat') do
         it { is_expected.to be_file }
         it { is_expected.to be_owned_by 'root' }
         it { is_expected.to be_grouped_into 'root' }
         it { is_expected.to contain 'export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=9991"' }
         it { is_expected.to contain 'export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:/opt/kafka/config/log4j.properties"' }
       end
 
       describe file('/etc/systemd/system/kafka-mirror.service'), if: (fact('operatingsystemmajrelease') == '7' && fact('osfamily') == 'RedHat') do
         it { is_expected.to be_file }
         it { is_expected.to be_owned_by 'root' }
         it { is_expected.to be_grouped_into 'root' }
         it { is_expected.to contain 'Environment=\'KAFKA_JMX_OPTS=-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=9991\'' }
         it { is_expected.to contain 'Environment=\'KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:/opt/kafka/config/log4j.properties\'' }
       end
 
       describe service('kafka-mirror') do
         it { is_expected.to be_running }
         it { is_expected.to be_enabled }
       end
     end
   end
 end
diff --git a/spec/acceptance/producer_spec.rb b/spec/acceptance/producer_spec.rb
index 3a1c38f..c97d2f3 100644
--- a/spec/acceptance/producer_spec.rb
+++ b/spec/acceptance/producer_spec.rb
@@ -1,147 +1,147 @@
 require 'spec_helper_acceptance'
 
 describe 'kafka::producer', if: (fact('operatingsystemmajrelease') == '6' && fact('osfamily') == 'RedHat') do # systemd systems not supported by kafka::producer::service
   it 'works with no errors' do
     pp = <<-EOS
       exec { 'create fifo':
         command => '/usr/bin/mkfifo /tmp/kafka-producer',
         user    => 'kafka',
         creates => '/tmp/kafka-producer',
       } ->
       class { 'kafka::producer':
         service_config => {
           'broker-list' => 'localhost:9092',
           topic         => 'demo',
         },
         input => '3<>/tmp/kafka-producer 0>&3',
       }
     EOS
 
     apply_manifest(pp, catch_failures: true)
     apply_manifest(pp, catch_changes: true)
   end
 
   describe 'kafka::producer::install' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           exec { 'create fifo':
             command => '/usr/bin/mkfifo /tmp/kafka-producer',
             user    => 'kafka',
             creates => '/tmp/kafka-producer',
           } ->
           class { 'kafka::producer':
             service_config => {
               'broker-list' => 'localhost:9092',
               topic         => 'demo',
             },
             input          => '3<>/tmp/kafka-producer 0>&3',
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
         apply_manifest(pp, catch_changes: true)
       end
 
       describe group('kafka') do
         it { is_expected.to exist }
       end
 
       describe user('kafka') do
         it { is_expected.to exist }
         it { is_expected.to belong_to_group 'kafka' }
         it { is_expected.to have_login_shell '/bin/bash' }
       end
 
       describe file('/var/tmp/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka-2.12-2.4.1') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/opt/kafka') do
         it { is_expected.to be_linked_to('/opt/kafka-2.12-2.4.1') }
       end
 
       describe file('/opt/kafka/config') do
         it { is_expected.to be_directory }
-        it { is_expected.to be_owned_by 'root' }
-        it { is_expected.to be_grouped_into 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
+        it { is_expected.to be_grouped_into 'kafka' }
       end
 
       describe file('/var/log/kafka') do
         it { is_expected.to be_directory }
         it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 
   describe 'kafka::producer::config' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           exec { 'create fifo':
             command => '/usr/bin/mkfifo /tmp/kafka-producer',
             user    => 'kafka',
             creates => '/tmp/kafka-producer',
           } ->
           class { 'kafka::producer':
             service_config => {
               'broker-list' => 'localhost:9092',
               topic         => 'demo',
             },
             input          => '3<>/tmp/kafka-producer 0>&3',
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
         apply_manifest(pp, catch_changes: true)
       end
 
       describe file('/opt/kafka/config/producer.properties') do
         it { is_expected.to be_file }
-        it { is_expected.to be_owned_by 'root' }
+        it { is_expected.to be_owned_by 'kafka' }
         it { is_expected.to be_grouped_into 'kafka' }
       end
     end
   end
 
   describe 'kafka::producer::service' do
     context 'with default parameters' do
       it 'works with no errors' do
         pp = <<-EOS
           class { 'kafka::producer':
             service_config => {
               'broker-list' => 'localhost:9092',
               topic         => 'demo',
             },
             input          => '3<>/tmp/kafka-producer 0>&3',
           }
         EOS
 
         apply_manifest(pp, catch_failures: true)
         apply_manifest(pp, catch_changes: true)
       end
 
       describe file('/etc/init.d/kafka-producer') do
         it { is_expected.to be_file }
         it { is_expected.to be_owned_by 'root' }
         it { is_expected.to be_grouped_into 'root' }
         it { is_expected.to contain 'export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=9992"' }
         it { is_expected.to contain 'export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:/opt/kafka/config/log4j.properties"' }
       end
 
       describe service('kafka-producer') do
         it { is_expected.to be_running }
         it { is_expected.to be_enabled }
       end
     end
   end
 end