@Override
    public final synchronized void writeRecord(In record) {

        updateMetric(record);

        if (!isValidRowKind(record)) {
            return;
        }

        checkFlushException();

        try {
            addToBatch(record, jdbcRecordExtractor.apply(record));
            batchCount++;
            if (executionOptions.getBatchSize() > 0
                    && batchCount >= executionOptions.getBatchSize()) {
                flush();
                if (sinkMetricData != null && dirtySink == null) {
                    sinkMetricData.invoke(rowSize, dataSize);
                }
                resetStateAfterFlush();
            }
        } catch (Exception e) {
            LOG.error(String.format("jdbc batch write record error, raw data: %s", record), e);
            handleDirtyData(record, DirtyType.EXTRACT_ROWDATA_ERROR, e);
            resetStateAfterFlush();
        }
    }
update before is used for metric computing only
@param record
@return
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcBatchingOutputFormat.java