Page MenuHomeSoftware Heritage
Files F9696762

No OneTemporary
Actions

View Options

diff --git a/spec/classes/mirror_spec.rb b/spec/classes/mirror_spec.rb
index 501d5d8..2df6b5d 100644
--- a/spec/classes/mirror_spec.rb
+++ b/spec/classes/mirror_spec.rb
@@ -1,125 +1,142 @@
require 'spec_helper'
require 'shared_examples_param_validation'
describe 'kafka::mirror', type: :class do
let :facts do
{
osfamily: 'Debian',
os: { family: 'Debian' },
operatingsystem: 'Ubuntu',
operatingsystemrelease: '14.04',
lsbdistcodename: 'trusty',
architecture: 'amd64',
service_provider: 'upstart'
}
end
let :common_params do
{
consumer_config: {
'group.id' => 'kafka-mirror',
'zookeeper.connect' => 'localhost:2181'
},
producer_config: {
'bootstrap.servers' => 'localhost:9092'
}
}
end
let :params do
common_params
end
it { is_expected.to contain_class('kafka::mirror::install').that_comes_before('Class[kafka::mirror::config]') }
it { is_expected.to contain_class('kafka::mirror::config').that_comes_before('Class[kafka::mirror::service]') }
it { is_expected.to contain_class('kafka::mirror::service').that_comes_before('Class[kafka::mirror]') }
it { is_expected.to contain_class('kafka::mirror') }
context 'on Debian' do
describe 'kafka::mirror::install' do
context 'defaults' do
it { is_expected.to contain_class('kafka') }
end
end
describe 'kafka::mirror::config' do
context 'defaults' do
it { is_expected.to contain_class('kafka::consumer::config') }
it { is_expected.to contain_class('kafka::producer::config') }
end
end
describe 'kafka::mirror::service' do
context 'defaults' do
it { is_expected.to contain_file('/etc/init.d/kafka-mirror') }
+ it { is_expected.to contain_file('/etc/init.d/kafka-mirror').with_content %r{whitelist='.*'} }
it { is_expected.to contain_file('/etc/init.d/kafka-mirror').with_content %r{/opt/kafka/config/(?=.*consumer)|(?=.*producer).properties} }
it { is_expected.to contain_service('kafka-mirror') }
end
+ context 'new consumer enabled' do
+ let :params do
+ common_params.merge(consumer_config: { 'group.id' => 'kafka-mirror', 'bootstrap.servers' => 'localhost:9092' })
+ end
+
+ it { is_expected.to contain_file('/etc/init.d/kafka-mirror').with_content %r{whitelist '.*'} }
+ end
end
end
context 'on CentOS' do
let :facts do
{
osfamily: 'RedHat',
os: { family: 'RedHat' },
operatingsystem: 'CentOS',
operatingsystemrelease: '7',
operatingsystemmajrelease: '7',
architecture: 'amd64',
path: '/usr/local/sbin',
service_provider: 'systemd'
}
end
describe 'kafka::mirror::install' do
context 'defaults' do
it { is_expected.to contain_class('kafka') }
end
end
describe 'kafka::mirror::config' do
context 'defaults' do
it { is_expected.to contain_class('kafka::consumer::config') }
it { is_expected.to contain_class('kafka::producer::config') }
end
end
describe 'kafka::mirror::service' do
context 'defaults' do
it { is_expected.to contain_file('/etc/systemd/system/kafka-mirror.service').that_notifies('Exec[systemctl-daemon-reload]') }
it { is_expected.to contain_file('/etc/systemd/system/kafka-mirror.service').with_content %r{^LimitNOFILE=65536$} }
+ it { is_expected.to contain_file('/etc/systemd/system/kafka-mirror.service').with_content %r{whitelist='.*'} }
it { is_expected.to contain_file('/etc/systemd/system/kafka-mirror.service').with_content %r{/opt/kafka/config/(?=.*consumer)|(?=.*producer).properties} }
it do
is_expected.to contain_file('/etc/init.d/kafka-mirror').with(
ensure: 'absent'
)
end
it { is_expected.to contain_exec('systemctl-daemon-reload').that_comes_before('Service[kafka-mirror]') }
it { is_expected.to contain_service('kafka-mirror') }
end
+ context 'new consumer enabled' do
+ let :params do
+ common_params.merge(consumer_config: { 'group.id' => 'kafka-mirror', 'bootstrap.servers' => 'localhost:9092' })
+ end
+
+ it { is_expected.to contain_file('/etc/systemd/system/kafka-mirror.service').with_content %r{whitelist '.*'} }
+ end
+
context 'service_requires_zookeeper disabled' do
let :params do
common_params.merge(service_requires_zookeeper: false)
end
it { is_expected.not_to contain_file('/etc/systemd/system/kafka-mirror.service').with_content %r{^Requires=zookeeper.service$} }
end
context 'service_requires_zookeeper enabled' do
let :params do
common_params.merge(service_requires_zookeeper: true)
end
it { is_expected.to contain_file('/etc/systemd/system/kafka-mirror.service').with_content %r{^Requires=zookeeper.service$} }
end
end
end
it_validates_parameter 'mirror_url'
end
diff --git a/templates/init.erb b/templates/init.erb
index 2f19f35..76aac6f 100644
--- a/templates/init.erb
+++ b/templates/init.erb
@@ -1,148 +1,149 @@
#!/bin/sh
#
# Init file for Apache Kafka <%= @service_name.split(/-/)[1] and @service_name.split(/-/)[1].capitalize %>
#
<%- if @osfamily == 'Debian' -%>
### BEGIN INIT INFO
# Provides: <%- @service_name -%>
<%- if @service_requires_zookeeper -%>
# Required-Start: zookeeper
<%- else -%>
# Required-Start:
<%- end -%>
# Required-Stop:
# Default-Start: 2 3 4 5
# Default-Stop: 0 1 6
# X-Interactive: true
# Short-Description: Apache Kafka is a distributed publish-subscribe messaging system
### END INIT INFO
<%- else -%>
# chkconfig: 35 85 15
# description: Apache Kafka is a distributed publish-subscribe messaging system
# pidfile: /var/run/<%= @service_name -%>.pid
<%- end -%>
NAME=<%= @service_name %>
<% @environment.sort.map do |k,v| -%>
<% unless v.to_s.strip.empty? -%>
export <%= k %>="<%= v %>"
<% end -%>
<% end -%>
PID_FILE="/var/run/$NAME.pid"
KAFKA_USER=<%= @user %>
<%- case @service_name
when 'kafka' -%>
PGREP_PATTERN=kafka.Kafka
DAEMON="<%= @bin_dir %>/kafka-server-start.sh"
DAEMON_OPTS="<%= @config_dir %>/server.properties"
<%- when 'kafka-consumer' -%>
PGREP_PATTERN=kafka.tools.ConsoleConsumer
DAEMON="<%= @bin_dir %>/kafka-console-consumer.sh"
DAEMON_OPTS="<% @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' -%>
PGREP_PATTERN=kafka.tools.MirrorMaker
DAEMON="<%= @bin_dir %>/kafka-run-class.sh"
-DAEMON_OPTS="kafka.tools.MirrorMaker --consumer.config <%= @config_dir %>/consumer.properties --num.streams <%= @num_streams -%> --producer.config <%= @config_dir %>/producer.properties<%- 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 %>"
+DAEMON_OPTS="kafka.tools.MirrorMaker --consumer.config <%= @config_dir %>/consumer.properties --num.streams <%= @num_streams -%> --producer.config <%= @config_dir %>/producer.properties
+<%- if (scope.function_versioncmp([scope.lookupvar('kafka::version'), '0.9.0.0']) < 0) -%> --num.producers <%= @num_producers -%><%- end -%><%- if !@whitelist.eql?('') -%> <%- if @consumer_config['bootstrap.servers'] -%> --whitelist '<%= @whitelist -%>'<%- else -%> --whitelist='<%= @whitelist -%>'<%- end -%> <%- end -%><%- if !@blacklist.eql?('') -%><%- if @consumer_config['bootstrap.servers'] -%> --blacklist '<%= @blacklist -%>'<%- else -%> --blacklist='<%= @blacklist -%>'<%- end -%><%- end -%> <%= @abort_on_send_failure_opt %>"
<%- when 'kafka-producer' -%>
PGREP_PATTERN=kafka.tools.ConsoleProducer
DAEMON="<%= @bin_dir %>/kafka-console-producer.sh"
DAEMON_OPTS="<% @producer_service_config.sort.each do |k,v| -%><% unless v.to_s.strip.empty? -%>--<%= k -%>=<%= v.is_a?(Array) ? v.join(',') : v %> <% end -%><% end -%>"
PRODUCER_INPUT="<%= @input %>"
<%- end -%>
if [ -f /etc/default/kafka ]; then
. /etc/default/kafka
fi
start() {
ulimit -n 65536
ulimit -s 10240
ulimit -c unlimited
if [ -f "$PID_FILE" ]; then
PID=`cat "$PID_FILE"`
if [ `ps -p "$PID" -o pid= || echo 1` -eq `pgrep -f "$PGREP_PATTERN" || echo 2` ] ; then
echo "$PID_FILE exists, process is already running"
exit 0
fi
echo "$PID_FILE exists but the process is not running. Deleting $PID_FILE and re-trying"
rm -f -- "$PID_FILE"
start
return $?
fi
/bin/su "$KAFKA_USER" -c "KAFKA_JMX_OPTS=\"$KAFKA_JMX_OPTS\" $DAEMON $DAEMON_OPTS<%- if @service_name == 'kafka-producer' -%> $PRODUCER_INPUT<%- end -%> >/dev/null 2>&1 &"
sleep 2
PID=`pgrep -f "$PGREP_PATTERN"`
if [ -z "$PID" ]; then
echo "$NAME could not be started"
exit 1
fi
echo "$PID" > "$PID_FILE";
echo "$NAME started"
return 0
}
stop() {
if ! [ -f "$PID_FILE" ]; then
echo -n "$PID_FILE does not exist"
if PID=`pgrep -f "$PGREP_PATTERN"` ; then
echo -n ", but process is running"
echo "$PID" > "$PID_FILE"
else
echo -n ", and process is not running"
return 1
fi
fi
kill `cat "$PID_FILE"`;
rm -f -- "$PID_FILE";
echo "$NAME stopped"
return 0
}
status() {
if ! [ -f "$PID_FILE" ]; then
echo "$NAME stopped"
exit 1
fi
PID=`cat "$PID_FILE"`
if ! [ `ps -p "$PID" -o pid= || echo 1` -eq `pgrep -f "$PGREP_PATTERN" || echo 2` ] ; then
echo "$NAME stopped but pid file exists"
exit 1
fi
echo "$NAME running with pid $PID"
exit 0
}
case "$1" in
status)
status
;;
start)
echo "Starting daemon: $NAME"
start
;;
stop)
echo "Stopping daemon: $NAME"
stop
;;
restart)
echo "Restarting daemon: $NAME"
stop
sleep 2
start
;;
*)
echo "Usage: "$1" {status|start|stop|restart}"
exit 1
esac
exit 0
diff --git a/templates/unit.erb b/templates/unit.erb
index c40c37b..c130218 100644
--- a/templates/unit.erb
+++ b/templates/unit.erb
@@ -1,36 +1,36 @@
[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=network.target syslog.target zookeeper.service
Requires=zookeeper.service
<%- else -%>
After=network.target syslog.target
<%- end -%>
[Service]
Type=simple
User=<%= @user %>
Group=<%= @group %>
SyslogIdentifier=<%= @service_name %>
<% @environment.sort.map do |k,v| -%>
<% unless v.to_s.strip.empty? -%>
Environment='<%= k %>=<%= v %>'
<% end -%>
<% end -%>
<%- case @service_name
when 'kafka' -%>
ExecStart=<%= @bin_dir %>/kafka-server-start.sh <%= @config_dir %>/server.properties
<%- when 'kafka-consumer' -%>
ExecStart=<%= @bin_dir %>/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' -%>
-ExecStart=<%= @bin_dir %>/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config <%= @config_dir %>/consumer.properties --num.streams <%= @num_streams -%> --producer.config <%= @config_dir %>/producer.properties<%- 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 %>
+ExecStart=<%= @bin_dir %>/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config <%= @config_dir %>/consumer.properties --num.streams <%= @num_streams -%> --producer.config <%= @config_dir %>/producer.properties<%- if (scope.function_versioncmp([scope.lookupvar('kafka::version'), '0.9.0.0']) < 0) -%> --num.producers <%= @num_producers -%><%- end -%><%- if !@whitelist.eql?('') -%><%- if @consumer_config['bootstrap.servers'] -%> --whitelist '<%= @whitelist -%>'<%- else -%> --whitelist='<%= @whitelist -%>'<%- end -%><%- end -%><%- if !@blacklist.eql?('') -%><%- if @consumer_config['bootstrap.servers'] -%> --blacklist '<%= @blacklist -%>'<%- else -%> --blacklist='<%= @blacklist -%>'<%- end -%><%- end -%> <%= @abort_on_send_failure_opt %>
<%- when 'kafka-producer' -%>
ExecStart=<%= @bin_dir %>/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

File Metadata

Mime Type
text/x-diff
Expires
Mon, Aug 18, 9:21 PM (6 d, 1 h ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3331615

Event Timeline

About · Developer information · Legal