python - 在 confluence-kafka-python 中设置主题日志保留

Question

我在文档中找不到如何在使用 confluence-kafka 创建生产者时设置保留时间。

如果我只指定“bootstrap-servers”，则默认保留时间为 1 天。我希望能够改变这一点。

(我想在 python API 中而不是在命令行上执行此操作。)

Answer 1

保留时间是在您创建主题时设置的，而不是在生产者配置上设置的。

如果您的server.properties允许自动创建主题，然后您将获得其中设置的默认值。

否则，您可以使用 AdminClient API发送 NewTopic支持 config 的请求dict<str,str> 的属性

from confluent_kafka.admin import AdminClient, NewTopic

# a = AdminClient(...) 

topics = list()
t = NewTopic(topic, num_partitions=3, replication_factor=1, config={'log.retention.hours': '168'})
topics.append(t)

# Call create_topics to asynchronously create topics, a dict
# of <topic,future> is returned.
fs = a.create_topics(topics)

# Wait for operation to finish.
# Timeouts are preferably controlled by passing request_timeout=15.0
# to the create_topics() call.
# All futures will finish at the same time.
for topic, f in fs.items():
    try:
        f.result()  # The result itself is None
        print("Topic {} created".format(topic))
    except Exception as e:
        print("Failed to create topic {}: {}".format(topic, e))

在同一链接中，您可以找到更改主题请求