从最大偏移量读取Kafka消费属性
我已经编写了一个Java Kafka消费者。我想确定如何明确确保一旦Kafka消费者启动，它只读取从那时起由制作人发送的消息，即它不应读取制作人已发送给Kafka的任何消息。有人能解释一下如何确保这一点吗
这是我使用的属性的片段
Properties properties = new Properties();
properties.put("zookeeper.connect", zookeeperHost);
properties.put("group.id", group);
properties.put("auto.offset.reset","largest");
ConsumerConfig consumerConfig = new ConsumerConfig(properties);
consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);
更新9月14日：
我使用的是以下属性，似乎消费者有时仍然从一开始就阅读，有人能告诉我现在出了什么问题吗？
我使用Kafka版本0.8.2
properties.put("auto.offset.reset","largest");
properties.put("auto.commit.enable","false");
共有1个答案
根据上面的回答，似乎正确的机制是如下设置消费者的属性：
properties.put("auto.offset.reset","largest");
properties.put("auto.commit.enable","false");
这样可以确保从最大偏移量读取
-
相反，我需要做的是将更改为新的内容，然后它将从最早的偏移量恢复。 会不会有其他的犯罪行为？ 更新 根据我的理解，这看起来像是每次auto commit enable为false时，它都将提交偏移量。这是Camel Kafka组件的一个特性，因为即使启用了自动提交，它也将在x条消息之后同步
-
如有任何帮助，我们将不胜感激。
-
我对非常陌生，我们正在使用。 我需要做的是使用来自主题的消息。为此，我必须用Java编写一个消费者，它将消费来自主题的消息，然后将该消息保存到数据库。保存消息后，将向Java消费者发送一些确认。如果确认为true，则应使用主题中的下一条消息。如果AcknowlDement为false（这意味着由于某些错误消息，从主题读取的信息无法保存到数据库中），则应再次读取该消息。 我认为我需要使用＜code＞
-
我已经开始让我的制作人向Kafka发送数据，也让我的消费者提取相同的数据。当我在ApacheNIFI中使用ConsumerKafka处理器（kafka版本1.0）时，我脑海中很少有与kafka consumer相关的查询。 Q.1）当我第一次启动ConsumeKafka处理器时，我如何从开始和当前消息中读取消息？ 问题2）以及在Kafka消费者关闭的情况下，如何在最后一条消费信息之后阅读信息？ 在
-
我的用例是使用kafka消费者api，这样我们就可以从kafka主题中手动读取最后一次成功处理的数据的偏移量，然后手动确认Kafka的成功处理数据。（这是为了减少数据丢失）。然而，在我当前的实现中，程序向前移动并从下一个偏移读取，即使我注释掉了“ack.acknowledge（）”。我是新来的Kafka和实现我的消费者下面的方式（我们使用Spring引导） 问题是：即使我注释掉ack.acknow
-
我正在使用Kafka2.0版和java消费者API来消费来自一个主题的消息。我们使用的是一个单节点Kafka服务器，每个分区有一个使用者。我注意到消费者正在丢失一些消息。场景是：消费者投票主题。我为每个线程创建了一个消费者。获取消息并将其交给处理程序来处理消息。然后使用“至少一次”的Kafka消费者语义来提交Kafka偏移量来提交偏移量。同时，我有另一个消费者使用不同的group-id运行。在这个