Logging with Kafka

build.sbt

// Use `sbt run` to run the example

name := "kafka-logging"

version := "0.1.0"

scalaVersion := "2.10.3"

sbtVersion := "0.13.0"

libraryDependencies ++= Seq(
  "org.apache.kafka" % "kafka_2.10" % "0.8.1.1"
    exclude("javax.jms", "jms")
    exclude("com.sun.jdmk", "jmxtools")
    exclude("com.sun.jmx", "jmxri")
)

KafkaLogging.scala

package localhost

import Thread.sleep
import org.apache.log4j.Logger.getLogger

object KafkaLogging extends App {
  val logger = getLogger(KafkaLogging.getClass)
  import logger._

  info("Start")
  for (iteration ← 1 to 10) {
    debug("Heavy computations " + iteration)
    sleep(1000)
  }
  info("Stop")
}

log4j.properties

log4j.rootLogger=DEBUG
log4j.logger.kafka=INFO, stdout
log4j.logger.localhost=DEBUG, stdout, kafka

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%-5p [%t]: %m%n

log4j.appender.kafka=kafka.producer.NicerKafkaLog4jAppender
log4j.appender.kafka.BrokerList=localhost:9092
log4j.appender.kafka.Topic=logs
#log4j.appender.kafka.ProducerType=async <- prevents JVM shutdown
log4j.appender.kafka.layout=org.apache.log4j.PatternLayout
log4j.appender.kafka.layout.ConversionPattern=%-5p [%t]: %m

NicerKafkaLog4jAppender.scala

package kafka.producer

import org.apache.log4j.helpers.LogLog
import org.apache.log4j.spi.LoggingEvent

import java.net.ConnectException
import kafka.common.FailedToSendMessageException

class NicerKafkaLog4jAppender extends KafkaLog4jAppender {
  // See [KAFKA-847] bug
  override def requiresLayout = true

  // Prevent Kafka logging failures from breaking client code
  // when in synchronous mode
  override def append(event: LoggingEvent)  {
    try {
      super.append(event)
    } catch {
      case e@ (_: ConnectException | _: FailedToSendMessageException) =>
        LogLog.error("Message lost due to connectivity problems", e)
    }
  }
}