分析记录一次线上问题
发现有个别的topic 数据过期不清理 topic retention设置是10d server等日志也没发现异常日志
现象
- 磁盘使用率告警 查看磁盘分区大小 发现了几个特别大
- 去分区下面查看segment 发现有超过10d前的都还保留topic/partition/segement
- 查看server日志发现从最老segment的那天开始就没有deletion日志了
- 有新的segment生成
- log clean线程还在
Topic的清理策略
cleanup.policy: deletekafka log的清理策略有两种：delete,compact, 默认是delete
DUMP文件
-
bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files
000000000000119
0.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_V
0
=
0;
-
byte MAGIC_
VALUE_V
1
=
1;
-
byte MAGIC_
VALUE_V
2
=
2;
-
-
/
**
-
* The current
"magic"
value
-
*
/
-
byte CURRENT_MAGIC_
VALUE
= MAGIC_
VALUE_V
2;
-
-
/
**
-
* 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://blog.csdn.net/weishuai90/article/details/128821742