Kafka Log4j appender 不发送消息

Question

我对 apache Kafka 和 log4j 很陌生。我正在尝试将我的日志消息发送到 Kafka。这是我的 log4j 属性文件

log4j.rootLogger=INFO, stdout

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L %% %m%n

log4j.appender.KAFKA=kafka.producer.KafkaLog4jAppender
log4j.appender.KAFKA.BrokerList=localhost:9092
log4j.appender.KAFKA.Topic=kfkLogs


log4j.appender.KAFKA.SerializerClass=kafka.producer.DefaultStringEncoder
log4j.appender.KAFKA.layout=org.apache.log4j.PatternLayout
log4j.appender.KAFKA.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L %% - %m%n

log4j.logger.logGen=DEBUG, KAFKA

但是，我无法在我的消费者中收到任何消息。我已经用其他一些生产者代码测试了消费者，它工作正常。

另外，我收到此警告

log4j:WARN No such property [serializerClass] in kafka.producer.KafkaLog4jAppender.

编辑

这是生成我的日志消息的代码

    package logGen;

import org.apache.log4j.Logger;

public class TestLog4j {

    static Logger log = Logger.getLogger(TestLog4j.class.getName());

    public static void main(String[] args) {

        log.debug("Debug message");

        log.info("Info message");

        log.error("Error Message");

        log.fatal("Fatal Message");

        log.warn("Warn Message");

        log.trace("Trace Message");
    }

}

此外，如果我使用类似

的方式将日志消息写入文件

log4j.appender.KAFKA=org.apache.log4j.DailyRollingFileAppender
log4j.appender.KAFKA.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.KAFKA.File=logs/server.log

我可以在 server.log 文件中看到日志消息

感谢大家的建议。我认为我看到的奇怪行为可能与我的 kafka 设置有关。这是我用来启动 kafka 服务器的 server.properties 文件的内容。你能看出它有什么奇怪的地方吗？

broker.id=0

port=9092

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

log.dirs=/Users/xyz/kafka/kafka-logs

num.partitions=1

num.recovery.threads.per.data.dir=1

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

log.cleaner.enable=false

zookeeper.connect=localhost:2181

zookeeper.connection.timeout.ms=6000

delete.topic.enable=true

Answer 1

您需要像这样将 KAFKA 添加到您的 log4j.rootLogger：

log4j.rootLogger=INFO, stdout, KAFKA

这会将 KAFKA appender 添加到您的 rootLogger。

Answer 2

我必须指定

log4j.appender.KAFKA.producer.type=异步

log4j.logger.logGen.TestLog4j=TRACE, KAFKA

这似乎行得通。但是我遇到了 10-30 秒的延迟。具体来说，如果我现在发布并且可以在消费者中看到消息，那么下次我发布时必须在大约 30 秒后，否则我在我的消费者中看不到任何东西。关于为什么会发生这种情况的任何想法？可能是 eclispe 问题？

Answer 3

log4j.appender.KAFKA.SerializerClass=kafka.producer.DefaultStringEncoder

你能试试吗

log4j.appender.KAFKA.Serializer=kafka.producer.DefaultStringEncoder

相反？

我相信发送消息的异步模式是通过批处理来实现的，因此会延迟，您是否尝试过使用同步发送？

Answer 4

我查看了 KafkaLog4jAppender.scala 的源代码，这里是 Kafka 的 valid 和 exhaustive 属性从版本 0.8.2.1 开始的 log4j appender：主题、brokerList、compressionType、requiredNumAcks、syncSend。

对我有用的log4j.properties如下：

log4j.rootLogger=ERROR, stdout

log4j.logger.logGen=DEBUG, KAFKA

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L %% %m%n

log4j.appender.KAFKA=kafka.producer.KafkaLog4jAppender
log4j.appender.KAFKA.topic=LOG
log4j.appender.KAFKA.brokerList=localhost:9092
log4j.appender.KAFKA.compressionType=none
log4j.appender.KAFKA.requiredNumAcks=0
log4j.appender.KAFKA.syncSend=true
log4j.appender.KAFKA.layout=org.apache.log4j.PatternLayout
log4j.appender.KAFKA.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L %% - %m%n

Kafka Log4j appender 不发送消息

Kafka Log4j appender not sending messages

java

log4j

apache-kafka