Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/4818#discussion_r144546384
--- Diff:
flink-filesystems/flink-s3-fs-hadoop/src/main/java/org/apache/flink/fs/s3hadoop/S3FileSystemFactory.java
---
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.fs.s3hadoop;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.FileSystemFactory;
+import org.apache.flink.runtime.fs.hdfs.HadoopFileSystem;
+import org.apache.flink.runtime.util.HadoopUtils;
+
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+
+/**
+ * Simple factory for the S3 file system.
+ */
+public class S3FileSystemFactory implements FileSystemFactory {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(S3FileSystemFactory.class);
+
+ /** The prefixes that Flink adds to the Hadoop config under 'fs.s3a.'.
*/
+ private static final String[] CONFIG_PREFIXES = { "s3.", "s3a.",
"fs.s3a." };
+
+ /** Keys that are replaced (after prefix replacement, to give a more
uniform experience
+ * across different file system implementations. */
+ private static final String[][] MIRRORED_CONFIG_KEYS = {
+ { "fs.s3a.access-key", "fs.s3a.access.key" },
+ { "fs.s3a.secret-key", "fs.s3a.secret.key" }
--- End diff --
Thanks, I think we will probably add more keys here over time.
---
