[ZEPPELIN-4848]. Switch for enabling Spark REPL output for each paragraph

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();

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 {