Feature/253 remove log4j2 (#256)

* Remove log4j2

* Log4j is not configured initially, only after some time

* Use Logging trait to ensure initialization of log config

Author: kevinwallimann

component-archetype/src/main/resources/archetype-resources/src/main/scala/reader/mycomponent/MyStreamReaderImpl.scala

@@ -18,7 +18,7 @@
 package ${package}.reader.mycomponent
 
 import org.apache.commons.configuration2.Configuration
-import org.apache.logging.log4j.LogManager
+import org.slf4j.LoggerFactory
 import org.apache.spark.sql.{DataFrame, SparkSession}
 import za.co.absa.hyperdrive.ingestor.api.reader.{StreamReader, StreamReaderFactory, StreamReaderFactoryProvider}
 import za.co.absa.hyperdrive.ingestor.api.{HasComponentAttributes, PropertyMetadata}
@@ -33,7 +33,7 @@ private[reader] class MyStreamReaderImpl() extends StreamReader {
 }
 
 object MyStreamReaderImpl extends StreamReaderFactory with MyStreamReaderImplAttributes {
-  private val logger = LogManager.getLogger
+  private val logger = LoggerFactory.getLogger(this.getClass)
 
   override def apply(conf: Configuration): StreamReader = {
     logger.info("Building MyStreamReaderImpl")

component-archetype/src/main/resources/archetype-resources/src/main/scala/transformer/mycomponent/MyStreamTransformerImpl.scala

@@ -18,7 +18,7 @@
 package ${package}.transformer.mycomponent
 
 import org.apache.commons.configuration2.Configuration
-import org.apache.logging.log4j.LogManager
+import org.slf4j.LoggerFactory
 import org.apache.spark.sql.DataFrame
 import za.co.absa.hyperdrive.ingestor.api.transformer.{StreamTransformer, StreamTransformerFactory, StreamTransformerFactoryProvider}
 import za.co.absa.hyperdrive.ingestor.api.{HasComponentAttributes, PropertyMetadata}
@@ -33,7 +33,7 @@ private[transformer] class MyStreamTransformerImpl() extends StreamTransformer {
 }
 
 object MyStreamTransformerImpl extends StreamTransformerFactory with MyStreamTransformerImplAttributes {
-  private val logger = LogManager.getLogger
+  private val logger = LoggerFactory.getLogger(this.getClass)
 
   override def apply(conf: Configuration): StreamTransformer = {
     logger.info("Building MyStreamTransformerImpl")

component-archetype/src/main/resources/archetype-resources/src/main/scala/writer/mycomponent/MyStreamWriterImpl.scala

@@ -18,7 +18,7 @@
 package ${package}.writer.mycomponent
 
 import org.apache.commons.configuration2.Configuration
-import org.apache.logging.log4j.LogManager
+import org.slf4j.LoggerFactory
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.streaming.StreamingQuery
 import za.co.absa.hyperdrive.ingestor.api.writer.{StreamWriter, StreamWriterFactory, StreamWriterFactoryProvider}
@@ -34,7 +34,7 @@ private[writer] class MyStreamWriterImpl(val destination: String) extends Stream
 }
 
 object MyStreamWriterImpl extends StreamWriterFactory with MyStreamWriterImplAttributes {
-  private val logger = LogManager.getLogger
+  private val logger = LoggerFactory.getLogger(this.getClass)
 
   override def apply(conf: Configuration): StreamWriter = {
     logger.info("Building MyStreamWriterImpl")

component-scanner/src/main/resources/log4j2.xml

@@ -19,7 +19,7 @@ import java.net.URLClassLoader
 import java.nio.file.{Files, Path}
 import java.util.ServiceLoader
 
-import org.apache.logging.log4j.LogManager
+import org.slf4j.LoggerFactory
 import za.co.absa.hyperdrive.ingestor.api.reader.{StreamReaderFactory, StreamReaderFactoryProvider}
 import za.co.absa.hyperdrive.ingestor.api.transformer.{StreamTransformerFactory, StreamTransformerFactoryProvider}
 import za.co.absa.hyperdrive.ingestor.api.writer.{StreamWriterFactory, StreamWriterFactoryProvider}
@@ -37,7 +37,7 @@ case class ComponentDescriptor(attributes: HasComponentAttributes,
                                jarPath: Path)
 
 object ComponentScanner {
-  private val logger = LogManager.getLogger
+  private val logger = LoggerFactory.getLogger(this.getClass)
   private val jarSuffix = ".jar"
 
   def getComponents(baseDirectory: Path): Try[ComponentDescriptors] = getComponents(List(baseDirectory))

component-scanner/src/main/scala/za/co/absa/hyperdrive/scanner/ComponentScanner.scala

@@ -16,7 +16,7 @@
 package za.co.absa.hyperdrive.driver
 
 import org.apache.commons.configuration2.Configuration
-import org.apache.logging.log4j.LogManager
+import org.slf4j.LoggerFactory
 import za.co.absa.hyperdrive.ingestor.api.reader.StreamReader
 import za.co.absa.hyperdrive.ingestor.api.transformer.StreamTransformer
 import za.co.absa.hyperdrive.ingestor.api.writer.StreamWriter
@@ -25,7 +25,7 @@ import za.co.absa.hyperdrive.ingestor.implementation.transformer.factories.Strea
 import za.co.absa.hyperdrive.ingestor.implementation.writer.factories.StreamWriterAbstractFactory
 
 private[driver] class IngestionDriver {
-  private val logger = LogManager.getLogger
+  private val logger = LoggerFactory.getLogger(this.getClass)
   val ListDelimiter = ','
 
   def ingest(configuration: Configuration): Unit = {

driver/src/main/resources/log4j2.xml

@@ -18,7 +18,7 @@ package za.co.absa.hyperdrive.driver
 import java.util.UUID
 
 import org.apache.commons.configuration2.Configuration
-import org.apache.logging.log4j.LogManager
+import org.slf4j.LoggerFactory
 import org.apache.spark.sql.SparkSession
 import za.co.absa.hyperdrive.driver.TerminationMethodEnum.{AwaitTermination, ProcessAllAvailable, TerminationMethod}
 import za.co.absa.hyperdrive.ingestor.api.reader.StreamReader
@@ -39,7 +39,7 @@ class SparkIngestor(val spark: SparkSession,
                     val awaitTerminationTimeout: Option[Long],
                     val conf: Configuration) {
 
-  private val logger = LogManager.getLogger
+  private val logger = LoggerFactory.getLogger(this.getClass)
 
   /**
    * This method performs the ingestion according to the components it receives.
@@ -100,7 +100,7 @@ class SparkIngestor(val spark: SparkSession,
 
 object SparkIngestor extends SparkIngestorAttributes {
 
-  private val logger = LogManager.getLogger
+  private val logger = LoggerFactory.getLogger(this.getClass)
 
   def apply(conf: Configuration): SparkIngestor = {
     ComponentFactoryUtil.validateConfiguration(conf, getProperties)

driver/src/main/scala/za/co/absa/hyperdrive/driver/IngestionDriver.scala

@@ -19,25 +19,24 @@ import org.apache.commons.configuration2.builder.BasicConfigurationBuilder
 import org.apache.commons.configuration2.builder.fluent.Parameters
 import org.apache.commons.configuration2.convert.DefaultListDelimiterHandler
 import org.apache.commons.configuration2.{BaseConfiguration, Configuration}
-import org.apache.logging.log4j.LogManager
+import org.apache.spark.internal.Logging
 import za.co.absa.hyperdrive.driver.IngestionDriver
 import za.co.absa.hyperdrive.driver.utils.DriverUtil
 
-object CommandLineIngestionDriver extends IngestionDriver {
+object CommandLineIngestionDriver extends IngestionDriver with Logging {
 
-  private val logger = LogManager.getLogger
   private val PropertyDelimiter = "="
 
   def main(args: Array[String]): Unit = {
     if (args.isEmpty) {
       throw new IllegalArgumentException("No configuration provided.")
     }
 
-    logger.info(s"Starting Hyperdrive ${DriverUtil.getVersionString}")
+    logInfo(s"Starting Hyperdrive ${DriverUtil.getVersionString}")
 
-    logger.info(s"Going to load ${args.length} configurations from command line.")
+    logInfo(s"Going to load ${args.length} configurations from command line.")
     val configuration = parseConfiguration(args)
-    logger.info("Configuration loaded. Going to invoke ingestion.")
+    logInfo("Configuration loaded. Going to invoke ingestion.")
     ingest(configuration)
   }
 

driver/src/main/scala/za/co/absa/hyperdrive/driver/SparkIngestor.scala

@@ -16,36 +16,33 @@
 package za.co.absa.hyperdrive.driver.drivers
 
 import java.nio.file.{Files, Paths}
-
 import org.apache.commons.configuration2.builder.FileBasedConfigurationBuilder
 import org.apache.commons.configuration2.builder.fluent.Parameters
 import org.apache.commons.configuration2.convert.DefaultListDelimiterHandler
 import org.apache.commons.configuration2.{Configuration, PropertiesConfiguration}
-import org.apache.logging.log4j.LogManager
+import org.apache.spark.internal.Logging
 import za.co.absa.hyperdrive.driver.IngestionDriver
 import za.co.absa.hyperdrive.driver.utils.DriverUtil
 
 /**
   * This driver launches ingestion by loading the configurations from a properties file.
   */
-object PropertiesIngestionDriver extends IngestionDriver {
-
-  private val logger = LogManager.getLogger
+object PropertiesIngestionDriver extends IngestionDriver with Logging {
 
   def main(args: Array[String]): Unit = {
     val propertiesFile = getPropertiesFilePath(args)
     if (propertiesFile.isEmpty) {
       throw new IllegalArgumentException("No properties file supplied.")
     }
-    logger.info(s"Starting Hyperdrive ${DriverUtil.getVersionString}")
+    logInfo(s"Starting Hyperdrive ${DriverUtil.getVersionString}")
 
     if (isInvalid(propertiesFile.get)) {
       throw new IllegalArgumentException(s"Invalid properties file: '${propertiesFile.get}'.")
     }
 
-    logger.info(s"Going to load ingestion configurations from '${propertiesFile.get}'.")
+    logInfo(s"Going to load ingestion configurations from '${propertiesFile.get}'.")
     val configurations = loadConfiguration(propertiesFile.get)
-    logger.info(s"Configurations loaded. Going to invoke ingestion: [$configurations]")
+    logInfo(s"Configurations loaded. Going to invoke ingestion: [$configurations]")
     ingest(configurations)
   }
 
@@ -63,7 +60,7 @@ object PropertiesIngestionDriver extends IngestionDriver {
       case v if v == 0 => None
       case v =>
         if (v > 1) {
-          logger.warn(s"Expected only properties file path, but got extra parameters. Returning first as the path. All parameters = [${args.mkString(",")}]")
+          logWarning(s"Expected only properties file path, but got extra parameters. Returning first as the path. All parameters = [${args.mkString(",")}]")
         }
         Some(args(0))
     }