同一段代码,在main里面可以正常正常,在Test里面却直接结束
StreamExecutionEnvironment bsEnv =
StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings bsSettings =
EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment bsTableEnv =
StreamTableEnvironment.create(bsEnv, bsSettings);
bsTableEnv.executeSql(
DDLSourceSQLManager.createStreamFromKafka("localhost:9092",
"test",
"test",
"test",
"json"));
bsTableEnv.executeSql(com.ddlsql.DDLSourceSQLManager.createDynamicPrintlnRetractSinkTbl("printlnRetractSink"));
bsTableEnv.executeSql("insert into printlnRetractSink select
msg,count(*) as cnt from test group by msg");
--
Sent from:
http://apache-flink.147419.n8.nabble.com/