@Override
        public void run() {
            while (!isClose) {
                try {
                    TaskSnapshotRequest request = snapshotQueue.poll(1, TimeUnit.SECONDS);
                    if (request == null || CollectionUtils.isEmpty(request.getSnapshotList())) {
                        continue;
                    }

                    List<TaskSnapshotMessage> requestList = request.getSnapshotList();
                    for (TaskSnapshotMessage message : requestList) {
                        Integer id = message.getJobId();
                        StreamSourceEntity entity = new StreamSourceEntity();
                        entity.setId(id);
                        entity.setSnapshot(message.getSnapshot());
                        entity.setReportTime(request.getReportTime());

                        // update snapshot
                        sourceMapper.updateSnapshot(entity);
                    }
                } catch (Throwable t) {
                    LOGGER.error("source snapshot task runnable error", t);
                }
            }
        }
The task of saving source task snapshot into DB.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/SourceSnapshotOperator.java