[jira] [Updated] (FLINK-18769) MiniBatch doesn't work with FLIP-95 source

["Jark Wu (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-18769?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Jark Wu updated FLINK-18769:
----------------------------
    Summary: MiniBatch doesn't work with FLIP-95 source  (was: Streaming Table job stuck when enabling minibatching)

> MiniBatch doesn't work with FLIP-95 source
> ------------------------------------------
>
>                 Key: FLINK-18769
>                 URL: https://issues.apache.org/jira/browse/FLINK-18769
>             Project: Flink
>          Issue Type: Bug
>          Components: Table SQL / Runtime
>    Affects Versions: 1.11.1
>            Reporter: Nico Kruber
>            Assignee: Jark Wu
>            Priority: Critical
>             Fix For: 1.12.0, 1.11.2
>
>
> The following Table API streaming job is stuck when enabling mini batching
> {code}
>     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>     EnvironmentSettings settings =
>         EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
>     StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
>     // disable mini-batching completely to get a result
>     Configuration tableConf = tableEnv.getConfig()
>         .getConfiguration();
>     tableConf.setString("table.exec.mini-batch.enabled", "true");
>     tableConf.setString("table.exec.mini-batch.allow-latency", "5 s");
>     tableConf.setString("table.exec.mini-batch.size", "5000");
>     tableConf.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");
>     tableEnv.executeSql(
>         "CREATE TABLE input_table ("
>             + "location STRING, "
>             + "population INT"
>             + ") WITH ("
>             + "'connector' = 'kafka', "
>             + "'topic' = 'kafka_batching_input', "
>             + "'properties.bootstrap.servers' = 'localhost:9092', "
>             + "'format' = 'csv', "
>             + "'scan.startup.mode' = 'earliest-offset'"
>             + ")");
>     tableEnv.executeSql(
>         "CREATE TABLE result_table WITH ('connector' = 'print') LIKE input_table (EXCLUDING OPTIONS)");
>     tableEnv
>         .from("input_table")
>         .groupBy($("location"))
>         .select($("location").cast(DataTypes.CHAR(2)).as("location"), $("population").sum().as("population"))
>         .executeInsert("result_table");
> {code}
> I am using a pre-populated Kafka topic called {{kafka_batching_input}} with these elements:
> {code}
> "Berlin",1
> "Berlin",2
> {code}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)