kafka 关于设置消息存放周期_kafka设置消息保存时间

kafka设置消息存放时间有两种配置

1. 通过kafka server.properties配置文件

log.retention.hours=8
1

2. 新建topic的时候添加参数

from confluent_kafka.admin import AdminClient, NewTopic

# Configure Kafka admin client
admin_conf = {'bootstrap.servers': 'localhost:9092'}
admin_client = AdminClient(admin_conf)

# Define topic configuration
topic_config = {
    'retention.ms': '604800000'  # 7 days
}

# Define topic properties
topic_properties = NewTopic(
    topic='my-topic',
    num_partitions=1,
    replication_factor=1,
    config=topic_config
)

# Create topic
admin_client.create_topics([topic_properties])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

3. 假如broker配置的retention和topic的设置的retention不一致
   如果topic的retention时间比broker log的时间小，kafka仍然会删除消息，即使topic的retention没到期。换句话说kafka会取二者的最小值设置消息的生命周期。

比如topic设置消息保存时间是7天，broker设置保存1天。这种情况下，kafka会在一天后删除消息

为了避免消息在topic设置的到期之前被删除，至少需要保证broker的log retention时间大于等于topic的retention时间