Thanks Guozhang.
The jira is filed: [KAFKA-9169] Standby Tasks point ask for incorrect offsets
on resuming post suspension - ASF JIRA
|
|
| |
[KAFKA-9169] Standby Tasks point ask for incorrect offsets on resuming p...
|
|
|
On Monday, 11 November, 2019, 03:10:37 am IST, Guozhang Wang
<wangg...@gmail.com> wrote:
Could you file a JIRA report for this so that we can keep track of it and fix?
Guozhang
On Sun, Nov 10, 2019 at 1:39 PM Guozhang Wang <wangg...@gmail.com> wrote:
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
--
-- Guozhang
