日志目录

• ${kafka.logs.dir}/server.log ：服务器日志
• ${kafka.logs.dir}/state-change.log：状态变化日志
• ${kafka.logs.dir}/kafka-request.log：请求处理日志
• ${kafka.logs.dir}/log-cleaner.log：日志清理
• ${kafka.logs.dir}/controller.log：管理日志
• ${kafka.logs.dir}/kafka-authorizer.log：认证日志

log4j.properties 文件

log4j.properties 是 kafka 操作日志配置文件

log4j.properties 文件的一些说明

# 日志级别覆盖规则  优先级：ALL < DEBUG < INFO <WARN < ERROR < FATAL < OFF
# 1.子日志log4j.logger会覆盖主日志log4j.rootLogger，这里设置的是日志输出级别，Threshold设置appender的日志接收级别；
# 2.log4j.logger级别低于Threshold,appender接收级别取决于Threshold级别；
# 3.log4j.logger级别高于Threshold,appender接收级别取决于log4j.logger级别，因为输出里就没有Threshold要求的日志；
# 4.子logger设置，主要与rootLogger区分开打印日志 一般与log4j.additivity配合使用
# log4j.additivity 是否继承父Logger的输出源（appender),默认是true 
# true 在stdout, kafkaAppender里输出 也会在stateChangeAppender输出
# 这里需要单独输出 所以设置为false 只会在stateChangeAppender输出
# log4j.logger后面如果没有appender，则默认使用log4j.rootLogger后面设置的appender
# 如果使用org.apache.log4j.RollingFileAppender  可以使用MaxFileSize设置最大文件大小  MaxBackupIndex设置最大文件数量

config/log4j.properties中日志的级别设置有的是TRACE，在长时间运行过程中产生的日志大小吓人，所以如果没有特殊需求，强烈建议将其更改成INFO级别。

具体修改方法如下所示，将config/log4j.properties文件中最后的几行中的TRACE改成INFO或INFO以上

# 主日志设置 
log4j.rootLogger=ERROR, stdout, kafkaAppender# 控制台的appender和layout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n# kafkaAppender的appender和layout
log4j.appender.kafkaAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.kafkaAppender.File=${kafka.logs.dir}/server.log
log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n# 状态变化日志
log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.stateChangeAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log
log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n# 请求处理日志
log4j.appender.requestAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.requestAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.requestAppender.File=${kafka.logs.dir}/kafka-request.log
log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n# 日志清理格式
log4j.appender.cleanerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.cleanerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.cleanerAppender.File=${kafka.logs.dir}/log-cleaner.log
log4j.appender.cleanerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.cleanerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n# controller 日志格式
log4j.appender.controllerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.controllerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.controllerAppender.File=${kafka.logs.dir}/controller.log
log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n# authorizer日志格式
log4j.appender.authorizerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.authorizerAppender.File=${kafka.logs.dir}/kafka-authorizer.log
log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n######################################################
# zookeeper
log4j.logger.org.apache.zookeeper=INFO# kafka
log4j.logger.kafka=INFO# org.apache.kafka
log4j.logger.org.apache.kafka=INFO# 请求处理日志
log4j.logger.kafka.request.logger=INFO, requestAppender
log4j.additivity.kafka.request.logger=false
# log4j.logger.kafka.network.Processor=INFO, requestAppender
# log4j.additivity.kafka.network.Processor=false  # 配置文件有的没有
# log4j.logger.kafka.server.KafkaApis=INFO, requestAppender
log4j.additivity.kafka.server.KafkaApis=false
log4j.logger.kafka.network.RequestChannel$=INFO, requestAppender
log4j.additivity.kafka.network.RequestChannel$=false# controller
log4j.logger.kafka.controller=INFO, controllerAppender
log4j.additivity.kafka.controller=false#kafka-logs清理
log4j.logger.kafka.log.LogCleaner=INFO, cleanerAppender
log4j.additivity.kafka.log.LogCleaner=false#状态变化日志
# log4j.logger.state.change.logger=INFO, stateChangeAppender
# log4j.additivity.state.change.logger=false#authorizer
log4j.logger.kafka.authorizer.logger=INFO, authorizerAppender
log4j.additivity.kafka.authorizer.logger=false

