FlinkStreamSqlInterpreter extends FlinkSqlInterrpeter { public void cancel(InterpreterContext context) throws InterpreterException { this.flinkInterpreter.cancel(context); } FlinkStreamSqlInterpreter(Properties properties); @Override void open(); @Override void close(); @Override void callInnerSelect(String sql, InterpreterContext context); @Override void callInsertInto(String sql, InterpreterContext context); void cancel(InterpreterContext context); @Override Interpreter.FormType getFormType(); @Override int getProgress(InterpreterContext context); @Override Scheduler getScheduler();  }
@Test public void testCancelStreamSql() throws IOException, InterpreterException, InterruptedException, TimeoutException { String initStreamScalaScript = getInitStreamScript(1000); InterpreterResult result = flinkInterpreter.interpret(initStreamScalaScript, getInterpreterContext()); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); final Waiter waiter = new Waiter(); Thread thread = new Thread(() -> { try { InterpreterContext context = getInterpreterContext(); context.getLocalProperties().put("type", "update"); InterpreterResult result2 = sqlInterpreter.interpret("select url, count(1) as pv from " + "log group by url", context); waiter.assertTrue(context.out.toString().contains("Job was cancelled")); waiter.assertEquals(InterpreterResult.Code.ERROR, result2.code()); } catch (Exception e) { e.printStackTrace(); waiter.fail("Should not fail here"); } waiter.resume(); }); thread.start(); Thread.sleep(10 * 1000); InterpreterContext context = getInterpreterContext(); context.getLocalProperties().put("type", "update"); sqlInterpreter.cancel(context); waiter.await(10 * 1000); sqlInterpreter.interpret("select url, count(1) as pv from " + "log group by url", context); assertEquals(InterpreterResult.Code.SUCCESS, result.code()); List<InterpreterResultMessage> resultMessages = context.out.toInterpreterResultMessage(); assertEquals(InterpreterResult.Type.TABLE, resultMessages.get(0).getType()); assertTrue(resultMessages.toString(), resultMessages.get(0).getData().contains("url\tpv\n")); }