Kafka高级消费者使用Java API从主题获取所有消息（相当于--from-begind）

Question

我正在使用来自Kafka站点的ConsumerGroupExample代码测试Kafka高级消费者。我想检索我在Kafka服务器配置中拥有的关于名为“测试”的主题的所有现有消息。看看其他博客，Auto.offset.reset应该设置为“最小”，才能获取所有消息：

private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId)    {
    Properties props = new Properties();
    props.put("zookeeper.connect", a_zookeeper);
    props.put("group.id", a_groupId);
    props.put("auto.offset.reset", "smallest");
    props.put("zookeeper.session.timeout.ms", "10000");     

    return new ConsumerConfig(props);
}

Answer 1

基本上，每次新的使用者尝试使用一个主题时，它会从一开始就阅读消息。如果您每次都是出于测试目的从头开始使用，那么每次使用新的groupID初始化您的使用者时，它都将从头开始读取消息。我是这样做的：

properties.put("group.id", UUID.randomUUID().toString());

每次从头读消息！