kafka时间戳的详解及使用

分析记录一次线上问题

发现有个别的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

参考：

kafka时间戳的详解及使用

Kafka日志清理策略，compact和delete

Kafka中数据清理、日志压缩、日志删除、保留策略