[FLINK-24626] Flink JDBC Sink may lose data in left join

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Abandoned
• Component/s: Connectors / JDBC
• Labels: auto-deprioritized-major correctness
• Assignee: Unassigned
• Reporter: Kenyore
• Created: 25/Oct/21 02:03
• Updated: 24/Feb/22 09:21
• Resolved: 24/Feb/22 09:21

Description

The JDBC sink will lose some data while using TableBufferReducedStatementExecutor and left join.

Here are a snippet of executeBatch.

    @Override
    public void executeBatch() throws SQLException {
        for (Map.Entry<RowData, Tuple2<Boolean, RowData>> entry : reduceBuffer.entrySet()) {
            if (entry.getValue().f0) {
                upsertExecutor.addToBatch(entry.getValue().f1);
            } else {
                // delete by key
                deleteExecutor.addToBatch(entry.getKey());
            }
        }
        upsertExecutor.executeBatch();
        deleteExecutor.executeBatch();
        reduceBuffer.clear();
    }

Left join will generate a DETETE row before upsert row and the executeBatch will excute them in a wrong order.Whitch may causes data lose.