If a standby task is suspended, it will write the checkpoint file again after flushing its state stores, and when it resumes it does not re initialize the position on the consumer and hence it is still the task-manager's responsibility to set the right starting offset from the latest checkpoint file. If we did not do that, that should still be a bug.
Guozhang On Sat, Nov 9, 2019 at 11:33 AM Navinder Brar <navinder_b...@yahoo.com> wrote: > Hi Guozhang, > > Thanks for the reply. > > So, if I understand it correctly. In versions where KIP-429 was not > implemented and when we were suspending the standby tasks during rebalance > and they were resumed post rebalance, they will be reading from the > beginning of the offsets of changelog, since the will be reading from > standbyTask.checkpointedOffsets() > which was only updated during the first initialization. > > Regards, > Navinder > > On Sunday, 10 November, 2019, 12:50:39 am IST, Guozhang Wang < > wangg...@gmail.com> wrote: > > > Hello Navinder, > > Sorry for the late reply and thanks for bringing this up. I think this is > indeed a bug that needs to be fixed. > > The rationale behind was the following: for restoring active tasks and > processing standby tasks, we are using the same consumer client within the > thread (the restoreConsumer). And before ALL of the active tasks have > completed restoration, the consumer would not get assigned to any of the > standby tasks at all. So in a timeline it should be looking like this with > a rebalance assuming KIP-429 is already in place: > > T0: rebalance triggered, some tasks gets revoked but some others may still > be active; > T0-T1: a subset of active tasks (via the main consumer) and all standby > tasks (via the restore consumer) are still processing; > T1: rebalance finished, some new tasks gets assigned, and now needs to be > restored. Restore consumer re-assign to fetch from those restoring consumer > only. > T1-T2: the main consumer paused all partitions, hence no active tasks > processing; also restore consumer only fetching for restoring tasks, and > hence no standby tasks processing; > T2: restoration completed, restore consumer reassigned to those standby > tasks. > > Note in T1, the standby tasks are all still "running" but they just do not > proceed any more since the consumer has switched to fetch other partitions; > so at T2 when the consumer switch back it should just resume from where it > has switched off. > > > Guozhang > > > On Mon, Nov 4, 2019 at 4:47 AM Navinder Brar > <navinder_b...@yahoo.com.invalid> wrote: > > > Hi, > > Please let me know if this is not the correct forum to ask this. But I > > have a doubt, I was hoping someone can clear it for me. > > 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). I have checked > > and this goes way to 1.1 version when the rebalance protocol was old and > > standby tasks were suspended during rebalance and then resumed on > > assignment. > > I want to know, why post resumption we were/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. > > Regards, > > Navinder > > > > > -- > -- Guozhang > > -- -- Guozhang
