Preserve S3-compatible filesystem schemes

utils/src/main/scala/za/co/absa/enceladus/utils/fs/FileSystemUtils.scala

@@ -29,6 +29,7 @@ import scala.util.{Failure, Success, Try}
 object FileSystemUtils {
 
   val log: Logger = LoggerFactory.getLogger(this.getClass)
+  private val SchemeSeparator = "://"
 
   /**
    * Will yeild a [[FileSystem]] for path. If path prefix suggest S3, S3 FS is returned, HDFS otherwise.
@@ -40,13 +41,17 @@ object FileSystemUtils {
     path.toSimpleS3Location match {
 
       case Some(s3Location) => // s3 over hadoop fs api
-        val s3BucketUri: String = s"s3://${s3Location.bucketName}" // s3://<bucket>
-        val s3uri: URI = new URI(s3BucketUri)
+        val s3uri: URI = s3BucketUri(path, s3Location.bucketName)
         FileSystem.get(s3uri, hadoopConf)
 
       case None =>
         FileSystem.get(hadoopConf) // HDFS
     }
   }
-}
 
+  private[fs] def s3BucketUri(path: String, bucketName: String): URI = {
+    val separatorIndex = path.indexOf(SchemeSeparator)
+    val scheme = if (separatorIndex >= 0) path.substring(0, separatorIndex) else "s3"
+    new URI(s"$scheme://$bucketName")
+  }
+}

utils/src/test/scala/za/co/absa/enceladus/utils/fs/FileSystemUtilsSuite.scala

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed 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 za.co.absa.enceladus.utils.fs
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class FileSystemUtilsSuite extends AnyFunSuite {
+
+  test("s3BucketUri preserves S3-compatible path schemes") {
+    assert(FileSystemUtils.s3BucketUri("s3://bucket/path/to/file", "bucket").toString == "s3://bucket")
+    assert(FileSystemUtils.s3BucketUri("s3a://bucket/path/to/file", "bucket").toString == "s3a://bucket")
+    assert(FileSystemUtils.s3BucketUri("s3n://bucket/path/to/file", "bucket").toString == "s3n://bucket")
+  }
+}