[ZEPPELIN-4870][hotfix] fix edge case, and remove duplicate parser

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java

@@ -194,40 +194,10 @@ public void parseText() {
     // parse text to get interpreter component
     if (this.text != null) {
       // clean localProperties, otherwise previous localProperties will be used for the next run
-      this.localProperties.clear();
-      Matcher matcher = REPL_PATTERN.matcher(this.text);
-      if (matcher.matches()) {
-        String headingSpace = matcher.group(1);
-        setIntpText(matcher.group(2));
-
-        if (matcher.groupCount() == 3 && matcher.group(3) != null) {
-          String localPropertiesText = matcher.group(3);
-          String[] splits = localPropertiesText.substring(1, localPropertiesText.length() -1)
-              .split(",");
-          for (String split : splits) {
-            String[] kv = split.split("=");
-            if (StringUtils.isBlank(split) || kv.length == 0) {
-              continue;
-            }
-            if (kv.length > 2) {
-              throw new RuntimeException("Invalid paragraph properties format: " + split);
-            }
-            if (kv.length == 1) {
-              localProperties.put(kv[0].trim(), kv[0].trim());
-            } else {
-              localProperties.put(kv[0].trim(), kv[1].trim());
-            }
-          }
-          this.scriptText = this.text.substring(headingSpace.length() + intpText.length() +
-              localPropertiesText.length() + 1).trim();
-        } else {
-          this.scriptText = this.text.substring(headingSpace.length() + intpText.length() + 1).trim();
-        }
-        config.putAll(localProperties);
-      } else {
-        setIntpText("");
-        this.scriptText = this.text.trim();
-      }
+      ParagraphTextParser.ParseResult result = ParagraphTextParser.parse(this.text);
+      localProperties = result.getLocalProperties();
+      setIntpText(result.getIntpText());
+      this.scriptText = result.getScriptText();
     }
   }
 

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/ParagraphTextParser.java

@@ -122,6 +122,10 @@ private static int parseLocalProperties(
           if (insideQuotes) {
             sb.append(ch);
           } else {
+            if (!parseKey) {
+              throw new RuntimeException(
+                      "Invalid paragraph properties format");
+            }
             propKey = sb.toString().trim();
             sb.delete(0, sb.length());
             parseKey = false;
@@ -141,10 +145,10 @@ private static int parseLocalProperties(
             } else {
               localProperties.put(propKey, sb.toString().trim());
             }
+            propKey = null;
+            parseKey = true;
+            sb.delete(0, sb.length());
           }
-          propKey = null;
-          parseKey = true;
-          sb.delete(0, sb.length());
           break;
         }
         default:
@@ -169,14 +173,15 @@ public static ParseResult parse(String text) {
       String headingSpace = matcher.group(1);
       intpText = matcher.group(2);
       int startPos = headingSpace.length() + intpText.length() + 1;
-      if (text.charAt(startPos) == '(') {
+      if (startPos < text.length() && text.charAt(startPos) == '(') {
         startPos = parseLocalProperties(text, startPos, localProperties);
       }
       scriptText = text.substring(startPos).trim();
     } else {
       intpText = "";
-      scriptText = text;
+      scriptText = text.trim();
     }
+
     return new ParseResult(intpText, scriptText, localProperties);
   }
 

zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/ParagraphTest.java

@@ -77,7 +77,7 @@ public class ParagraphTest extends AbstractInterpreterTest {
   public void scriptBodyWithReplName() {
     Note note = createNote();
     Paragraph paragraph = new Paragraph(note, null);
-    paragraph.setText("%test(1234567");
+    paragraph.setText("%test (1234567");
     assertEquals("test", paragraph.getIntpText());
     assertEquals("(1234567", paragraph.getScriptText());
 

zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/ParagraphTextParserTest.java

@@ -34,6 +34,18 @@ public void testJupyter() {
     assertEquals("", parseResult.getScriptText());
   }
 
+
+  @Test
+  public void testCassandra() {
+    ParagraphTextParser.ParseResult parseResult = ParagraphTextParser.parse(
+            "%cassandra(locale=ru_RU, timeFormat=\"E, d MMM yy\", floatPrecision = 5, output=cql)\n"
+                    + "select * from system_auth.roles;");
+    assertEquals("cassandra", parseResult.getIntpText());
+    assertEquals(4, parseResult.getLocalProperties().size());
+    assertEquals("E, d MMM yy", parseResult.getLocalProperties().get("timeFormat"));
+    assertEquals("select * from system_auth.roles;", parseResult.getScriptText());
+  }
+
   @Test
   public void testParagraphTextLocalPropertiesAndText() {
     ParagraphTextParser.ParseResult parseResult = ParagraphTextParser.parse("%spark.pyspark(pool=pool_1) sc.version");