如何制作可重启的生产者? [英] How to make restart-able producer?

问题描述

最新版本的 kafka 支持恰好一次语义(EoS).为了支持这个概念，每条消息都添加了额外的细节.这意味着在您的消费者；如果您打印消息的偏移量，它们不一定是连续的.这使得轮询主题以读取最后提交的消息变得更加困难.

Latest version of kafka support exactly-once-semantics (EoS). To support this notion, extra details are added to each message. This means that at your consumer; if you print offsets of messages they won't be necessarily sequential. This makes harder to poll a topic to read the last committed message.

就我而言，消费者打印了这样的东西

In my case, consumer printed something like this

Offset-0 0
Offset-2 1
Offset-4 2

问题:为了编写可重启的骄傲器；我轮询主题并阅读最后一条消息的内容.在这种情况下;最后一条消息将是 offset#5，这不是有效的消费者记录.因此，我在代码中看到了错误.

Problem: In order to write restart-able proudcer; I poll the topic and read the content of last message. In this case; last message would be offset#5 which is not a valid consumer record. Hence, I see errors in my code.

我可以使用以下提供的解决方案:获取最后一个发送到 kafka 主题的消息.唯一的问题是，而不是使用 consumer.seek(partition, last_offset=1);我会使用consumer.seek(partition, last_offset-2).这可以立即解决我的问题，但不是理想的解决方案.

I can use the solution provided at : Getting the last message sent to a kafka topic. The only problem is that instead of using consumer.seek(partition, last_offset=1); I would use consumer.seek(partition, last_offset-2). This can immediately resolve my issue, but it's not an ideal solution.

为用 Java 编写的消费者获取最后提交的消息的最可靠和最佳解决方案是什么?或

是否可以使用本地状态存储用于分区?或

Is it possible to use local state-store for a partition? OR

最推荐的存储最后一条消息以承受生产者失败的方法是什么?或

kafka 连接器可以重启吗?是否有任何特定的 API 可用于使生产者可重新启动?

仅供参考 - 我不是在寻找快速修复

FYI- I am not looking for quick fix

推荐答案

就我而言，多个生产者将数据推送到一个大主题.因此，阅读整个主题将是一场噩梦.

In my case, multiple producers push data to one big topic. Therefore, reading entire topic would be nightmare.

我找到的解决方案是维护另一个主题，即P1_Track"，生产者可以在其中存储元数据.在事务中，生产者将数据发送到一个大主题和 P1_Track.

The solution that I found is to maintain another topic i.e. "P1_Track" where producer can store metadata. Within a transaction a producer will send data to one big topic and P1_Track.

当我重新启动生产者时，它会读取 P1_Track 并找出从哪里开始.

When I restart a producer, it will read P1_Track and figure out where to start from.

考虑将最后提交的消息存储在数据库中并在生产者进程重新启动时使用它.

Thinking about storing last committed message in a database and using it when producer process restarts.

这篇关于如何制作可重启的生产者?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！