You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Piotr Nowojski <pn...@apache.org> on 2021/03/03 19:47:45 UTC
Re: Job downgrade
Hi,
I'm not sure what's the reason behind this. Probably classes are somehow
attached to the state and this would explain why you are experiencing this
issue. I've asked someone else from the community to chip in, but in the
meantime, can not you just prepare a new "version 1" of the job, with just
some empty `class B` on the class path? Or if this doesn't work, just copy
the whole `class B` from version 2?
Best,
Piotrek
sob., 27 lut 2021 o 19:10 Alexey Trenikhun <ye...@msn.com> napisał(a):
> Hello,
> Let's have version 1 of my job uses keyed state with name "a" and type A,
> which some Avro generated class. Then I upgrade to version 2, which in
> addition uses keyed state "b" and type B (another concrete Avro generated
> class), I take savepoint with version 2 and decided to downgrade to version
> 1 and start with taken savepoint, can I do it? On one hand, version 1
> doesn't have state "b", but seems Flink still tries to create call
> restoreSerializer and it tries to read runtimeType (`class B`) which is
> not available in version 1
>
> Thanks,
> Alexey
>
Re: Job downgrade
Posted by "Tzu-Li (Gordon) Tai" <tz...@apache.org>.
Hi Alexey,
Thanks for confirming.
Can you send me a copy of the exception stack trace? That could help me
pinpoint the exact issue.
Cheers,
Gordon
On Fri, Mar 5, 2021 at 2:02 PM Alexey Trenikhun <ye...@msn.com> wrote:
> Hi Gordon,
> I was using RocksDB backend
> Alexey
>
> ------------------------------
> *From:* Tzu-Li (Gordon) Tai <tz...@apache.org>
> *Sent:* Thursday, March 4, 2021 12:58:01 AM
> *To:* Alexey Trenikhun <ye...@msn.com>
> *Cc:* Piotr Nowojski <pn...@apache.org>; Flink User Mail List <
> user@flink.apache.org>
> *Subject:* Re: Job downgrade
>
> Hi Alexey,
>
> Are you using the heap backend? If that's the case, then for whatever
> state was registered at the time of a savepoint, Flink will attempt to
> restore it to the heap backends.
> This essentially means that state "B" will be read as well, that would
> explain why Flink is trying to locate class B in the classpath.
>
> For this scenario, class B needs to be in the classpath if you downgrade
> back to version 1, with a savepoint taken with version 2 of the job.
>
> - Gordon
>
> On Thu, Mar 4, 2021 at 4:04 AM Alexey Trenikhun <ye...@msn.com> wrote:
>
> If I copy class A into version 1+ it works. But it is the problem from CD
> perspective - I want to introduce feature which required new state: 1st I
> need make version 1+ with class B, but no other changes, then version 2 with
> class B and logic changes, upgrade job and if job doesn’t do what expected
> “rollback” to version 1+.
>
> ------------------------------
> *From:* Piotr Nowojski <pn...@apache.org>
> *Sent:* Wednesday, March 3, 2021 11:47:45 AM
> *To:* Alexey Trenikhun <ye...@msn.com>
> *Cc:* Flink User Mail List <us...@flink.apache.org>
> *Subject:* Re: Job downgrade
>
> Hi,
>
> I'm not sure what's the reason behind this. Probably classes are somehow
> attached to the state and this would explain why you are experiencing this
> issue. I've asked someone else from the community to chip in, but in the
> meantime, can not you just prepare a new "version 1" of the job, with just
> some empty `class B` on the class path? Or if this doesn't work, just copy
> the whole `class B` from version 2?
>
> Best,
> Piotrek
>
> sob., 27 lut 2021 o 19:10 Alexey Trenikhun <ye...@msn.com> napisał(a):
>
> Hello,
> Let's have version 1 of my job uses keyed state with name "a" and type A,
> which some Avro generated class. Then I upgrade to version 2, which in
> addition uses keyed state "b" and type B (another concrete Avro generated
> class), I take savepoint with version 2 and decided to downgrade to version
> 1 and start with taken savepoint, can I do it? On one hand, version 1
> doesn't have state "b", but seems Flink still tries to create call
> restoreSerializer and it tries to read runtimeType (`class B`) which is
> not available in version 1
>
> Thanks,
> Alexey
>
>
Re: Job downgrade
Posted by Alexey Trenikhun <ye...@msn.com>.
Hi Gordon,
I was using RocksDB backend
Alexey
________________________________
From: Tzu-Li (Gordon) Tai <tz...@apache.org>
Sent: Thursday, March 4, 2021 12:58:01 AM
To: Alexey Trenikhun <ye...@msn.com>
Cc: Piotr Nowojski <pn...@apache.org>; Flink User Mail List <us...@flink.apache.org>
Subject: Re: Job downgrade
Hi Alexey,
Are you using the heap backend? If that's the case, then for whatever state was registered at the time of a savepoint, Flink will attempt to restore it to the heap backends.
This essentially means that state "B" will be read as well, that would explain why Flink is trying to locate class B in the classpath.
For this scenario, class B needs to be in the classpath if you downgrade back to version 1, with a savepoint taken with version 2 of the job.
- Gordon
On Thu, Mar 4, 2021 at 4:04 AM Alexey Trenikhun <ye...@msn.com>> wrote:
If I copy class A into version 1+ it works. But it is the problem from CD perspective - I want to introduce feature which required new state: 1st I need make version 1+ with class B, but no other changes, then version 2 with class B and logic changes, upgrade job and if job doesn’t do what expected “rollback” to version 1+.
________________________________
From: Piotr Nowojski <pn...@apache.org>>
Sent: Wednesday, March 3, 2021 11:47:45 AM
To: Alexey Trenikhun <ye...@msn.com>>
Cc: Flink User Mail List <us...@flink.apache.org>>
Subject: Re: Job downgrade
Hi,
I'm not sure what's the reason behind this. Probably classes are somehow attached to the state and this would explain why you are experiencing this issue. I've asked someone else from the community to chip in, but in the meantime, can not you just prepare a new "version 1" of the job, with just some empty `class B` on the class path? Or if this doesn't work, just copy the whole `class B` from version 2?
Best,
Piotrek
sob., 27 lut 2021 o 19:10 Alexey Trenikhun <ye...@msn.com>> napisał(a):
Hello,
Let's have version 1 of my job uses keyed state with name "a" and type A, which some Avro generated class. Then I upgrade to version 2, which in addition uses keyed state "b" and type B (another concrete Avro generated class), I take savepoint with version 2 and decided to downgrade to version 1 and start with taken savepoint, can I do it? On one hand, version 1 doesn't have state "b", but seems Flink still tries to create call restoreSerializer and it tries to read runtimeType (`class B`) which is not available in version 1
Thanks,
Alexey
Re: Job downgrade
Posted by "Tzu-Li (Gordon) Tai" <tz...@apache.org>.
Hi Alexey,
Are you using the heap backend? If that's the case, then for whatever state
was registered at the time of a savepoint, Flink will attempt to restore it
to the heap backends.
This essentially means that state "B" will be read as well, that would
explain why Flink is trying to locate class B in the classpath.
For this scenario, class B needs to be in the classpath if you downgrade
back to version 1, with a savepoint taken with version 2 of the job.
- Gordon
On Thu, Mar 4, 2021 at 4:04 AM Alexey Trenikhun <ye...@msn.com> wrote:
> If I copy class A into version 1+ it works. But it is the problem from CD
> perspective - I want to introduce feature which required new state: 1st I
> need make version 1+ with class B, but no other changes, then version 2 with
> class B and logic changes, upgrade job and if job doesn’t do what expected
> “rollback” to version 1+.
>
> ------------------------------
> *From:* Piotr Nowojski <pn...@apache.org>
> *Sent:* Wednesday, March 3, 2021 11:47:45 AM
> *To:* Alexey Trenikhun <ye...@msn.com>
> *Cc:* Flink User Mail List <us...@flink.apache.org>
> *Subject:* Re: Job downgrade
>
> Hi,
>
> I'm not sure what's the reason behind this. Probably classes are somehow
> attached to the state and this would explain why you are experiencing this
> issue. I've asked someone else from the community to chip in, but in the
> meantime, can not you just prepare a new "version 1" of the job, with just
> some empty `class B` on the class path? Or if this doesn't work, just copy
> the whole `class B` from version 2?
>
> Best,
> Piotrek
>
> sob., 27 lut 2021 o 19:10 Alexey Trenikhun <ye...@msn.com> napisał(a):
>
> Hello,
> Let's have version 1 of my job uses keyed state with name "a" and type A,
> which some Avro generated class. Then I upgrade to version 2, which in
> addition uses keyed state "b" and type B (another concrete Avro generated
> class), I take savepoint with version 2 and decided to downgrade to version
> 1 and start with taken savepoint, can I do it? On one hand, version 1
> doesn't have state "b", but seems Flink still tries to create call
> restoreSerializer and it tries to read runtimeType (`class B`) which is
> not available in version 1
>
> Thanks,
> Alexey
>
>
Re: Job downgrade
Posted by Alexey Trenikhun <ye...@msn.com>.
If I copy class A into version 1+ it works. But it is the problem from CD perspective - I want to introduce feature which required new state: 1st I need make version 1+ with class B, but no other changes, then version 2 with class B and logic changes, upgrade job and if job doesn’t do what expected “rollback” to version 1+.
________________________________
From: Piotr Nowojski <pn...@apache.org>
Sent: Wednesday, March 3, 2021 11:47:45 AM
To: Alexey Trenikhun <ye...@msn.com>
Cc: Flink User Mail List <us...@flink.apache.org>
Subject: Re: Job downgrade
Hi,
I'm not sure what's the reason behind this. Probably classes are somehow attached to the state and this would explain why you are experiencing this issue. I've asked someone else from the community to chip in, but in the meantime, can not you just prepare a new "version 1" of the job, with just some empty `class B` on the class path? Or if this doesn't work, just copy the whole `class B` from version 2?
Best,
Piotrek
sob., 27 lut 2021 o 19:10 Alexey Trenikhun <ye...@msn.com>> napisał(a):
Hello,
Let's have version 1 of my job uses keyed state with name "a" and type A, which some Avro generated class. Then I upgrade to version 2, which in addition uses keyed state "b" and type B (another concrete Avro generated class), I take savepoint with version 2 and decided to downgrade to version 1 and start with taken savepoint, can I do it? On one hand, version 1 doesn't have state "b", but seems Flink still tries to create call restoreSerializer and it tries to read runtimeType (`class B`) which is not available in version 1
Thanks,
Alexey