日志目錄

• ${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