server.properties 文件

kafka 数据

在server.properties中配置了log.dirs值，其实表示的是kafka数据的存放目录，而非日志目录

log.dirs=/data/kafka

kafka 日志清理

Kafka日志管理器允许定制删除策略。目前的策略是删除修改时间在N天之前的日志（按时间删除），也可以使用另外一个策略：保留最后的N GB数据的策略(按大小删除)。为了避免在删除时阻塞读操作，采用了copy-on-write形式的实现，删除操作进行时，读取操作的二分查找功能实际是在一个静态的快照副本上进行的，这类似于Java的CopyOnWriteArrayList。

Kafka消费日志删除思想：Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用

log.cleanup.policy=delete启用删除策略
# 直接删除，删除后的消息不可恢复。可配置以下两个策略：
# 清理超过指定时间清理： 
log.retention.hours=168
# 超过指定大小后，删除旧的消息：
log.retention.bytes=1073741824

也可以写个定时清理脚本来清除日志结合

crontab 脚本

0 0 2 \* \* ? /home/cluster/kafka211/bin/cleanupkafkalog.sh

清理脚本

#!/bin/bash# log dir
logDir=/log/kafka
#keep 60 file
count=60
count=$[$count+1]
LOGNUM=`ls -l /home/cluster/kafka211/logs/server.log.* |wc -l`
if [ $LOGNUM -gt 0 ]; thenls -t $logDir/server.log.* | tail -n +$count | xargs rm -f
fi#kafkaServer.out 
if [ -e "$logDir/kafkaServer.out" ]; thenrm -f /home/cluster/kafka211/logs/kafkaServer.out
fi

压缩策略

只保留每个key最后一个版本的数据。

log.cleaner.enable=true    # 首先在broker的配置中设置启用cleaner，这个默认是关闭的。
log.cleanup.policy=compact # 在Topic的配置中设置，启用压缩策略。

bin/kafka-run-class.sh 文件

GC 日志

kafka默认打印GC日志，如下，

$ ls
kafka-authorizer.log          kafkaServer-gc.log.3  kafkaServer-gc.log.8      server.log.2018-10-22-14
kafka-request.log             kafkaServer-gc.log.4  kafkaServer-gc.log.9      server.log.2018-10-22-15
kafkaServer-gc.log.0          kafkaServer-gc.log.5  kafkaServer.out
kafkaServer-gc.log.1          kafkaServer-gc.log.6  server.log
kafkaServer-gc.log.2.current  kafkaServer-gc.log.7  server.log.2018-10-22-13

生产环境不需要GC。修改 bin/kafka-run-class.sh脚本,将参数 KAFKA_GC_LOG_OPTS=" " 设置为空格即可，重启kafka之后就不再打印GC日志了。

[cluster@PCS102 bin]$ vim kafka-run-class.shGC_FILE_SUFFIX='-gc.log'
GC_LOG_FILE_NAME=''
if [ "x$GC_LOG_ENABLED" = "xtrue" ]; thenGC_LOG_FILE_NAME=$DAEMON_NAME$GC_FILE_SUFFIXKAFKA_GC_LOG_OPTS="-Xloggc:$LOG_DIR/$GC_LOG_FILE_NAME -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=100M"
KAFKA_GC_LOG_OPTS=" "
fi

日志保存目录

Kafka运行时日志默认输出到$KAFKA_HOME/logs目录下，容易撑爆分区，造成操作系统崩溃。需要将日志输出到指定分区，比如/var/log目录下
修改 bin/kafka-run-class.sh 脚本增加一行 LOG_DIR="/var/log/kafka"

LOG_DIR=/var/log/kafka  # 我是新增的
# Log directory to use
if [ "x$LOG_DIR" = "x" ]; thenLOG_DIR="$base_dir/logs"
fi