@Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        LOG.info("redis start snapshotState, id: {}", functionSnapshotContext.getCheckpointId());
        synchronized (lock) {
            listState.clear();
            listState.addAll(rows);
        }
        if (sinkMetricData != null && metricStateListState != null) {
            MetricStateUtils.snapshotMetricStateForSinkMetricData(metricStateListState, sinkMetricData,
                    getRuntimeContext().getIndexOfThisSubtask());
        }
        LOG.info("redis end snapshotState, id: {}", functionSnapshotContext.getCheckpointId());
    }
The stop watch to measure time duration.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/sink/AbstractRedisSinkFunction.java