分析记录一次线上问题

发现有个别的topic 数据过期不清理 topic retention设置是10d server等日志也没发现异常日志

现象
    ●磁盘使用率告警 查看磁盘分区大小 发现了几个特别大

    ●去分区下面查看segment 发现有超过10d前的都还保留topic/partition/segement

    ●查看server日志发现从最老segment的那天开始就没有deletion日志了

    ●有新的segment生成

    ●log clean线程还在

Topic的清理策略

cleanup.policy: delete

kafka log的清理策略有两种：delete,compact, 默认是delete

DUMP文件

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files 0000000000001190.timeindex --print-data-log
或者
bin/kafka-dump-log --files 00000000001895067862.timeindex --print-data-log

timestamp: 1859915778000000 offset: 1996
timestamp: 1859915778000003 offset: 2083

这些时间戳转换后时间是个未来的非常大的时间！！！

修复方式见下文：

sh bin/kafka-delete-records.sh --bootstrap-server xxxx --offset-json-file config/offset-
json-file.json

offset-json-file.json

{"partitions":
[{"topic": "test1", "partition": 0,
"offset": 1024}],
"version":1
}

# 从最开始的地方删除消息到 1024的offset

问题分析

概述

Kafka从0.10.0.0版本起，在消息内新增加了个timestamp字段，
在Kafka 0.10.1.0以前(不包含0.10.1.0)，对于一个Topic而言，其Log Segment是由一个.log文档和一个.index文档组合而成，分别用来存储具体的消息数据和对应的偏移量
从Kafka 0.10.1.0开始，对于日志文档，新增一个.timeindex文档，即每个Segment分别由.log、.index和.timeindex这三个文档组成。

kafka时间戳的作用

基于时间戳的日志切分策略
基于时间戳的日志清除策略
根据时间戳来定位消息：之前的索引文件是根据offset信息的，从逻辑语义上并不方便使用，引入了时间戳之后，Kafka支持根据时间戳来查找定位消息

kafka 参数

message.timestamp.type: 定义消息中的时间戳是消息创建时间还是日志附加时间。
值应该是“CreateTime”或“LogAppendTime”, 默认为：CreateTime

LogAppendTime
为 ingestion-time 不可以自定义指定时间，系统自动生成当前producer时间

CreateTime event-time
可以自定义自定timestamp，如果不指定则默认为系统自动生成当前producer时间，即LogAppendTime时间

如下：

如果为CreateTime则会生效，若为LogAppendTime则不会生效
producer.send(new ProducerRecord<String, String> ("test.5",null,1000000000000L,null,value)).get();
# 1000000000000L为timestamp

配置方式

1）在broker的配置文件指定

message.timestamp.type=CreateTime
或
message.timestamp.type=LogAppendTime

2）创建topic时指定

kafka-topics.sh --zookeeper 127.0.0.1:2181/kafka \
--create \
--topic test.4 \
--partitions 1 --replication-factor 1 \
--config message.timestamp.type=CreateTime
或
kafka-topics.sh --zookeeper 127.0.0.1:2181/kafka \
--create \
--topic test.4 \
--partitions 1 --replication-factor 1 \
--config message.timestamp.type=LogAppendTime

message.timestamp.difference.max.ms:
broker接收消息时所允许的时间戳与消息中指定的时间戳之间的最大差异。

    如果message.timestamp.type=CreateTime，则如果时间戳的差异超过此阈值，则将拒绝消息。
    如果message.timestamp.type=LogAppendTime，则忽略此配置。默认为Long.MAX_VALUE

关于从ConsumerRecord获取时间戳为-1的说明
部分源码为：
ConsumerRecord.java

/**
* Creates a record to be received from a specified topic and partition (provided for
* compatibility with Kafka 0.9 before the message format supported timestamps and before
* serialized metadata were exposed).
*
* @param topic The topic this record is received from
* @param partition The partition of the topic this record is received from
* @param offset The offset of this record in the corresponding Kafka partition
* @param key The key of the record, if one exists (null is allowed)
* @param value The record contents
*/
public ConsumerRecord(String topic,
int partition,
long offset,
K key,
V value) {
this(topic, partition, offset, NO_TIMESTAMP, TimestampType.NO_TIMESTAMP_TYPE,
NULL_CHECKSUM, NULL_SIZE, NULL_SIZE, key, value);
}

RecordBatch.java

public interface RecordBatch extends Iterable<Record> {

    /**
    * The "magic" values
    */
    byte MAGIC_VALUE_V0 = 0;
    byte MAGIC_VALUE_V1 = 1;
    byte MAGIC_VALUE_V2 = 2;

    /**
     * The current "magic" value
     */
    byte CURRENT_MAGIC_VALUE = MAGIC_VALUE_V2;

    /**
     * Timestamp value for records without a timestamp
     */
    long NO_TIMESTAMP = -1L;
    ...

TimestampType.java

public enum TimestampType {
    NO_TIMESTAMP_TYPE(-1, "NoTimestampType"), CREATE_TIME(0, "CreateTime"), LOG_APPEND_TIME(1, "LogAppendTime");
    ...

为了兼容Kafka 0.9（包含）以前的版本，如果为kafka0.9之前的无时间戳的版本，则返回的timestamp则为 -1

免责申明：

本文系转载，版权归原作者所有，如若侵权请联系我们进行删除！

《数据治理行业实践白皮书》下载地址：https://fs80.cn/4w2atu

《数栈V6.0产品白皮书》下载地址：https://fs80.cn/cw0iw1

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs

同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack