[ https://issues.apache.org/jira/browse/KAFKA-9169?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16975693#comment-16975693 ]
Navinder Brar commented on KAFKA-9169: -------------------------------------- I can confirm that this has been there since 1.1 version. > Standby Tasks point ask for incorrect offsets on resuming post suspension > ------------------------------------------------------------------------- > > Key: KAFKA-9169 > URL: https://issues.apache.org/jira/browse/KAFKA-9169 > Project: Kafka > Issue Type: Bug > Components: streams > Affects Versions: 2.3.0 > Reporter: Navinder Brar > Assignee: John Roesler > Priority: Critical > Fix For: 2.5.0 > > > In versions(check 2.0) where standby tasks are suspended on each rebalance > the checkpoint file is updated post the flush and the expected behaviour is > that post assignment the same standby task gets assigned back on the machine > it will start reading data from changelog from the same offset from it left > off. > > But there looks like a bug in the code, every time post rebalance it starts > reading from the offset from where it read the first time the task was > assigned on this machine. This has 2 repercussions: > # After every rebalance the standby tasks start restoring huge amount of > data which they have already restored earlier(Verified this via 300x increase > Network IO on all streams instances post rebalance even when no change in > assignment) . > # If changelog has time retention those offsets will not be available in the > changelog, which leads to offsetOutOfRange exceptions and the stores get > deleted and recreated again. > > I have gone through the code and I think I know the issue. > In TaskManager# updateNewAndRestoringTasks(), the function > assignStandbyPartitions() gets called for all the running standby tasks where > it populates the Map: checkpointedOffsets from the > standbyTask.checkpointedOffsets() which is only updated at the time of > initialization of a StandbyTask(i.e. in it's constructor). > > This has an easy fix. > Post resumption we are reading standbyTask.checkpointedOffsets() to know the > offset from where the standby task should start running and not from > stateMgr.checkpointed() which gets updated on every commit to the checkpoint > file. In the former case it's always reading from the same offset, even those > which it had already read earlier and in cases where changelog topic has a > retention time, it gives offsetOutOfRange exception. So, > standbyTask.checkpointedOffsets() is quite useless and we should use > stateMgr.checkpointed() instead to return offsets to task manager. -- This message was sent by Atlassian Jira (v8.3.4#803005)