confluentinc / confluentinc/confluent-kafka-python

manual checkpoint

Open
#1,617 0 comments 0 reactions 0 assignees View on GitHub
question
Dominant language
Python
Stars
509
Forks
964
Avg merge
2d 2h
Merged PRs (30d)
14

Description

Description
===========
I am using 2 process to read from kafka and write to destination (design of ELT tool) hence, I would like to maintain checkpoint/offset manually and read it from where it left on last run . however, it is reading old record i.e last checkpoint record.

How to reproduce
================

def _reset_offset_(self, consumer, partitions):
"""
set the topic starting point
"""
for p in partitions:
name = str(p.partition)
if name in self.state_update:
p.offset = self.state_update[name] ## from checkpoint lookup
else:
p.offset = OFFSET_BEGINNING
consumer.assign(partitions)

self.consumer.subscribe([self.topic], on_assign=self._reset_offset_)

It always return last read record on next run. I would like to skip and read only next record from last checkpoint.

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.