Filed https://issues.apache.org/jira/browse/BEAM-2948
On Tue, Sep 12, 2017 at 2:10 AM, Pawel Bartoszek <[email protected] > wrote: > Hi, > > I am running a flink v1.2.1 job on a EMR cluster using Beam v2.0.0. > When I try to restore job from savepoint on one task manager I get the > exception *Unable to find registrar for s3n*. > The job can write files to s3 acting as a sink. So S3 access works except > when restoring from savepoint. > > I am passing the following configuration as the pipeline > HadoopFileSystemOptions > options: > > Configuration configuration = new Configuration(); > configuration.set("fs.defaultFS", String.format("s3n://%s", > jobOptions.getOutputFileSystemRoot())); > configuration.set("fs.default.name", String.format("s3n://%s", > jobOptions.getOutputFileSystemRoot())); > configuration.set("fs.s3.impl", > "org.apache.hadoop.fs.s3native.NativeS3FileSystem"); > configuration.set("fs.s3n.awsAccessKeyId", > jobOptions.getAwsAccessKey()); > configuration.set("fs.s3n.awsSecretAccessKey", > jobOptions.getAwsSecretKey()); > > > > From my investigation it looks like Beam FileSystems. > setDefaultPipelineOptions method is not called before > org.apache.beam.sdk.io.FileBasedSink$FileResultCoder relay on initialised > FileSystems.SCHEME_TO_FILESYSTEM map > > > > java.lang.IllegalStateException: Could not initialize keyed state backend. > at > org.apache.flink.streaming.api.operators.AbstractStreamOperator. > initKeyedState(AbstractStreamOperator.java:293) > at > org.apache.flink.streaming.api.operators.AbstractStreamOperator. > initializeState(AbstractStreamOperator.java:204) > at > org.apache.flink.streaming.runtime.tasks.StreamTask.initializeOperators( > StreamTask.java:653) > at > org.apache.flink.streaming.runtime.tasks.StreamTask. > initializeState(StreamTask.java:640) > at > org.apache.flink.streaming.runtime.tasks.StreamTask. > invoke(StreamTask.java:246) > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:665) > at java.lang.Thread.run(Thread.java:748) > Caused by: java.lang.IllegalStateException: Unable to find registrar for > s3n > at > org.apache.beam.sdk.io.FileSystems.getFileSystemInternal( > FileSystems.java:447) > at org.apache.beam.sdk.io.FileSystems.matchNewResource( > FileSystems.java:523) > at > org.apache.beam.sdk.io.FileBasedSink$FileResultCoder. > decode(FileBasedSink.java:1059) > at > org.apache.beam.sdk.io.FileBasedSink$FileResultCoder. > decode(FileBasedSink.java:1020) > at > org.apache.beam.runners.flink.translation.types.CoderTypeSerializer. > deserialize(CoderTypeSerializer.java:87) > at > org.apache.flink.runtime.state.ArrayListSerializer.deserialize( > ArrayListSerializer.java:87) > at > org.apache.flink.runtime.state.ArrayListSerializer.deserialize( > ArrayListSerializer.java:27) > at > org.apache.flink.runtime.state.heap.HeapKeyedStateBackend. > readStateTableForKeyGroup(HeapKeyedStateBackend.java:370) > at > org.apache.flink.runtime.state.heap.HeapKeyedStateBackend. > restorePartitionedState(HeapKeyedStateBackend.java:340) > at > org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore( > HeapKeyedStateBackend.java:243) > at > org.apache.flink.streaming.runtime.tasks.StreamTask. > createKeyedStateBackend(StreamTask.java:788) > at > org.apache.flink.streaming.api.operators.AbstractStreamOperator. > initKeyedState(AbstractStreamOperator.java:284) > ... 6 more > > > > Pawel Bartoszek >
