@Override
  public InterpreterResult interpret(String st, InterpreterContext context)
      throws InterpreterException {
    LOGGER.debug("Interpret code: {}", st);
    this.z.setInterpreterContext(context);
    this.z.setGui(context.getGui());
    this.z.setNoteGui(context.getNoteGui());

    // set ClassLoader of current Thread to be the ClassLoader of Flink scala-shell,
    // otherwise codegen will fail to find classes defined in scala-shell
    ClassLoader originClassLoader = Thread.currentThread().getContextClassLoader();
    try {
      Thread.currentThread().setContextClassLoader(getFlinkScalaShellLoader());
      createPlannerAgain();
      setParallelismIfNecessary(context);
      setSavepointPathIfNecessary(context);
      return innerIntp.interpret(st, context);
    } finally {
      Thread.currentThread().setContextClassLoader(originClassLoader);
    }
  }
@Test
  void testBatchWordCount() throws InterpreterException, IOException {
    InterpreterContext context = getInterpreterContext();
    InterpreterResult result = interpreter.interpret(
            "val data = benv.fromElements(\"hello world\", \"hello flink\", \"hello hadoop\")",
        context);
    assertEquals(InterpreterResult.Code.SUCCESS, result.code());
    context = getInterpreterContext();
    result = interpreter.interpret(
            "data.flatMap(line => line.split(\"\\\\s\"))\n" +
            "  .map(w => (w, 1))\n" +
            "  .groupBy(0)\n" +
            "  .sum(1)\n" +
            "  .print()", context);
    assertEquals(InterpreterResult.Code.SUCCESS, result.code(), context.out.toString());

    String[] expectedCounts = {"(hello,3)", "(world,1)", "(flink,1)", "(hadoop,1)"};
    Arrays.sort(expectedCounts);

    String[] counts = context.out.toInterpreterResultMessage().get(0).getData().split("\n");
    Arrays.sort(counts);

    assertArrayEquals(expectedCounts, counts);
  }