spring kafka 1.3.9容器级错误时手动提交偏移量
我使用MANUAL_IMMEDIATEack模式，Spring-kafka 1.3.9（不能更改为Java8），并在监听器代码中完成处理时提交偏移量。我使用自定义反序列化器及其工作正常，除非我遇到反序列化异常。然后Kafka卡住了。我已经处理了这个由Deserializer，喜欢而不是抛出异常（当反序列化异常发生）我得到一个反序列化对象的新实例，并设置原始消息（导致反序列化异常）在一个字段（异常数据），让它传递给监听器代码（注释与@KafkaListener）。在监听器代码中，我检查字段异常数据是否为空，并相应地决定“好消息”或“坏消息”，因为我也需要处理“坏消息”。
现在我想知道有没有办法从容器级别处理所有这些。我坚持的地方是我没有在这个版本的Spring-kafka 1.3.9中得到消费者参考。我需要提交偏移错误，任何错误。
共有1个答案
不可能；不可能获得该版本的消费者。在2.0中添加了消费者数据库。
-
我有一个Kafka消费者，我从它消费数据从一个特定的主题，我看到下面的例外。我使用的是Kafka版本。 我添加了这两个额外的消费者属性，但仍然没有帮助： 那个错误意味着什么？我该如何解决它？我需要添加一些其他消费者属性吗？
-
我有一个Kafka接收器任务，通过方法收听Kafka主题 但我不想自动提交偏移量，因为一旦从Kafka取出记录，我就有一些处理逻辑 从Kafka获取记录后，如果处理成功，则只有我想提交偏移量，否则它应该再次从同一偏移量读取。 我可以在Kafka consumer中看到方法，但在中找不到替代方法。
-
我目前正在从具有特定偏移量的主题中获取消息。我正在使用寻求（）来实现它。但是当我将enable.auto.commit设置为true或使用手动同步（委托同步（）/委托同步（））时，Seek（）不起作用，因为它没有轮询来自特定偏移量的消息，而是从最后提交的偏移量中选择。 因此，在使用Seek（）时，是否必须将偏移量存储在外部DB中，而不提交给Kafka？Seek和Commit不能并行工作吗？ 客户端
-
我正在实现spring kafka批处理侦听器，它读取来自kafka主题的消息列表，并将数据发布到REST服务。我想了解在REST服务停止的情况下的偏移管理，不应该提交批处理的偏移，应该为下一次轮询处理消息。我已经阅读了spring kafka文档，但在理解侦听器错误处理程序和批量查找当前容器错误处理程序之间的区别时存在困惑。我使用的是spring-boot-2.0.0。M7及以下版本是我的代码。
-
我试图使用Kafka Utils Api从Kafka（0.10.0.0）到Spark（1.6.0）流媒体应用程序使用数据 Kafka提尔。createDirectStream[String，String，StringDecoder，StringDecoder]（ssc，kafkaParams，InputOpicSet） 要求是将偏移范围手动提交给Kafka本身。 请注意，当在java中使用Kafk
-
我使用的是camel kafka组件，我不清楚在提交补偿时引擎盖下发生了什么。如下所示，我正在聚合记录，我认为对于我的用例来说，只有在记录保存到SFTP后提交偏移量才有意义。 是否可以手动控制何时可以执行提交？