I responded on the issue.
> On 12. Sep 2017, at 22:25, Lukasz Cwik <lc...@google.com.INVALID> wrote:
>
> Filed https://issues.apache.org/jira/browse/BEAM-2948
>
> On Tue, Sep 12, 2017 at 2:10 AM, Pawel Bartoszek <pawelbartosze...@gmail.com
>> 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
>>
