[SPARK-26753][CORE] Fix for ensuring custom log levels for packages/classes work for spark-shell

This fix replaces the Threshold with a Filter for ConsoleAppender which checks
to ensure that either the logLevel is greater than thresholdLevel (shell log
level) or the log originated from a custom defined logger. In these cases, it
lets a log event go through, otherwise it doesn't.

Testing Done:
1. Ensured that custom log level works when set by default (via log4j.properties)
2. Ensured that logs are not printed twice when log level is changed by setLogLevel
3. Ensured that custom logs are printed when log level is changed back by setLogLevel

Author: ankuriitg

core/src/main/scala/org/apache/spark/internal/Logging.scala

@@ -25,7 +25,7 @@ import org.apache.log4j._
 import org.slf4j.{Logger, LoggerFactory}
 import org.slf4j.impl.StaticLoggerBinder
 
-import org.apache.spark.util.Utils
+import org.apache.spark.util.{SparkShellLoggingFilter, Utils}
 
 /**
  * Utility trait for classes that want to log data. Creates a SLF4J logger for the class and allows
@@ -156,14 +156,7 @@ trait Logging {
           }
           rootLogger.getAllAppenders().asScala.foreach {
             case ca: ConsoleAppender =>
-              Option(ca.getThreshold()) match {
-                case Some(t) =>
-                  Logging.consoleAppenderToThreshold.put(ca, t)
-                  if (!t.isGreaterOrEqual(replLevel)) {
-                    ca.setThreshold(replLevel)
-                  }
-                case None => ca.setThreshold(replLevel)
-              }
+              ca.addFilter(new SparkShellLoggingFilter(replLevel))
             case _ => // no-op
           }
         }
@@ -182,7 +175,6 @@ private[spark] object Logging {
   @volatile private var initialized = false
   @volatile private var defaultRootLevel: Level = null
   @volatile private var defaultSparkLog4jConfig = false
-  private val consoleAppenderToThreshold = new ConcurrentHashMap[ConsoleAppender, Priority]()
 
   val initLock = new Object()
   try {
@@ -213,7 +205,21 @@ private[spark] object Logging {
         rootLogger.setLevel(defaultRootLevel)
         rootLogger.getAllAppenders().asScala.foreach {
           case ca: ConsoleAppender =>
-            ca.setThreshold(consoleAppenderToThreshold.get(ca))
+            // SparkShellLoggingFilter is the last filter
+            ca.getFirstFilter() match {
+              case ssf: SparkShellLoggingFilter =>
+                ca.clearFilters()
+              case f: org.apache.log4j.spi.Filter =>
+                var previous = f
+                var current = previous.getNext()
+                while (current != null && !current.isInstanceOf[SparkShellLoggingFilter]) {
+                  previous = current;
+                  current = previous.getNext()
+                }
+                if (current != null) {
+                  previous.setNext(current.getNext())
+                }
+            }
           case _ => // no-op
         }
       }

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -53,6 +53,8 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
 import org.apache.hadoop.security.UserGroupInformation
 import org.apache.hadoop.yarn.conf.YarnConfiguration
+import org.apache.log4j.{Level, LogManager}
+import org.apache.log4j.spi.{Filter, LoggingEvent}
 import org.eclipse.jetty.util.MultiException
 import org.slf4j.Logger
 
@@ -2285,7 +2287,15 @@ private[spark] object Utils extends Logging {
     val rootLogger = org.apache.log4j.Logger.getRootLogger()
     rootLogger.setLevel(l)
     rootLogger.getAllAppenders().asScala.foreach {
-      case ca: org.apache.log4j.ConsoleAppender => ca.setThreshold(l)
+      case ca: org.apache.log4j.ConsoleAppender =>
+        var f = ca.getFirstFilter()
+        while (f != null) {
+          f match {
+            case ssf: SparkShellLoggingFilter =>
+              ssf.setThresholdLevel(l)
+          }
+          f = f.getNext()
+        }
       case _ => // no-op
     }
   }
@@ -2991,3 +3001,32 @@ private[spark] class CircularBuffer(sizeInBytes: Int = 10240) extends java.io.Ou
     new String(nonCircularBuffer, StandardCharsets.UTF_8)
   }
 }
+
+private[spark] class SparkShellLoggingFilter(var thresholdLevel: Level) extends Filter {
+
+  /**
+   * If log level of event is lower than thresholdLevel, then the decision is made based on
+   * whether the log came from root or some custom configuration
+   * @param loggingEvent
+   * @return decision for accept/deny log event
+   */
+  def decide(loggingEvent: LoggingEvent): Int = {
+    val rootLevel = LogManager.getRootLogger().getLevel()
+    if (loggingEvent.getLevel().isGreaterOrEqual(thresholdLevel) ||
+          !loggingEvent.getLevel().eq(rootLevel)) {
+      return Filter.NEUTRAL
+    }
+    var logger = loggingEvent.getLogger()
+    while(logger.getParent() != null) {
+      if (logger.getLevel() != null) {
+        return Filter.NEUTRAL
+      }
+      logger = logger.getParent()
+    }
+    return Filter.DENY
+  }
+
+  private[spark] def setThresholdLevel(level: Level): Unit = {
+    thresholdLevel = level
+  }
+}