I am processing messages from DataStream and pass them to the Process Function:

Table raw_table = tableEnv.sqlQuery(data_query);

DataStream<Row> join_stream = tableEnv.toDataStream(raw_table);

SingleOutputStreamOperator<List<double[][][]>> outputStreamOperator = join_stream.process(new EMPADProcessor(commands, taskENT.getTaskId()));

where

public void processElement(Row row, ProcessFunction<Row, List<double[][][]>>.Context context, Collector<List<double[][][]>> collector) throws Exception {

inside

processElement

I am doing some data analysis which takes a few moments for every single message (row) and it this short processing time, causes duplicated messages and I don’t know why?