Unable to find registrar for s3n when restoring flink job from savepoint

[Pawel Bartoszek]

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