Как коммитить только одно сообщение в kafka-python?

Question

[LXSTVAYNE]

Отложенные задачи помещаются в кафку, отдельный контейнер берёт доступные сообщения и запускает отложенные задачи в фоне, как только задача выполнилась, коммитится офсет, но проблема в том, что не факт, что задачи идут в порядке очереди, поэтому при коммите офсета могут закомититься задачи, которые не были ещё выполнены, соответственно при следующем перезапуске они потеряются. Надеюсь, вы меня поняли.

Вызываю следующий код при коммите офсета:

def commit_message(self, msg):
    offsets = {TopicPartition(msg.topic, msg.partition): OffsetAndMetadata(msg.offset + 1, None)}
    self._consumer.commit(offsets)

Вопрос, как мне коммитить только одно сообщение или, если я делаю что-то не так, как перестроить логику, чтобы сообщения не терялись при перезапуске?

Comments

[Сергей Горностаев]

В учебнике O’Reilly описывается такая проблема и способ её решения.

[LXSTVAYNE]

Сергей Горностаев, большое спасибо, займусь изучением.

Answer 1

[mayton2019]

Проблема не очень понятна. Но вы можете сделать джобы полностью независимыми создав для каждого класса джобов свой топик.

Comments

[LXSTVAYNE]

В кафку приходит json сообщение со след структурой:

message = {"text": "Hello World", "datetime": "some_date_time_value"}

Сервис рассылки берёт сообщение и запускает в фоне. Дело в том, что каждое сообщение может идти не в порядке очереди рассылки, даты могут быть самые разные. Я решил эту проблему добавив хранилище для джобов, но, интересно, можно ли одной лишь кафкой обойтись?