Aws-POC merge to develop

.gitignore

@@ -61,6 +61,7 @@ myTestCheckpoints
 **/main/resources/application.yml
 **/main/resources/spline.properties
 **/main/resources/kafka.properties
+**/menas/*/main/resources/*.properties
 **/enceladus_env.sh
 *.config
 *.keytab

dao/src/main/scala/za/co/absa/enceladus/dao/auth/MenasCredentials.scala

@@ -18,7 +18,7 @@ package za.co.absa.enceladus.dao.auth
 import com.typesafe.config.ConfigFactory
 import org.apache.spark.sql.SparkSession
 import sun.security.krb5.internal.ktab.KeyTab
-import za.co.absa.enceladus.utils.fs.FileSystemVersionUtils
+import za.co.absa.enceladus.utils.fs.{FileSystemUtils, HadoopFsUtils}
 
 sealed abstract class MenasCredentials {
   val username: String
@@ -40,9 +40,9 @@ object MenasPlainCredentials {
     * @return An instance of Menas Credentials.
     */
   def fromFile(path: String)(implicit spark: SparkSession): MenasPlainCredentials = {
-    val fsUtils = new FileSystemVersionUtils(spark.sparkContext.hadoopConfiguration)
+    val fs =  FileSystemUtils.getFileSystemFromPath(path)(spark.sparkContext.hadoopConfiguration)
 
-    val conf = ConfigFactory.parseString(fsUtils.getFileContent(path))
+    val conf = ConfigFactory.parseString(HadoopFsUtils.getOrCreate(fs).getLocalOrDistributedFileContent(path))
     MenasPlainCredentials(conf.getString("username"), conf.getString("password"))
   }
 }
@@ -55,9 +55,9 @@ object MenasKerberosCredentials {
     * @return An instance of Menas Credentials.
     */
   def fromFile(path: String)(implicit spark: SparkSession): MenasKerberosCredentials = {
-    val fsUtils = new FileSystemVersionUtils(spark.sparkContext.hadoopConfiguration)
+    val fs =  FileSystemUtils.getFileSystemFromPath(path)(spark.sparkContext.hadoopConfiguration)
 
-    val localKeyTabPath = fsUtils.getLocalPathToFile(path)
+    val localKeyTabPath = HadoopFsUtils.getOrCreate(fs).getLocalPathToFileOrCopyToLocal(path)
     val keytab = KeyTab.getInstance(localKeyTabPath)
     val username = keytab.getOneName.getName
 

dao/src/test/scala/za/co/absa/enceladus/dao/rest/BaseTestSuite.scala

@@ -15,10 +15,12 @@
 
 package za.co.absa.enceladus.dao.rest
 
-import org.scalatest.mockito.MockitoSugar
-import org.scalatest.{BeforeAndAfter, Matchers, WordSpec}
+import org.mockito.scalatest.MockitoSugar
+import org.scalatest.wordspec.AnyWordSpec
+import org.scalatest.BeforeAndAfter
+import org.scalatest.matchers.should.Matchers
 
-abstract class BaseTestSuite extends WordSpec
+abstract class BaseTestSuite extends AnyWordSpec
   with Matchers
   with MockitoSugar
   with BeforeAndAfter

dao/src/test/scala/za/co/absa/enceladus/dao/rest/RestDaoFactorySuite.scala

@@ -15,11 +15,12 @@
 
 package za.co.absa.enceladus.dao.rest
 
-import org.scalatest.{Matchers, WordSpec}
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.enceladus.dao.UnauthorizedException
 import za.co.absa.enceladus.dao.auth.{InvalidMenasCredentials, MenasKerberosCredentials, MenasPlainCredentials}
 
-class RestDaoFactorySuite extends WordSpec with Matchers {
+class RestDaoFactorySuite extends AnyWordSpec with Matchers {
 
   private val menasApiBaseUrls = List("http://localhost:8080/menas/api")
 

dao/src/test/scala/za/co/absa/enceladus/dao/rest/auth/AuthClientSuite.scala

@@ -16,15 +16,17 @@
 package za.co.absa.enceladus.dao.rest.auth
 
 import org.mockito.stubbing.OngoingStubbing
-import org.scalatest.mockito.MockitoSugar
-import org.scalatest.{BeforeAndAfter, Matchers, WordSpec}
+import org.scalatest.matchers.should.Matchers
+import org.mockito.scalatest.MockitoSugar
+import org.scalatest.wordspec.AnyWordSpec
+import org.scalatest.BeforeAndAfter
 import org.springframework.http.{HttpHeaders, ResponseEntity}
 import org.springframework.util.LinkedMultiValueMap
 import org.springframework.web.client.RestTemplate
 import za.co.absa.enceladus.dao.UnauthorizedException
 import za.co.absa.enceladus.dao.rest.{ApiCaller, ApiCallerStub, AuthClient}
 
-abstract class AuthClientSuite() extends WordSpec
+abstract class AuthClientSuite() extends AnyWordSpec
   with Matchers
   with MockitoSugar
   with BeforeAndAfter {

dao/src/test/scala/za/co/absa/enceladus/dao/rest/auth/MenasPlainCredentialsSuite.scala

@@ -15,12 +15,12 @@
 
 package za.co.absa.enceladus.dao.rest.auth
 
-import org.scalatest.WordSpec
+import org.scalatest.wordspec.AnyWordSpec
 import za.co.absa.enceladus.dao.auth.MenasPlainCredentials
-import za.co.absa.enceladus.utils.fs.FileSystemVersionUtils
+import za.co.absa.enceladus.utils.fs.LocalFsUtils
 import za.co.absa.enceladus.utils.testUtils.SparkTestBase
 
-class MenasPlainCredentialsSuite extends WordSpec with SparkTestBase {
+class MenasPlainCredentialsSuite extends AnyWordSpec with SparkTestBase {
 
   "MenasPlainCredentials" should {
     "be read from *.conf" in {
@@ -42,9 +42,7 @@ class MenasPlainCredentialsSuite extends WordSpec with SparkTestBase {
       val homeDir = System.getProperty("user.home")
       val expected = s"$homeDir/dir/file"
 
-      val fsUtils = new FileSystemVersionUtils(spark.sparkContext.hadoopConfiguration)
-
-      val actual = fsUtils.replaceHome("~/dir/file")
+      val actual = LocalFsUtils.replaceHome("~/dir/file")
       assert(actual == expected)
     }
   }

data-model/src/main/scala/za/co/absa/enceladus/model/Dataset.scala

@@ -90,11 +90,10 @@ case class Dataset(name: String,
 
   private def substituteMappingConformanceRuleCharacter(dataset: Dataset, from: Char, to: Char): Dataset = {
     val conformanceRules = dataset.conformance.map {
-      case m: MappingConformanceRule => {
+      case m: MappingConformanceRule =>
         m.copy(attributeMappings = m.attributeMappings.map(key => {
           (key._1.replace(from, to), key._2)
         }))
-      }
       case c: ConformanceRule => c
     }
 

data-model/src/test/scala/za/co/absa/enceladus/model/DatasetTest.scala

@@ -15,10 +15,10 @@
 
 package za.co.absa.enceladus.model
 
-import org.scalatest.FunSuite
+import org.scalatest.funsuite.AnyFunSuite
 import za.co.absa.enceladus.model.conformanceRule.LiteralConformanceRule
 
-class DatasetTest extends FunSuite {
+class DatasetTest extends AnyFunSuite {
   private val conformanceRule1 = LiteralConformanceRule(
     order = 0,
     controlCheckpoint = true,

data-model/src/test/scala/za/co/absa/enceladus/model/MappingTableTest.scala

@@ -15,9 +15,10 @@
 
 package za.co.absa.enceladus.model
 
-import org.scalatest.FunSuite
+import org.scalatest.funsuite.AnyFunSuite
 
-class MappingTableTest extends FunSuite {
+
+class MappingTableTest extends AnyFunSuite {
   private val defaultValue1 = DefaultValue("someColumn", "alfa")
   private val defaultValue2 = DefaultValue("otherColumn", "beta")
 

data-model/src/test/scala/za/co/absa/enceladus/model/PropertyDefinitionTest.scala

@@ -15,12 +15,12 @@
 
 package za.co.absa.enceladus.model
 
-import org.scalatest.FunSuite
+import org.scalatest.funsuite.AnyFunSuite
 import za.co.absa.enceladus.model.properties.PropertyDefinition
 import za.co.absa.enceladus.model.properties.essentiality.Mandatory
 import za.co.absa.enceladus.model.properties.propertyType.{EnumPropertyType, PropertyTypeValidationException, StringPropertyType}
 
-class PropertyDefinitionTest extends FunSuite {
+class PropertyDefinitionTest extends AnyFunSuite {
 
   private val modelVersion = ModelVersion
 

data-model/src/test/scala/za/co/absa/enceladus/model/SchemaFieldTest.scala

@@ -15,9 +15,9 @@
 
 package za.co.absa.enceladus.model
 
-import org.scalatest.FunSuite
+import org.scalatest.funsuite.AnyFunSuite
 
-class SchemaFieldTest extends FunSuite {
+class SchemaFieldTest extends AnyFunSuite {
   private val schemaFieldChildSecondLevel = SchemaField(
     name = "String",
     `type` = "string",

data-model/src/test/scala/za/co/absa/enceladus/model/SchemaTest.scala

@@ -15,9 +15,9 @@
 
 package za.co.absa.enceladus.model
 
-import org.scalatest.FunSuite
+import org.scalatest.funsuite.AnyFunSuite
 
-class SchemaTest extends FunSuite {
+class SchemaTest extends AnyFunSuite {
   private val schemaField = SchemaField(
     name = "someField",
     `type` = "StringType",

data-model/src/test/scala/za/co/absa/enceladus/model/conformanceRule/ConformanceRuleTest.scala

@@ -17,9 +17,10 @@ package za.co.absa.enceladus.model.conformanceRule
 
 import com.fasterxml.jackson.databind.{ObjectMapper, SerializationFeature}
 import com.fasterxml.jackson.module.scala.DefaultScalaModule
-import org.scalatest.{Matchers, WordSpec}
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
 
-class ConformanceRuleTest extends WordSpec with Matchers {
+class ConformanceRuleTest extends AnyWordSpec with Matchers {
 
   private val objectMapper = new ObjectMapper()
     .registerModule(DefaultScalaModule)

data-model/src/test/scala/za/co/absa/enceladus/model/dataFrameFilter/DataFrameFilterSuite.scala

@@ -1,10 +1,10 @@
 package za.co.absa.enceladus.model.dataFrameFilter
 
-import org.scalatest.FunSuite
 import org.apache.spark.sql.functions.{col, lit, not => columnNot}
 import org.apache.spark.sql.types._
+import org.scalatest.funsuite.AnyFunSuite
 
-class DataFrameFilterSuite extends FunSuite {
+class DataFrameFilterSuite extends AnyFunSuite {
   test("Filter for equal value") {
     val filterExpr = EqualsFilter("column1", "v").filter.expr
     val expected = (col("column1") === lit("v")).expr

data-model/src/test/scala/za/co/absa/enceladus/model/menas/audit/AuditableTest.scala

@@ -15,12 +15,12 @@
 
 package za.co.absa.enceladus.model.menas.audit
 
-import org.scalatest.FunSuite
+import org.scalatest.funsuite.AnyFunSuite
 import za.co.absa.enceladus.model.Dataset
 import za.co.absa.enceladus.model.conformanceRule.{DropConformanceRule, LiteralConformanceRule}
 import za.co.absa.enceladus.model.conformanceRule.ConformanceRule
 
-class AuditableTest extends FunSuite {
+class AuditableTest extends AnyFunSuite {
   val obj1 = Dataset(name = "Test DS",
     version = 0,
     hdfsPath = "oldPath",

examples/src/main/scala/za/co/absa/enceladus/examples/CustomRuleSample1.scala

@@ -25,7 +25,7 @@ import za.co.absa.enceladus.examples.interpreter.rules.custom.UppercaseCustomCon
 import za.co.absa.enceladus.model.Dataset
 import za.co.absa.enceladus.utils.time.TimeZoneNormalizer
 
-object CustomRuleSample1 {
+object CustomRuleSample1 extends CustomRuleSampleFs {
 
   case class ExampleRow(id: Int, makeUpper: String, leave: String)
   case class OutputRow(id: Int, makeUpper: String, leave: String, doneUpper: String)
@@ -78,7 +78,7 @@ object CustomRuleSample1 {
       .setCatalystWorkaroundEnabled(isCatalystWorkaroundEnabled)
       .setControlFrameworkEnabled(enableCF)
 
-    val outputData: DataFrame = DynamicInterpreter.interpret(conformanceDef, inputData)
+    val outputData: DataFrame = DynamicInterpreter().interpret(conformanceDef, inputData)
 
     outputData.show(false)
     //scalastyle:on magicnumber

examples/src/main/scala/za/co/absa/enceladus/examples/CustomRuleSample2.scala

@@ -26,7 +26,7 @@ import za.co.absa.enceladus.examples.interpreter.rules.custom.LPadCustomConforma
 import za.co.absa.enceladus.model.Dataset
 import za.co.absa.enceladus.utils.time.TimeZoneNormalizer
 
-object CustomRuleSample2 {
+object CustomRuleSample2 extends CustomRuleSampleFs {
 
   case class ExampleRow(id: Int, addPad: String, leave: String)
   case class OutputRow(id: Int, addPad: String, leave: String, donePad: String)
@@ -81,7 +81,7 @@ object CustomRuleSample2 {
       .setCatalystWorkaroundEnabled(isCatalystWorkaroundEnabled)
       .setControlFrameworkEnabled(enableCF)
 
-    val outputData: DataFrame = DynamicInterpreter.interpret(conformanceDef, inputData)
+    val outputData: DataFrame = DynamicInterpreter().interpret(conformanceDef, inputData)
 
     outputData.show(false)
     // scalastyle:on magic.number

examples/src/main/scala/za/co/absa/enceladus/examples/CustomRuleSample3.scala

@@ -26,7 +26,7 @@ import za.co.absa.enceladus.examples.interpreter.rules.custom.{LPadCustomConform
 import za.co.absa.enceladus.model.Dataset
 import za.co.absa.enceladus.utils.time.TimeZoneNormalizer
 
-object CustomRuleSample3 {
+object CustomRuleSample3 extends CustomRuleSampleFs {
   implicit val spark: SparkSession = SparkSession.builder
     .master("local[*]")
     .appName("CustomRuleSample3")
@@ -79,7 +79,7 @@ object CustomRuleSample3 {
       .setCatalystWorkaroundEnabled(isCatalystWorkaroundEnabled)
       .setControlFrameworkEnabled(enableCF)
 
-    val outputData: DataFrame = DynamicInterpreter.interpret(conformanceDef, inputData)
+    val outputData: DataFrame = DynamicInterpreter().interpret(conformanceDef, inputData)
 
     outputData.show()
   }

examples/src/main/scala/za/co/absa/enceladus/examples/CustomRuleSample4.scala

@@ -28,7 +28,7 @@ import za.co.absa.enceladus.examples.interpreter.rules.custom.{LPadCustomConform
 import za.co.absa.enceladus.model.Dataset
 import za.co.absa.enceladus.utils.time.TimeZoneNormalizer
 
-object CustomRuleSample4 {
+object CustomRuleSample4 extends CustomRuleSampleFs {
   TimeZoneNormalizer.normalizeJVMTimeZone() //normalize JVM time zone as soon as possible
 
   /**
@@ -135,9 +135,10 @@ object CustomRuleSample4 {
     result
   }
 
+  implicit val spark: SparkSession = buildSparkSession()
+
   def main(args: Array[String]): Unit = {
     val cmd: CmdConfigLocal = getCmdLineArguments(args)
-    implicit val spark: SparkSession = buildSparkSession()
 
     val conf = ConfigFactory.load()
     val menasBaseUrls = MenasConnectionStringParser.parse(conf.getString("menas.rest.uri"))
@@ -186,7 +187,7 @@ object CustomRuleSample4 {
       .setCatalystWorkaroundEnabled(true)
       .setControlFrameworkEnabled(false)
 
-    val outputData: DataFrame = DynamicInterpreter.interpret(conformanceDef, inputData)
+    val outputData: DataFrame = DynamicInterpreter().interpret(conformanceDef, inputData)
     outputData.show()
     saveToCsv(outputData, cmd.outPath)
   }

examples/src/main/scala/za/co/absa/enceladus/examples/CustomRuleSampleFs.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.examples
+
+import org.apache.hadoop.fs.FileSystem
+import org.apache.spark.sql.SparkSession
+import za.co.absa.enceladus.utils.fs.HadoopFsUtils
+
+trait CustomRuleSampleFs {
+  def spark: SparkSession
+
+  implicit val fs: FileSystem = FileSystem.get(spark.sparkContext.hadoopConfiguration)
+  implicit val fsUtils: HadoopFsUtils = HadoopFsUtils.getOrCreate(fs)
+}