Kafka保留策略未按预期工作

Question

我想为我们拥有的一些用例实现数据回放，为此，我需要使用Kafka保留策略（我使用的是连接，我需要窗口时间才能准确）。附言：我正在使用Kafka版本0.10.1.1

我将数据发送到主题中，如下所示：

 kafkaProducer.send(
                    new ProducerRecord<>(kafkaTopic, 0, (long) r.get("date_time") ,r.get(keyFieldName).toString(), r)
            );

我创建的主题如下：

Kafka主题--创建--zookeeper localhost:2181--复制因子1--分区1--主题myTopic
Kafka主题--zookeeper localhost--alter--主题myTopic--配置保留。ms=17280000Kafka主题--zookeeper localhost--alter--主题myTopic--配置段。ms=17280000

所以通过上面的设置，我应该将我的主题的保留时间设置为48小时。

我扩展TimestampExtractor以记录每条消息的实际时间。

public class ConsumerRecordOrWallclockTimestampExtractor implements TimestampExtractor {
    private static final Logger LOG = LoggerFactory.getLogger(ConsumerRecordOrWallclockTimestampExtractor.class);
    @Override
    public long extract(ConsumerRecord<Object, Object> consumerRecord) {
        LOG.info("TIMESTAMP : " + consumerRecord.timestamp() + " - Human readable : " + new Date(consumerRecord.timestamp()));
        return consumerRecord.timestamp() >= 0.1 ? consumerRecord.timestamp() : System.currentTimeMillis();
    }
}

为了测试，我向我的主题发送了4条消息，我收到了这4条日志消息。

2017-02-28 10:23:39信息消费者记录墙时钟时间提取器：21-时间戳：1488295086292人类可读-2017年2月28日星期二10:18:06 EST 2017
2017-02-28 10:24:01信息消费者记录墙时钟时间提取器：21-时间戳：1483272000000人类可读-2017年1月1日07:00:00 EST
2017-02-28 10:26:11信息消费者记录墙时钟时间提取器：21-时间戳：1485820800000人类可读性-周一2017年1月30日19:00:00 EST 2017-02-28 10:27:22信息消费者记录墙时钟时间提取器：21-时间戳：1488295604411人类可读性-周二2017年2月28日10:26:44 EST

因此，根据Kafka的保留政策，我希望看到我的两条消息在5分钟后被清除/删除（第2条和第3条消息，因为它们是1月1日和1月30日的消息）。但是我试着花了一个小时来阅读我的主题，每次我阅读完我的主题，我就收到了所有的4条信息。

kafka avro控制台消费者——zookeeper本地主机：2181——从一开始——主题myTopic

我的Kafka配置如下：

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion
log.retention.hours=168

# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
# segments don't drop below log.retention.bytes.
#log.retention.bytes=1073741824

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

我是做错了什么还是错过了什么？

Answer 1

Kafka通过删除日志段来实现其保留策略。Kafka从不删除活动段，即它将在其中附加发送到分区的新消息的段。Kafka只删除旧片段。当新消息发送到分区时，Kafka将活动段滚动到旧段，并且

• 带有新消息的活动段的大小将超过log.segment。字节，或
• 活动段中第一条消息的时间戳早于log.roll。ms（默认为7天）

因此，在您的示例中，您必须在美国东部标准时间2017年2月2日28 10:18:06后等待7天，发送一条新消息，然后所有4条初始消息都将被删除。