有没有一种方法可以从一个主题中删除所有的数据，或者在每次运行之前删除主题？

Question

有没有一种方法可以从一个主题中删除所有的数据，或者在每次运行之前删除主题？

我可以修改kafkaconfig.scala文件以更改logretentionhours属性吗？有没有一种方法可以在消费者阅读后立即删除这些消息？

我正在使用生产者从某个地方获取数据，并将数据发送到消费者消费的特定主题，我可以在每次运行时删除该主题中的所有数据吗？我只想在主题中每次都有新的数据。有什么方法可以重新初始化主题吗？

Answer 1

<罢工> 还不认为它得到支持。 看看这个JIRA问题“添加删除主题支持”。

手动删除：

1. 关闭群集
2. 清除kafka日志目录（由kafka配置文件中的log.dir属性指定）以及zookeeper数据
3. 重新启动群集

null

另外，是否有一种方法，当用户阅读消息时，消息就会被删除？

从Kafaka的文件：

Kafka集群在一段可配置的时间内保留所有已发布的消息（无论它们是否已被使用）。例如，如果将日志保留设置为两天，那么在消息发布后的两天内，该消息可供使用，之后将丢弃该消息以释放空间。Kafka的性能相对于数据大小是有效恒定的，因此保留大量数据不是问题。

您还可以在那里找到用于管理消费者端的偏移量的示例代码。

    public static long getLastOffset(SimpleConsumer consumer, String topic, int partition,
                                 long whichTime, String clientName) {
    TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
    Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
    requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
    kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(),clientName);
    OffsetResponse response = consumer.getOffsetsBefore(request);

    if (response.hasError()) {
        System.out.println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition) );
        return 0;
    }
    long[] offsets = response.offsets(topic, partition);
    return offsets[0];
}