[SPARK-13584][SQL][TESTS] Make ContinuousQueryManagerSuite not output logs to the console

## What changes were proposed in this pull request?

Make ContinuousQueryManagerSuite not output logs to the console. The logs will still output to `unit-tests.log`.

I also updated `SQLListenerMemoryLeakSuite` to use `quietly` to avoid changing the log level which won't output logs to `unit-tests.log`.

## How was this patch tested?

Just check Jenkins output.

Author: Shixiong Zhu <shixiong@databricks.com>

Closes #11439 from zsxwing/quietly-ContinuousQueryManagerSuite.

sql/core/src/test/resources/log4j.properties

@@ -23,6 +23,7 @@ log4j.appender.CA=org.apache.log4j.ConsoleAppender
 log4j.appender.CA.layout=org.apache.log4j.PatternLayout
 log4j.appender.CA.layout.ConversionPattern=%d{HH:mm:ss.SSS} %p %c: %m%n
 log4j.appender.CA.Threshold = WARN
+log4j.appender.CA.follow = true
 
 
 #File Appender

sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLListenerSuite.scala

@@ -25,6 +25,7 @@ import org.apache.spark.{SparkConf, SparkContext, SparkException, SparkFunSuite}
 import org.apache.spark.executor.TaskMetrics
 import org.apache.spark.scheduler._
 import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spark.sql.catalyst.util.quietly
 import org.apache.spark.sql.execution.{SparkPlanInfo, SQLExecution}
 import org.apache.spark.sql.execution.metric.{LongSQLMetricValue, SQLMetrics}
 import org.apache.spark.sql.test.SharedSQLContext
@@ -376,9 +377,7 @@ class SQLListenerSuite extends SparkFunSuite with SharedSQLContext {
 class SQLListenerMemoryLeakSuite extends SparkFunSuite {
 
   test("no memory leak") {
-    val oldLogLevel = org.apache.log4j.Logger.getRootLogger().getLevel()
-    try {
-      org.apache.log4j.Logger.getRootLogger().setLevel(org.apache.log4j.Level.FATAL)
+    quietly {
       val conf = new SparkConf()
         .setMaster("local")
         .setAppName("test")
@@ -413,8 +412,6 @@ class SQLListenerMemoryLeakSuite extends SparkFunSuite {
       } finally {
         sc.stop()
       }
-    } finally {
-      org.apache.log4j.Logger.getRootLogger().setLevel(oldLogLevel)
     }
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/streaming/ContinuousQueryManagerSuite.scala

@@ -49,7 +49,7 @@ class ContinuousQueryManagerSuite extends StreamTest with SharedSQLContext with
     sqlContext.streams.resetTerminated()
   }
 
-  test("listing") {
+  testQuietly("listing") {
     val (m1, ds1) = makeDataset
     val (m2, ds2) = makeDataset
     val (m3, ds3) = makeDataset
@@ -83,7 +83,7 @@ class ContinuousQueryManagerSuite extends StreamTest with SharedSQLContext with
         require(!q2.isActive)
         require(q2.exception.isDefined)
       }
-      val ex2 = withClue("no error while getting non-active query") {
+      withClue("no error while getting non-active query") {
         intercept[IllegalArgumentException] {
           sqlContext.streams.get(q2.name).eq(q2)
         }
@@ -93,7 +93,7 @@ class ContinuousQueryManagerSuite extends StreamTest with SharedSQLContext with
     }
   }
 
-  test("awaitAnyTermination without timeout and resetTerminated") {
+  testQuietly("awaitAnyTermination without timeout and resetTerminated") {
     val datasets = Seq.fill(5)(makeDataset._2)
     withQueriesOn(datasets: _*) { queries =>
       require(queries.size === datasets.size)
@@ -139,7 +139,7 @@ class ContinuousQueryManagerSuite extends StreamTest with SharedSQLContext with
     }
   }
 
-  test("awaitAnyTermination with timeout and resetTerminated") {
+  testQuietly("awaitAnyTermination with timeout and resetTerminated") {
     val datasets = Seq.fill(6)(makeDataset._2)
     withQueriesOn(datasets: _*) { queries =>
       require(queries.size === datasets.size)

sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala

@@ -214,6 +214,20 @@ private[sql] trait SQLTestUtils
   protected implicit def logicalPlanToSparkQuery(plan: LogicalPlan): DataFrame = {
     DataFrame(sqlContext, plan)
   }
+
+  /**
+   * Disable stdout and stderr when running the test. To not output the logs to the console,
+   * ConsoleAppender's `follow` should be set to `true` so that it will honors reassignments of
+   * System.out or System.err. Otherwise, ConsoleAppender will still output to the console even if
+   * we change System.out and System.err.
+   */
+  protected def testQuietly(name: String)(f: => Unit): Unit = {
+    test(name) {
+      quietly {
+        f
+      }
+    }
+  }
 }
 
 private[sql] object SQLTestUtils {