From b6997d99ba2bda38c96181297654799ea91b6bec Mon Sep 17 00:00:00 2001
From: Jeff Zhang <zjffdu@apache.org>
Date: Fri, 5 Jun 2020 22:43:22 +0800
Subject: [PATCH] [ZEPPELIN-4848]. Switch for enabling Spark REPL output for
 each paragraph

---
 .../zeppelin/spark/SparkInterpreterTest.java  | 44 +++++++++++++++++++
 .../spark/BaseSparkScalaInterpreter.scala     |  7 ++-
 2 files changed, 50 insertions(+), 1 deletion(-)

diff --git a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java
index f3f9dec9b41..711635f6d83 100644
--- a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java
+++ b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/SparkInterpreterTest.java
@@ -418,6 +418,50 @@ public void testDisableReplOutput() throws InterpreterException {
     assertEquals("hello world", output);
   }
 
+  @Test
+  public void testDisableReplOutputForParagraph() throws InterpreterException {
+    Properties properties = new Properties();
+    properties.setProperty("spark.master", "local");
+    properties.setProperty("spark.app.name", "test");
+    properties.setProperty("zeppelin.spark.maxResult", "100");
+    properties.setProperty("zeppelin.spark.printREPLOutput", "true");
+    // disable color output for easy testing
+    properties.setProperty("zeppelin.spark.scala.color", "false");
+    properties.setProperty("zeppelin.spark.deprecatedMsg.show", "false");
+
+    InterpreterContext.set(getInterpreterContext());
+    interpreter = new SparkInterpreter(properties);
+    interpreter.setInterpreterGroup(mock(InterpreterGroup.class));
+    interpreter.open();
+
+    InterpreterResult result = interpreter.interpret("val a=\"hello world\"", getInterpreterContext());
+    assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+    assertEquals("a: String = hello world\n", output);
+
+    result = interpreter.interpret("print(a)", getInterpreterContext());
+    assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+    // output from print statement will still be displayed
+    assertEquals("hello world", output);
+
+    // disable REPL output
+    InterpreterContext context = getInterpreterContext();
+    context.getLocalProperties().put("printREPLOutput", "false");
+    result = interpreter.interpret("print(a)", context);
+    assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+    // output from print statement will disappear
+    assertEquals("", output);
+
+    // REPL output get back if we don't set printREPLOutput in paragraph local properties
+    result = interpreter.interpret("val a=\"hello world\"", getInterpreterContext());
+    assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+    assertEquals("a: String = hello world\n", output);
+
+    result = interpreter.interpret("print(a)", getInterpreterContext());
+    assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+    // output from print statement will still be displayed
+    assertEquals("hello world", output);
+  }
+
   @Test
   public void testSchedulePool() throws InterpreterException {
     Properties properties = new Properties();
diff --git a/spark/spark-scala-parent/src/main/scala/org/apache/zeppelin/spark/BaseSparkScalaInterpreter.scala b/spark/spark-scala-parent/src/main/scala/org/apache/zeppelin/spark/BaseSparkScalaInterpreter.scala
index 0361c940c1e..f05acb05406 100644
--- a/spark/spark-scala-parent/src/main/scala/org/apache/zeppelin/spark/BaseSparkScalaInterpreter.scala
+++ b/spark/spark-scala-parent/src/main/scala/org/apache/zeppelin/spark/BaseSparkScalaInterpreter.scala
@@ -103,11 +103,16 @@ abstract class BaseSparkScalaInterpreter(val conf: SparkConf,
   def interpret(code: String, context: InterpreterContext): InterpreterResult = {
 
     val originalOut = System.out
+    val printREPLOutput = context.getStringLocalProperty("printREPLOutput", "true").toBoolean
 
     def _interpret(code: String): scala.tools.nsc.interpreter.Results.Result = {
       Console.withOut(interpreterOutput) {
         System.setOut(Console.out)
-        interpreterOutput.setInterpreterOutput(context.out)
+        if (printREPLOutput) {
+          interpreterOutput.setInterpreterOutput(context.out)
+        } else {
+          interpreterOutput.setInterpreterOutput(null)
+        }
         interpreterOutput.ignoreLeadingNewLinesFromScalaReporter()
 
         val status = scalaInterpret(code) match {