You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Lukáš Drbal <lu...@gmail.com> on 2022/03/08 15:37:21 UTC
Move savepoint to another s3 bucket
Hello everyone,
I'm trying to move savepoint to another s3 account but restore always
failed with some weird 404 error.
We are using lyft k8s operator [1] and flink 1.13.6 (in stacktrace you can
see version 1.13.6-396a8d44-szn which is just internal build from flink
commit b2ca390d478aa855eb0f2028d0ed965803a98af1)
What I'm trying to do:
1. create savepoint for pipeline via ./bin/flink savepoint <JOB_ID>
2. copy data under path configured in state.savepoints.dir from source
s3 to new s3
3. change all configuration and restore pipeline
Is this steps correct or I'm doing something wrong or unsupported?
All options related to s3 have valid values for new s3 account but restore
failed with exception bellow. Error message includes original path
(s3://flink/savepoints/activity-searched-query) which doesn't exists on new
account so that 404 is expected. But I still don't understand why flink
tries that path because related config options contains new bucket info.
high-availability.storageDir:
's3://<NEW_BUCKET>/ha/pipelines-runner-activity-searched-query'
jobmanager.archive.fs.dir: 's3://<NEW_BUCKET>/history'
state.checkpoints.dir:
> 's3://<NEW_BUCKET>/checkpoints/activity-searched-query'
state.savepoints.dir:
> 's3://<NEW_BUCKET>/savepoints/activity-searched-query'
+ valid values for s3.access-key and s3.secret-key
I found original path in _metadata file in savepoint data but changing that
(search&replace) leads to some weird OOM, I hope this should not be needed
and that values should be ignored.
state.backend is hashmap if it is important.
Restore back from source butcket works as expected.
Thanks a lot!
Regards,
L.
Stacktrace:
2022-03-08 15:39:25,838 [flink-akka.actor.default-dispatcher-4] INFO
> org.apache.flink.runtime.executiongraph.ExecutionGraph -
> CombineToSearchedQuery -> (LateElementsCounter, TransformToStreamElement ->
> Sink: SearchedQueryKafkaSink) (1/2) (0c0f108c393b9a5b58f861c1032671d0)
> switched from INITIALIZING to FAILED on 10.67.158.155:45521-d8d19d @
> 10.67.158.155 (dataPort=36341).
> org.apache.flink.util.SerializedThrowable: Exception while creating
> StreamOperatorStateContext.
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at java.lang.Thread.run(Thread.java:832) ~[?:?]
> Caused by: org.apache.flink.util.SerializedThrowable: Could not restore
> keyed state backend for
> WindowOperator_bd2a73c53230733509ca171c6476fcc5_(1/2) from any of the 1
> provided restore options.
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> ... 10 more
> Caused by: org.apache.flink.util.SerializedThrowable: Failed when trying
> to restore heap backend
> at
> org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.restoreState(HeapKeyedStateBackendBuilder.java:177)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:111)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:131)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:73)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.runtime.state.StateBackend.createKeyedStateBackend(StateBackend.java:136)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> ... 10 more
> Caused by: org.apache.flink.util.SerializedThrowable:
> com.amazonaws.services.s3.model.AmazonS3Exception: null (Service: Amazon
> S3; Status Code: 404; Error Code: NoSuchBucket; Request ID:
> tx0000000000000011aacd6-0062276a9d-85d05-default; S3 Extended Request ID:
> 85d05-default-default; Proxy: null), S3 Extended Request ID:
> 85d05-default-default (Path:
> s3://flink/savepoints/activity-searched-query/savepoint-ff3caa-f4b6db96b68b/000e1fc6-0ed8-452a-a8f2-57650fa0594d)
> at
> com.facebook.presto.hive.s3.PrestoS3FileSystem$PrestoS3InputStream.lambda$openStream$1(PrestoS3FileSystem.java:917)
> ~[?:?]
> at com.facebook.presto.hive.RetryDriver.run(RetryDriver.java:138) ~[?:?]
>
[1] https://github.com/lyft/flinkk8soperator
Re: Move savepoint to another s3 bucket
Posted by Lukáš Drbal <lu...@gmail.com>.
Hi Dawid,
I just tried the same steps on flink builded from git branch release-1.13
and everything works as expected!
Thank you all!
L.
On Wed, Mar 9, 2022 at 8:49 AM Dawid Wysakowicz <dw...@apache.org>
wrote:
> Hi Lukas,
>
> I am afraid you're hitting this bug:
> https://issues.apache.org/jira/browse/FLINK-25952
>
> Best,
>
> Dawid
> On 08/03/2022 16:37, Lukáš Drbal wrote:
>
> Hello everyone,
>
> I'm trying to move savepoint to another s3 account but restore always
> failed with some weird 404 error.
>
> We are using lyft k8s operator [1] and flink 1.13.6 (in stacktrace you can
> see version 1.13.6-396a8d44-szn which is just internal build from flink
> commit b2ca390d478aa855eb0f2028d0ed965803a98af1)
>
> What I'm trying to do:
>
> 1. create savepoint for pipeline via ./bin/flink savepoint <JOB_ID>
> 2. copy data under path configured in state.savepoints.dir from source
> s3 to new s3
> 3. change all configuration and restore pipeline
>
> Is this steps correct or I'm doing something wrong or unsupported?
>
> All options related to s3 have valid values for new s3 account but restore
> failed with exception bellow. Error message includes original path
> (s3://flink/savepoints/activity-searched-query) which doesn't exists on new
> account so that 404 is expected. But I still don't understand why flink
> tries that path because related config options contains new bucket info.
> high-availability.storageDir:
> 's3://<NEW_BUCKET>/ha/pipelines-runner-activity-searched-query'
>
> jobmanager.archive.fs.dir: 's3://<NEW_BUCKET>/history'
>
> state.checkpoints.dir:
>> 's3://<NEW_BUCKET>/checkpoints/activity-searched-query'
>
> state.savepoints.dir:
>> 's3://<NEW_BUCKET>/savepoints/activity-searched-query'
>
>
> + valid values for s3.access-key and s3.secret-key
>
> I found original path in _metadata file in savepoint data but changing
> that (search&replace) leads to some weird OOM, I hope this should not be
> needed and that values should be ignored.
>
> state.backend is hashmap if it is important.
>
> Restore back from source butcket works as expected.
>
> Thanks a lot!
>
> Regards,
> L.
>
> Stacktrace:
>
> 2022-03-08 15:39:25,838 [flink-akka.actor.default-dispatcher-4] INFO
>> org.apache.flink.runtime.executiongraph.ExecutionGraph -
>> CombineToSearchedQuery -> (LateElementsCounter, TransformToStreamElement ->
>> Sink: SearchedQueryKafkaSink) (1/2) (0c0f108c393b9a5b58f861c1032671d0)
>> switched from INITIALIZING to FAILED on 10.67.158.155:45521-d8d19d @
>> 10.67.158.155 (dataPort=36341).
>> org.apache.flink.util.SerializedThrowable: Exception while creating
>> StreamOperatorStateContext.
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at java.lang.Thread.run(Thread.java:832) ~[?:?]
>> Caused by: org.apache.flink.util.SerializedThrowable: Could not restore
>> keyed state backend for
>> WindowOperator_bd2a73c53230733509ca171c6476fcc5_(1/2) from any of the 1
>> provided restore options.
>> at
>> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> ... 10 more
>> Caused by: org.apache.flink.util.SerializedThrowable: Failed when trying
>> to restore heap backend
>> at
>> org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.restoreState(HeapKeyedStateBackendBuilder.java:177)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:111)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:131)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:73)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.runtime.state.StateBackend.createKeyedStateBackend(StateBackend.java:136)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> ... 10 more
>> Caused by: org.apache.flink.util.SerializedThrowable:
>> com.amazonaws.services.s3.model.AmazonS3Exception: null (Service: Amazon
>> S3; Status Code: 404; Error Code: NoSuchBucket; Request ID:
>> tx0000000000000011aacd6-0062276a9d-85d05-default; S3 Extended Request ID:
>> 85d05-default-default; Proxy: null), S3 Extended Request ID:
>> 85d05-default-default (Path:
>> s3://flink/savepoints/activity-searched-query/savepoint-ff3caa-f4b6db96b68b/000e1fc6-0ed8-452a-a8f2-57650fa0594d)
>> at
>> com.facebook.presto.hive.s3.PrestoS3FileSystem$PrestoS3InputStream.lambda$openStream$1(PrestoS3FileSystem.java:917)
>> ~[?:?]
>> at com.facebook.presto.hive.RetryDriver.run(RetryDriver.java:138) ~[?:?]
>>
>
> [1] https://github.com/lyft/flinkk8soperator
>
>
Re: Move savepoint to another s3 bucket
Posted by Dawid Wysakowicz <dw...@apache.org>.
Hi Lukas,
I am afraid you're hitting this bug:
https://issues.apache.org/jira/browse/FLINK-25952
Best,
Dawid
On 08/03/2022 16:37, Lukáš Drbal wrote:
> Hello everyone,
>
> I'm trying to move savepoint to another s3 account but restore always
> failed with some weird 404 error.
>
> We are using lyft k8s operator [1] and flink 1.13.6 (in stacktrace you
> can see version 1.13.6-396a8d44-szn which is just internal build from
> flink commit b2ca390d478aa855eb0f2028d0ed965803a98af1)
>
> What I'm trying to do:
>
> 1. create savepoint for pipeline via ./bin/flink savepoint <JOB_ID>
> 2. copy data under path configured in state.savepoints.dir from
> source s3 to new s3
> 3. change all configuration and restore pipeline
>
> Is this steps correct or I'm doing something wrong or unsupported?
>
> All options related to s3 have valid values for new s3 account but
> restore failed with exception bellow. Error message includes original
> path (s3://flink/savepoints/activity-searched-query) which doesn't
> exists on new account so that 404 is expected. But I still don't
> understand why flink tries that path because related config options
> contains new bucket info.
> high-availability.storageDir:
> 's3://<NEW_BUCKET>/ha/pipelines-runner-activity-searched-query'
>
> jobmanager.archive.fs.dir: 's3://<NEW_BUCKET>/history'
>
> state.checkpoints.dir:
> 's3://<NEW_BUCKET>/checkpoints/activity-searched-query'
>
> state.savepoints.dir:
> 's3://<NEW_BUCKET>/savepoints/activity-searched-query'
>
> + valid values for s3.access-key and s3.secret-key
>
> I found original path in _metadata file in savepoint data but changing
> that (search&replace) leads to some weird OOM, I hope this should not
> be needed and that values should be ignored.
>
> state.backend is hashmap if it is important.
>
> Restore back from source butcket works as expected.
>
> Thanks a lot!
>
> Regards,
> L.
>
> Stacktrace:
>
> 2022-03-08 15:39:25,838 [flink-akka.actor.default-dispatcher-4]
> INFO org.apache.flink.runtime.executiongraph.ExecutionGraph -
> CombineToSearchedQuery -> (LateElementsCounter,
> TransformToStreamElement -> Sink: SearchedQueryKafkaSink) (1/2)
> (0c0f108c393b9a5b58f861c1032671d0) switched from INITIALIZING to
> FAILED on 10.67.158.155:45521-d8d19d @ 10.67.158.155 (dataPort=36341).
> org.apache.flink.util.SerializedThrowable: Exception while
> creating StreamOperatorStateContext.
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at java.lang.Thread.run(Thread.java:832) ~[?:?]
> Caused by: org.apache.flink.util.SerializedThrowable: Could not
> restore keyed state backend for
> WindowOperator_bd2a73c53230733509ca171c6476fcc5_(1/2) from any of
> the 1 provided restore options.
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> ... 10 more
> Caused by: org.apache.flink.util.SerializedThrowable: Failed when
> trying to restore heap backend
> at
> org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.restoreState(HeapKeyedStateBackendBuilder.java:177)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:111)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:131)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:73)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.runtime.state.StateBackend.createKeyedStateBackend(StateBackend.java:136)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> at
> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)
> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
> ... 10 more
> Caused by: org.apache.flink.util.SerializedThrowable:
> com.amazonaws.services.s3.model.AmazonS3Exception: null (Service:
> Amazon S3; Status Code: 404; Error Code: NoSuchBucket; Request ID:
> tx0000000000000011aacd6-0062276a9d-85d05-default; S3 Extended
> Request ID: 85d05-default-default; Proxy: null), S3 Extended
> Request ID: 85d05-default-default (Path:
> s3://flink/savepoints/activity-searched-query/savepoint-ff3caa-f4b6db96b68b/000e1fc6-0ed8-452a-a8f2-57650fa0594d)
> at
> com.facebook.presto.hive.s3.PrestoS3FileSystem$PrestoS3InputStream.lambda$openStream$1(PrestoS3FileSystem.java:917)
> ~[?:?]
> at com.facebook.presto.hive.RetryDriver.run(RetryDriver.java:138)
> ~[?:?]
>
>
> [1] https://github.com/lyft/flinkk8soperator
RE: Move savepoint to another s3 bucket
Posted by Schwalbe Matthias <Ma...@viseca.ch>.
Hi Lukáš,
Did you already find the path that was mentioned in your exception stack?
Anyway, as said, there are different types of StateHandle.
When a savepoint/checkpoint is made, the savepoint strategy decides which type to use:
* ByteStreamStateHandle for smaller states that fit into _metadata,
* RelativeFileStateHandle for bigger state (file is relocatable)
* Older implementations used FileStateHandle instead which is not relocatable
* Etc.
In your case, the old path in ByteStreamStateHandle is ok, the data sits right in _metadata, the path is only used to give it a (any) unique id.
Try to find the path from the exception, it must be somewhere in that object hierarchy … the we take a closer look…
Cheers Thias
From: Lukáš Drbal <lu...@gmail.com>
Sent: Dienstag, 8. März 2022 19:42
To: Schwalbe Matthias <Ma...@viseca.ch>; user <us...@flink.apache.org>
Subject: Re: Move savepoint to another s3 bucket
forgot to include ML, sorry for spam.
On Tue, Mar 8, 2022 at 7:27 PM Lukáš Drbal <lu...@gmail.com>> wrote:
Hi Matthias,
Thank you for your reply and confirmation. I just try to load that savepoint via
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
ExistingSavepoint savepoint =
Savepoint.load(env, "/tmp/savepoint-ff3caa-f4b6db96b68b/", new HashMapStateBackend());
Thank you for this hint!
But in stateHandle I don't see RelativeFileStateHandle like you have but ByteStreamStateHandle which contains the original path on s3. Is this correct?
[cid:image001.png@01D8338D.FEE05670]
I will try to dig more deeper here.
L.
On Tue, Mar 8, 2022 at 5:35 PM Schwalbe Matthias <Ma...@viseca.ch>> wrote:
Hi Lukáš,
Just as initial check (i.e. no offense intended 😊 ) , you set the ‘execution.savepoint.path’ config (or resp. command line option) to the new s3 path (that contains the _metadata file) when you started the job?
• In flink 1.13, savepoints in general are relocatable that means it should work out of the box.
• Manually patching _metadata files is bound to fail, you confuse the deserializer in the best case
• However you can load the relocated savepoint by means of the state processor API [1] in a debugger and look into the metadata
o var savepointFlink = Savepoint.load(env, savepointPath, stateBackend) will do
o this code only opens the _metadata file and can also be done with a local copy of the metadata file (savepointPath should not include “_metadata”)
o in there you’ll find a collection of operator states (metadata):
o [cid:image002.png@01D8338D.FEE05670]
o Inner structure of such operator state is:
o [cid:image003.png@01D8338D.FEE05670]
o Marked in green are the 6 different state types for keyed state / operator state / unaligned checkpoint I/O buffers
o In these State you’ll find references for different types of state handles
o [cid:image004.png@01D8338D.FEE05670]
o Some of them relate to files like RelativeFileStateHandle, some to state integrated into _metadata file etc.
o After you copy a savepoint, RelativeFileStateHandle of course still contains the original path, but when loading the state, this path is patched to represent the new path + the relativePath
• You might want to take a look into the metadata in order to find the old path references in your exception and find out what is missing
Curious to hear what you find out 😊
Thias
[1] https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/libs/state_processor_api/
From: Lukáš Drbal <lu...@gmail.com>>
Sent: Dienstag, 8. März 2022 16:37
To: user <us...@flink.apache.org>>
Subject: Move savepoint to another s3 bucket
Hello everyone,
I'm trying to move savepoint to another s3 account but restore always failed with some weird 404 error.
We are using lyft k8s operator [1] and flink 1.13.6 (in stacktrace you can see version 1.13.6-396a8d44-szn which is just internal build from flink commit b2ca390d478aa855eb0f2028d0ed965803a98af1)
What I'm trying to do:
1. create savepoint for pipeline via ./bin/flink savepoint <JOB_ID>
2. copy data under path configured in state.savepoints.dir from source s3 to new s3
3. change all configuration and restore pipeline
Is this steps correct or I'm doing something wrong or unsupported?
All options related to s3 have valid values for new s3 account but restore failed with exception bellow. Error message includes original path (s3://flink/savepoints/activity-searched-query) which doesn't exists on new account so that 404 is expected. But I still don't understand why flink tries that path because related config options contains new bucket info.
high-availability.storageDir: 's3://<NEW_BUCKET>/ha/pipelines-runner-activity-searched-query'
jobmanager.archive.fs.dir: 's3://<NEW_BUCKET>/history'
state.checkpoints.dir: 's3://<NEW_BUCKET>/checkpoints/activity-searched-query'
state.savepoints.dir: 's3://<NEW_BUCKET>/savepoints/activity-searched-query'
+ valid values for s3.access-key and s3.secret-key
I found original path in _metadata file in savepoint data but changing that (search&replace) leads to some weird OOM, I hope this should not be needed and that values should be ignored.
state.backend is hashmap if it is important.
Restore back from source butcket works as expected.
Thanks a lot!
Regards,
L.
Stacktrace:
2022-03-08 15:39:25,838 [flink-akka.actor.default-dispatcher-4] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - CombineToSearchedQuery -> (LateElementsCounter, TransformToStreamElement -> Sink: SearchedQueryKafkaSink) (1/2) (0c0f108c393b9a5b58f861c1032671d0) switched from INITIALIZING to FAILED on 10.67.158.155:45521-d8d19d @ 10.67.158.155 (dataPort=36341).
org.apache.flink.util.SerializedThrowable: Exception while creating StreamOperatorStateContext.
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at java.lang.Thread.run(Thread.java:832) ~[?:?]
Caused by: org.apache.flink.util.SerializedThrowable: Could not restore keyed state backend for WindowOperator_bd2a73c53230733509ca171c6476fcc5_(1/2) from any of the 1 provided restore options.
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
... 10 more
Caused by: org.apache.flink.util.SerializedThrowable: Failed when trying to restore heap backend
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.restoreState(HeapKeyedStateBackendBuilder.java:177) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:111) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:131) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:73) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.state.StateBackend.createKeyedStateBackend(StateBackend.java:136) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
... 10 more
Caused by: org.apache.flink.util.SerializedThrowable: com.amazonaws.services.s3.model.AmazonS3Exception: null (Service: Amazon S3; Status Code: 404; Error Code: NoSuchBucket; Request ID: tx0000000000000011aacd6-0062276a9d-85d05-default; S3 Extended Request ID: 85d05-default-default; Proxy: null), S3 Extended Request ID: 85d05-default-default (Path: s3://flink/savepoints/activity-searched-query/savepoint-ff3caa-f4b6db96b68b/000e1fc6-0ed8-452a-a8f2-57650fa0594d)
at com.facebook.presto.hive.s3.PrestoS3FileSystem$PrestoS3InputStream.lambda$openStream$1(PrestoS3FileSystem.java:917) ~[?:?]
at com.facebook.presto.hive.RetryDriver.run(RetryDriver.java:138) ~[?:?]
[1] https://github.com/lyft/flinkk8soperator
Diese Nachricht ist ausschliesslich für den Adressaten bestimmt und beinhaltet unter Umständen vertrauliche Mitteilungen. Da die Vertraulichkeit von e-Mail-Nachrichten nicht gewährleistet werden kann, übernehmen wir keine Haftung für die Gewährung der Vertraulichkeit und Unversehrtheit dieser Mitteilung. Bei irrtümlicher Zustellung bitten wir Sie um Benachrichtigung per e-Mail und um Löschung dieser Nachricht sowie eventueller Anhänge. Jegliche unberechtigte Verwendung oder Verbreitung dieser Informationen ist streng verboten.
This message is intended only for the named recipient and may contain confidential or privileged information. As the confidentiality of email communication cannot be guaranteed, we do not accept any responsibility for the confidentiality and the intactness of this message. If you have received it in error, please advise the sender by return e-mail and delete this message and any attachments. Any unauthorised use or dissemination of this information is strictly prohibited.
Diese Nachricht ist ausschliesslich für den Adressaten bestimmt und beinhaltet unter Umständen vertrauliche Mitteilungen. Da die Vertraulichkeit von e-Mail-Nachrichten nicht gewährleistet werden kann, übernehmen wir keine Haftung für die Gewährung der Vertraulichkeit und Unversehrtheit dieser Mitteilung. Bei irrtümlicher Zustellung bitten wir Sie um Benachrichtigung per e-Mail und um Löschung dieser Nachricht sowie eventueller Anhänge. Jegliche unberechtigte Verwendung oder Verbreitung dieser Informationen ist streng verboten.
This message is intended only for the named recipient and may contain confidential or privileged information. As the confidentiality of email communication cannot be guaranteed, we do not accept any responsibility for the confidentiality and the intactness of this message. If you have received it in error, please advise the sender by return e-mail and delete this message and any attachments. Any unauthorised use or dissemination of this information is strictly prohibited.
Re: Move savepoint to another s3 bucket
Posted by Lukáš Drbal <lu...@gmail.com>.
forgot to include ML, sorry for spam.
On Tue, Mar 8, 2022 at 7:27 PM Lukáš Drbal <lu...@gmail.com> wrote:
> Hi Matthias,
>
> Thank you for your reply and confirmation. I just try to load that
> savepoint via
>
> ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
> ExistingSavepoint savepoint =
> Savepoint.load(env, "/tmp/savepoint-ff3caa-f4b6db96b68b/", new HashMapStateBackend());
>
> Thank you for this hint!
> But in stateHandle I don't see RelativeFileStateHandle like you have
> but ByteStreamStateHandle which contains the original path on s3. Is this
> correct?
>
> [image: state.png]
>
> I will try to dig more deeper here.
>
> L.
>
> On Tue, Mar 8, 2022 at 5:35 PM Schwalbe Matthias <
> Matthias.Schwalbe@viseca.ch> wrote:
>
>> Hi Lukáš,
>>
>>
>>
>> Just as initial check (i.e. no offense intended 😊 ) , you set the ‘execution.savepoint.path’ config (or resp. command line option) to the new s3 path (that contains the _metadata file) when you started the job?
>>
>> · In flink 1.13, savepoints in general are relocatable that means it should work out of the box.
>>
>> · Manually patching _metadata files is bound to fail, you confuse the deserializer in the best case
>>
>> · However you can load the relocated savepoint by means of the state processor API [1] in a debugger and look into the metadata
>>
>> o var savepointFlink = Savepoint.load(env, savepointPath, stateBackend) will do
>>
>> o this code only opens the _metadata file and can also be done with a local copy of the metadata file (savepointPath should not include “_metadata”)
>>
>> o in there you’ll find a collection of operator states (metadata):
>>
>> o
>>
>> o Inner structure of such operator state is:
>>
>> o
>>
>> o Marked in green are the 6 different state types for keyed state / operator state / unaligned checkpoint I/O buffers
>>
>> o In these State you’ll find references for different types of state handles
>>
>> o
>>
>> o Some of them relate to files like RelativeFileStateHandle, some to state integrated into _metadata file etc.
>>
>> o After you copy a savepoint, RelativeFileStateHandle of course still contains the original path, but when loading the state, this path is patched to represent the new path + the relativePath
>>
>> · You might want to take a look into the metadata in order to find the old path references in your exception and find out what is missing
>>
>>
>>
>>
>>
>> Curious to hear what you find out 😊
>>
>>
>>
>>
>>
>> Thias
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> [1] https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/libs/state_processor_api/
>>
>>
>>
>>
>>
>>
>>
>> *From:* Lukáš Drbal <lu...@gmail.com>
>> *Sent:* Dienstag, 8. März 2022 16:37
>> *To:* user <us...@flink.apache.org>
>> *Subject:* Move savepoint to another s3 bucket
>>
>>
>>
>> Hello everyone,
>>
>>
>>
>> I'm trying to move savepoint to another s3 account but restore always
>> failed with some weird 404 error.
>>
>>
>>
>> We are using lyft k8s operator [1] and flink 1.13.6 (in stacktrace you
>> can see version 1.13.6-396a8d44-szn which is just internal build from flink
>> commit b2ca390d478aa855eb0f2028d0ed965803a98af1)
>>
>>
>>
>> What I'm trying to do:
>>
>> 1. create savepoint for pipeline via ./bin/flink savepoint <JOB_ID>
>> 2. copy data under path configured in state.savepoints.dir from
>> source s3 to new s3
>> 3. change all configuration and restore pipeline
>>
>> Is this steps correct or I'm doing something wrong or unsupported?
>>
>>
>>
>> All options related to s3 have valid values for new s3 account but
>> restore failed with exception bellow. Error message includes original path
>> (s3://flink/savepoints/activity-searched-query) which doesn't exists on new
>> account so that 404 is expected. But I still don't understand why flink
>> tries that path because related config options contains new bucket info.
>>
>> high-availability.storageDir:
>> 's3://<NEW_BUCKET>/ha/pipelines-runner-activity-searched-query'
>>
>> jobmanager.archive.fs.dir: 's3://<NEW_BUCKET>/history'
>>
>> state.checkpoints.dir:
>> 's3://<NEW_BUCKET>/checkpoints/activity-searched-query'
>>
>> state.savepoints.dir:
>> 's3://<NEW_BUCKET>/savepoints/activity-searched-query'
>>
>>
>>
>> + valid values for s3.access-key and s3.secret-key
>>
>>
>>
>> I found original path in _metadata file in savepoint data but changing
>> that (search&replace) leads to some weird OOM, I hope this should not be
>> needed and that values should be ignored.
>>
>>
>>
>> state.backend is hashmap if it is important.
>>
>>
>>
>> Restore back from source butcket works as expected.
>>
>>
>>
>> Thanks a lot!
>>
>>
>>
>> Regards,
>>
>> L.
>>
>>
>>
>> Stacktrace:
>>
>>
>>
>> 2022-03-08 15:39:25,838 [flink-akka.actor.default-dispatcher-4] INFO
>> org.apache.flink.runtime.executiongraph.ExecutionGraph -
>> CombineToSearchedQuery -> (LateElementsCounter, TransformToStreamElement ->
>> Sink: SearchedQueryKafkaSink) (1/2) (0c0f108c393b9a5b58f861c1032671d0)
>> switched from INITIALIZING to FAILED on 10.67.158.155:45521-d8d19d @
>> 10.67.158.155 (dataPort=36341).
>> org.apache.flink.util.SerializedThrowable: Exception while creating
>> StreamOperatorStateContext.
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at java.lang.Thread.run(Thread.java:832) ~[?:?]
>> Caused by: org.apache.flink.util.SerializedThrowable: Could not restore
>> keyed state backend for
>> WindowOperator_bd2a73c53230733509ca171c6476fcc5_(1/2) from any of the 1
>> provided restore options.
>> at
>> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> ... 10 more
>> Caused by: org.apache.flink.util.SerializedThrowable: Failed when trying
>> to restore heap backend
>> at
>> org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.restoreState(HeapKeyedStateBackendBuilder.java:177)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:111)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:131)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:73)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.runtime.state.StateBackend.createKeyedStateBackend(StateBackend.java:136)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> at
>> org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)
>> ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
>> ... 10 more
>> Caused by: org.apache.flink.util.SerializedThrowable:
>> com.amazonaws.services.s3.model.AmazonS3Exception: null (Service: Amazon
>> S3; Status Code: 404; Error Code: NoSuchBucket; Request ID:
>> tx0000000000000011aacd6-0062276a9d-85d05-default; S3 Extended Request ID:
>> 85d05-default-default; Proxy: null), S3 Extended Request ID:
>> 85d05-default-default (Path:
>> s3://flink/savepoints/activity-searched-query/savepoint-ff3caa-f4b6db96b68b/000e1fc6-0ed8-452a-a8f2-57650fa0594d)
>> at
>> com.facebook.presto.hive.s3.PrestoS3FileSystem$PrestoS3InputStream.lambda$openStream$1(PrestoS3FileSystem.java:917)
>> ~[?:?]
>> at com.facebook.presto.hive.RetryDriver.run(RetryDriver.java:138) ~[?:?]
>>
>>
>> [1] https://github.com/lyft/flinkk8soperator
>> Diese Nachricht ist ausschliesslich für den Adressaten bestimmt und
>> beinhaltet unter Umständen vertrauliche Mitteilungen. Da die
>> Vertraulichkeit von e-Mail-Nachrichten nicht gewährleistet werden kann,
>> übernehmen wir keine Haftung für die Gewährung der Vertraulichkeit und
>> Unversehrtheit dieser Mitteilung. Bei irrtümlicher Zustellung bitten wir
>> Sie um Benachrichtigung per e-Mail und um Löschung dieser Nachricht sowie
>> eventueller Anhänge. Jegliche unberechtigte Verwendung oder Verbreitung
>> dieser Informationen ist streng verboten.
>>
>> This message is intended only for the named recipient and may contain
>> confidential or privileged information. As the confidentiality of email
>> communication cannot be guaranteed, we do not accept any responsibility for
>> the confidentiality and the intactness of this message. If you have
>> received it in error, please advise the sender by return e-mail and delete
>> this message and any attachments. Any unauthorised use or dissemination of
>> this information is strictly prohibited.
>>
>
RE: Move savepoint to another s3 bucket
Posted by Schwalbe Matthias <Ma...@viseca.ch>.
Hi Lukáš,
Just as initial check (i.e. no offense intended 😊 ) , you set the ‘execution.savepoint.path’ config (or resp. command line option) to the new s3 path (that contains the _metadata file) when you started the job?
· In flink 1.13, savepoints in general are relocatable that means it should work out of the box.
· Manually patching _metadata files is bound to fail, you confuse the deserializer in the best case
· However you can load the relocated savepoint by means of the state processor API [1] in a debugger and look into the metadata
o var savepointFlink = Savepoint.load(env, savepointPath, stateBackend) will do
o this code only opens the _metadata file and can also be done with a local copy of the metadata file (savepointPath should not include “_metadata”)
o in there you’ll find a collection of operator states (metadata):
o [cid:image001.png@01D83311.10171B40]
o Inner structure of such operator state is:
o [cid:image002.png@01D83311.ABF991F0]
o Marked in green are the 6 different state types for keyed state / operator state / unaligned checkpoint I/O buffers
o In these State you’ll find references for different types of state handles
o [cid:image003.png@01D83312.DB971850]
o Some of them relate to files like RelativeFileStateHandle, some to state integrated into _metadata file etc.
o After you copy a savepoint, RelativeFileStateHandle of course still contains the original path, but when loading the state, this path is patched to represent the new path + the relativePath
· You might want to take a look into the metadata in order to find the old path references in your exception and find out what is missing
Curious to hear what you find out 😊
Thias
[1] https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/libs/state_processor_api/
From: Lukáš Drbal <lu...@gmail.com>
Sent: Dienstag, 8. März 2022 16:37
To: user <us...@flink.apache.org>
Subject: Move savepoint to another s3 bucket
Hello everyone,
I'm trying to move savepoint to another s3 account but restore always failed with some weird 404 error.
We are using lyft k8s operator [1] and flink 1.13.6 (in stacktrace you can see version 1.13.6-396a8d44-szn which is just internal build from flink commit b2ca390d478aa855eb0f2028d0ed965803a98af1)
What I'm trying to do:
1. create savepoint for pipeline via ./bin/flink savepoint <JOB_ID>
2. copy data under path configured in state.savepoints.dir from source s3 to new s3
3. change all configuration and restore pipeline
Is this steps correct or I'm doing something wrong or unsupported?
All options related to s3 have valid values for new s3 account but restore failed with exception bellow. Error message includes original path (s3://flink/savepoints/activity-searched-query) which doesn't exists on new account so that 404 is expected. But I still don't understand why flink tries that path because related config options contains new bucket info.
high-availability.storageDir: 's3://<NEW_BUCKET>/ha/pipelines-runner-activity-searched-query'
jobmanager.archive.fs.dir: 's3://<NEW_BUCKET>/history'
state.checkpoints.dir: 's3://<NEW_BUCKET>/checkpoints/activity-searched-query'
state.savepoints.dir: 's3://<NEW_BUCKET>/savepoints/activity-searched-query'
+ valid values for s3.access-key and s3.secret-key
I found original path in _metadata file in savepoint data but changing that (search&replace) leads to some weird OOM, I hope this should not be needed and that values should be ignored.
state.backend is hashmap if it is important.
Restore back from source butcket works as expected.
Thanks a lot!
Regards,
L.
Stacktrace:
2022-03-08 15:39:25,838 [flink-akka.actor.default-dispatcher-4] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - CombineToSearchedQuery -> (LateElementsCounter, TransformToStreamElement -> Sink: SearchedQueryKafkaSink) (1/2) (0c0f108c393b9a5b58f861c1032671d0) switched from INITIALIZING to FAILED on 10.67.158.155:45521-d8d19d @ 10.67.158.155 (dataPort=36341).
org.apache.flink.util.SerializedThrowable: Exception while creating StreamOperatorStateContext.
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at java.lang.Thread.run(Thread.java:832) ~[?:?]
Caused by: org.apache.flink.util.SerializedThrowable: Could not restore keyed state backend for WindowOperator_bd2a73c53230733509ca171c6476fcc5_(1/2) from any of the 1 provided restore options.
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
... 10 more
Caused by: org.apache.flink.util.SerializedThrowable: Failed when trying to restore heap backend
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.restoreState(HeapKeyedStateBackendBuilder.java:177) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:111) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:131) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.state.hashmap.HashMapStateBackend.createKeyedStateBackend(HashMapStateBackend.java:73) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.runtime.state.StateBackend.createKeyedStateBackend(StateBackend.java:136) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163) ~[flink-dist_2.11-1.13.6-396a8d44-szn.jar:1.13.6-396a8d44-szn]
... 10 more
Caused by: org.apache.flink.util.SerializedThrowable: com.amazonaws.services.s3.model.AmazonS3Exception: null (Service: Amazon S3; Status Code: 404; Error Code: NoSuchBucket; Request ID: tx0000000000000011aacd6-0062276a9d-85d05-default; S3 Extended Request ID: 85d05-default-default; Proxy: null), S3 Extended Request ID: 85d05-default-default (Path: s3://flink/savepoints/activity-searched-query/savepoint-ff3caa-f4b6db96b68b/000e1fc6-0ed8-452a-a8f2-57650fa0594d)
at com.facebook.presto.hive.s3.PrestoS3FileSystem$PrestoS3InputStream.lambda$openStream$1(PrestoS3FileSystem.java:917) ~[?:?]
at com.facebook.presto.hive.RetryDriver.run(RetryDriver.java:138) ~[?:?]
[1] https://github.com/lyft/flinkk8soperator
Diese Nachricht ist ausschliesslich für den Adressaten bestimmt und beinhaltet unter Umständen vertrauliche Mitteilungen. Da die Vertraulichkeit von e-Mail-Nachrichten nicht gewährleistet werden kann, übernehmen wir keine Haftung für die Gewährung der Vertraulichkeit und Unversehrtheit dieser Mitteilung. Bei irrtümlicher Zustellung bitten wir Sie um Benachrichtigung per e-Mail und um Löschung dieser Nachricht sowie eventueller Anhänge. Jegliche unberechtigte Verwendung oder Verbreitung dieser Informationen ist streng verboten.
This message is intended only for the named recipient and may contain confidential or privileged information. As the confidentiality of email communication cannot be guaranteed, we do not accept any responsibility for the confidentiality and the intactness of this message. If you have received it in error, please advise the sender by return e-mail and delete this message and any attachments. Any unauthorised use or dissemination of this information is strictly prohibited.