init.pp

class kafka_broker(
  $zookeeper_connect = 'localhost:2181'
) {
  class { 'kafka':
    version       => '0.8.2.0',
    scala_version => '2.9.2',
    install_java  => false
  }

  class { 'kafka::broker':
    service_install => false,
    config          => {
      'broker.id'                 => ip_to_int($::ipaddress),
      'enable.zookeeper'          => true,
      'topic.delete.enable'       => true,
      'auto.create.topics.enable' => true,
      'num.partitions'            => 8,
      'zookeeper.connect'         => "${zookeeper_connect}/kafka",
      'log.dirs'                  => '/var/log/kafka',
    }
  }

  file { '/usr/local/bin/zookeeper_connect.py':
    ensure => file,
    owner  => 'root',
    group  => 'root',
    mode   => '0755',
    source => 'puppet:///modules/kafka_broker/zookeeper_connect.py',
    alias  => 'zookeeper_connect.py',
  }

  class { 'upstart':
    init_dir => '/etc/init',
  }

  upstart::job { 'kafka':
    description   => 'Kafka broker service',
    author        => 'Spantree Technology Group',
    version       => '0.8.2.0',
    #instance     => 'kafka-broker',
    start_on      => 'runlevel [2345]',
    stop_on       => 'runlevel [!2345]',
    respawn       => true,
    respawn_limit => '2 5',
    #user         => 'kafka', # TODO
    #group        => 'kafka', # TODO
    pre_start     => '
      # Cloud init should populate a file with the exhibitor elb url 
      # to use for this kafka instance
      EXHIBITOR_URL_PATH=/var/log/kafka/exhibitor_url
      export EXHIBITOR_URL=$(cat "$EXHIBITOR_URL_PATH")

      if [ -z "$EXHIBITOR_URL" ]; then
        echo "Could not find exhibitor url at $EXHIBITOR_URL_PATH"
        exit -1
      fi

      CLUSTER_LIST_API_URL="$EXHIBITOR_URL/v1/cluster/list"
      ZK_CONNECT=$(curl -s "$CLUSTER_LIST_API_URL" | /usr/local/bin/zookeeper_connect.py)

      if [ -z "ZK_CONNECT" ]; then
        echo "Could not retrieve cluster node list from exhibitor at $CLUSTER_LIST_API_URL"
        exit -2
      fi
      
      sed -i "/^zookeeper\.connect=/c\zookeeper.connect=$ZK_CONNECT" $DAEMON_OPTS
    ',
    script        => '
      ulimit -n 65636
      ulimit -s 10240
      ulimit -c unlimited
      exec $DAEMON $DAEMON_OPTS >> /var/log/kafka/server.log 2>&1
    ',
    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=9990"',
      'DAEMON'         => '/opt/kafka/bin/kafka-server-start.sh',
      'DAEMON_OPTS'    => '/opt/kafka/config/server.properties',
    },
  }
}