I need help to understand the mechanism process function better and debug my code: Here is the implementation of acquiring DataStream and I will need to pass the results to a process function for further data analysis.

static void processWorkflow(
        StreamTableEnvironment tableEnv,
        DataStream<DataFileChunk> rawDataStream,
        DataStream<DataFileChunk> bkgdDataStream,
        String jsonCalibrationData
        ) {
    tableEnv.createTemporaryView("EMPAD_RAW_TBL", rawDataStream);
    tableEnv.createTemporaryView("EMPAD_BKGD_TBL", bkgdDataStream);

    String data_query = "select EMPAD_RAW_TBL.chunk_i as chunk_i, EMPAD_RAW_TBL.data as raw_enc_data, EMPAD_BKGD_TBL.data as bkgd_enc_data " +
            "FROM EMPAD_RAW_TBL join EMPAD_BKGD_TBL on EMPAD_RAW_TBL.chunk_i = EMPAD_BKGD_TBL.chunk_i";

    Table resultTable =
            tableEnv.sqlQuery(data_query);

    DataStream<Row> resultStream = tableEnv.toChangelogStream(resultTable);

and this is the implementation of my ProcessFunction:

package org.varimat.process;

import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

public class EMPADProcessor extends ProcessFunction<Row, Row> {

    @Override
    public void processElement(Row row, ProcessFunction<Row, Row>.Context context,
                               Collector<Row> collector) throws Exception {

        // process row here...
        System.out.println(row.getField(0));
        String rawObject = String.valueOf(row.getField(1));
        String bkgdObject = String.valueOf(row.getField(2));
    }
}

When I run my application, it never stops and prints duplicated data of

row.getField(0)

in the

processElement

method. Is there any suggestion? In other words, do I need to implement class

EMPADProcessor

different?