private void flushLegacyData() {
        if (legacyData == null || legacyData.isEmpty()) {
            return;
        }

        for (StarRocksSinkBufferEntity entity : legacyData) {
            for (byte[] data : entity.getBuffer()) {
                sinkManager.write(null, entity.getDatabase(), entity.getTable(),
                        new String(data, StandardCharsets.UTF_8));
            }
            log.info("Write {} legacy records from table '{}' of database '{}'",
                    entity.getBuffer().size(), entity.getDatabase(), entity.getTable());
        }
        legacyData.clear();
    }
The map to store the start time of each checkpoint.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/table/sink/table/StarRocksDynamicSinkFunctionV2.java