public List<UserTag> findByPage(Long lastId, int pageSize) {
    String sql = "SELECT ... WHERE user_id > ? ORDER BY user_id LIMIT ?";
    return jdbcTemplate.query(sql, (rs, rowNum) -> mapRowToUserTag(rs), lastId, pageSize);
}

OutputStream rawStream = storage.getOutputStream(tempFilename);
GZIPOutputStream gos = new GZIPOutputStream(rawStream);
CodedOutputStream cos = CodedOutputStream.newInstance(gos);

// 写入每条消息（delimited）
for (UserTag record : batch) {
    UserTagProto proto = convertToProto(record);
    proto.writeDelimitedTo(cos); // 自动添加长度前缀
}
cos.flush();
gos.close(); // 触发压缩结束

// 写入 .tmp 临时文件
String tempFile = filename + ".tmp";
try (OutputStream os = storage.getOutputStream(tempFile)) {
    writeData(os);
}

// 原子性发布：rename to final name
storage.publish(tempFile, filename); 

public void cleanupOld(String dataSource, long currentVersion) {
    try {
        String prefix = dataSource + "_v";

        for (String file : storage.listFiles(prefix)) {
            Long ver = extractVersion(file);
            if (ver != null && ver < currentVersion) {
                storage.delete(file);
                log.info("🗑️ 删除旧快照文件: {}", file);
            }
        }

        // 清理旧 manifest 和 checkpoint
        cleanupByPrefix("manifest/" + prefix, currentVersion);
        cleanupByPrefix("checkpoints/" + prefix, currentVersion);

    } catch (IOException e) {
        log.error("清理失败", e);
    }
}

private void cleanupByPrefix(String prefix, long currentVersion) throws IOException {
    for (String file : storage.listFiles(prefix)) {
        Long ver = extractVersion(file);
        if (ver != null && ver < currentVersion) {
            storage.delete(file);
            log.info("🗑️ 删除旧辅助文件: {}", file);
        }
    }
}

// 简化版实现
@KafkaListener(topics = "#{'${cache.topics}'.split(',')}")
public void handleCacheUpdate(ConsumerRecord<String, CacheUpdateMessage> record) {
    try {
        applyCacheUpdate(record.value());
        // 异步提交偏移量
        kafkaTemplate.commitSync();
    } catch (Exception e) {
        // 处理失败，记录异常但不提交偏移量
        log.error("Failed to process cache update", e);
    }
}

// 服务启动时全量加载 + 增量追赶
@PostConstruct
public void init() {
    // 1. 从OSS加载全量
    loadFullDataFromOSS();
    
    // 2. 寻找合适的Kafka偏移量开始位置
    long startOffset = calculateStartOffset();
    
    // 3. 从指定位置开始消费
    kafkaConsumer.seekToBeginning(Collections.singletonList(partition));
    kafkaConsumer.seek(partition, startOffset);
}

package com.healerjean.proj.hotcache.incr;


import com.healerjean.proj.hotcache.config.DatasetIncrementalConfig;
import com.healerjean.proj.hotcache.config.IncrementalExecutionConfig;
import com.healerjean.proj.hotcache.config.SnapshotGlobalConfig;
import com.healerjean.proj.hotcache.enums.SnapshotPathEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;

/**
 * 消息队列消费者 - 处理增量数据写入
 *
 * @author zhangyujin
 * @date 2025/11/3
 */
@Slf4j
@Component
public class IncrementalWriteService {


    /***
     * snapshotGlobalConfig
     */
    @Resource
    private SnapshotGlobalConfig snapshotGlobalConfig;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 增量任务写入
     *
     * @param datasetName datasetName
     */
    public void write(String data, String datasetName) {
        IncrementalExecutionConfig executionConfig = snapshotGlobalConfig.instanceIncrementalConfig(datasetName);
        DatasetIncrementalConfig incrementalConfig = executionConfig.getIncrementalConfig();
        Long ttlSeconds = incrementalConfig.getTtlSeconds();
        LocalDateTime dateTime = LocalDateTime.now();
        Duration ttl = Duration.ofSeconds(ttlSeconds);
        // 1. 获取全局唯一 offset


        String offsetKey =  SnapshotPathEnum.REDIS_INCR_LATEST_OFFSET_KEY.format(datasetName);
        Long offset = redisTemplate.opsForValue().increment(offsetKey);
        // 2. 如果 Redis 返回 null（如 key 不存在且 incr 失败），从 DB 恢复
        if (offset == null || offset <= 0) {
            offset = restoreOffsetFromDb(datasetName, offsetKey);
        }

        // 2. 根据 offset 计算分片 ID（每 MAX_MEMBERS_PER_ZSET 一个分片）
        Integer maxMembersSize = incrementalConfig.getMaxMembersSize();
        long shardId = (offset - 1) / maxMembersSize;
        String zsetKey =  SnapshotPathEnum.REDIS_INCR_CURRENT_SHARD_KEY.format(datasetName, shardId);

        // 3. 写入 ZSET（score = offset）
        Boolean added = redisTemplate.opsForZSet().add(zsetKey, data, (double) offset);
        if (Boolean.TRUE.equals(added)) {
            redisTemplate.expire(zsetKey, ttl);
        }

        // 4. 写入时间索引
        long incrTimeSecond = executionConfig.getIncrIntervalTimeSecond(dateTime, 1);
        String timePeriodKey =  SnapshotPathEnum.REDIS_INCR_TIME_PERIOD_KEY.format(datasetName);
        String timeOffset =  SnapshotPathEnum.REDIS_INCR_TIME_OFFSET_KEY.format(datasetName, String.valueOf(incrTimeSecond));
        redisTemplate.opsForValue().set(timeOffset, offset.toString(), ttl);
        redisTemplate.opsForZSet().add(timePeriodKey, String.valueOf(incrTimeSecond), (double) incrTimeSecond);
        Long currentSize = redisTemplate.opsForZSet().size(timePeriodKey);
        if (currentSize != null && currentSize > maxMembersSize) {
            long removeCount = currentSize - maxMembersSize;
            redisTemplate.opsForZSet().removeRange(timePeriodKey, 0, removeCount - 1);
        }
        redisTemplate.expire(timePeriodKey, ttl);

        Integer dbSaveOffsetInterval = incrementalConfig.getDbSaveOffsetInterval();
        if (offset % dbSaveOffsetInterval == 0) {
            // 6. 将 offset 写入数据库
        }

    }

    /**
     * 从数据库恢复
     *
     * @param datasetName datasetName
     * @param offsetKey   offsetKey
     * @return {@link Long}
     */
    private long restoreOffsetFromDb(String datasetName, String offsetKey) {
        String lockKey = "lock:offset:init:" + datasetName;
        Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", Duration.ofSeconds(10));
        if (Boolean.FALSE.equals(locked)) {
            throw new RuntimeException("Interrupted while waiting for offset init");
        }
        try {
            // 从 DB 读取最新 offset
            // Long dbOffset = offsetRepository.getCurrentOffsetFromDb(datasetName);
            // long nextOffset = dbOffset + 1;
            // redisTemplate.opsForValue().set(offsetKey, String.valueOf(nextOffset));
            // return nextOffset;
        } finally {
            redisTemplate.delete(lockKey);
        }
        return 1;
    }

}

long shardId = (offset - 1) / maxMembersSize;
String zsetKey = REDIS_INCR_CURRENT_SHARD_KEY.format(datasetName, shardId);

redisTemplate.opsForZSet().add(zsetKey, data, (double) offset);
redisTemplate.expire(zsetKey, ttlSeconds); // 如 86400s = 24h

long incrTimeSecond = executionConfig.getIncrIntervalTimeSecond(dateTime, 1); // 按秒对齐
String timeOffsetKey = "user_tag:incr:timeOffset:" + incrTimeSecond;
redisTemplate.opsForValue().set(timeOffsetKey, String.valueOf(offset), ttl);

package com.healerjean.proj.hotcache.incr;


import com.healerjean.proj.hotcache.config.DatasetIncrementalConfig;
import com.healerjean.proj.hotcache.config.IncrementalExecutionConfig;
import com.healerjean.proj.hotcache.config.SnapshotGlobalConfig;
import com.healerjean.proj.hotcache.enums.SnapshotPathEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;

/**
 * 消息队列消费者 - 处理增量数据写入
 *
 * @author zhangyujin
 * @date 2025/11/3
 */
@Slf4j
@Component
public class IncrementalWriteService {


    /***
     * snapshotGlobalConfig
     */
    @Resource
    private SnapshotGlobalConfig snapshotGlobalConfig;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 增量任务写入
     *
     * @param datasetName datasetName
     */
    public void write(String data, String datasetName) {
        IncrementalExecutionConfig executionConfig = snapshotGlobalConfig.instanceIncrementalConfig(datasetName);
        DatasetIncrementalConfig incrementalConfig = executionConfig.getIncrementalConfig();
        Long ttlSeconds = incrementalConfig.getTtlSeconds();
        LocalDateTime dateTime = LocalDateTime.now();
        Duration ttl = Duration.ofSeconds(ttlSeconds);

        // 1. 获取全局唯一 offset
        String offsetKey = SnapshotPathEnum.REDIS_INCR_LATEST_OFFSET_KEY.format(datasetName);
        Long offset = redisTemplate.opsForValue().increment(offsetKey);
        // 2. 如果 Redis 返回 null（如 key 不存在且 incr 失败），从 DB 恢复
        if (offset == null || offset <= 0) {
            offset = restoreOffsetFromDb(datasetName, offsetKey);
        }else {
            // 6. todo 将 offset 写入数据库
        }

        // 2. 根据 offset 计算分片 ID（每 MAX_MEMBERS_PER_ZSET 一个分片）
        Integer maxMembersSize = incrementalConfig.getMaxMembersSize();
        long shardId = (offset - 1) / maxMembersSize;
        String zsetKey = SnapshotPathEnum.REDIS_INCR_CURRENT_SHARD_KEY.format(datasetName, shardId);

        // 3. 写入 ZSET（score = offset）
        Boolean added = redisTemplate.opsForZSet().add(zsetKey, data, (double) offset);
        if (Boolean.TRUE.equals(added)) {
            redisTemplate.expire(zsetKey, ttl);
        }

        // 4. 写入时间索引
        long incrTimeSecond = executionConfig.getIncrIntervalTimeSecond(dateTime, 1);
        String timePeriodKey = SnapshotPathEnum.REDIS_INCR_TIME_PERIOD_KEY.format(datasetName);
        String timeOffset = SnapshotPathEnum.REDIS_INCR_TIME_OFFSET_KEY.format(datasetName, String.valueOf(incrTimeSecond));
        redisTemplate.opsForValue().set(timeOffset, offset.toString(), ttl);
        redisTemplate.opsForZSet().add(timePeriodKey, String.valueOf(incrTimeSecond), (double) incrTimeSecond);
        Long currentSize = redisTemplate.opsForZSet().size(timePeriodKey);
        if (currentSize != null && currentSize > maxMembersSize) {
            long removeCount = currentSize - maxMembersSize;
            redisTemplate.opsForZSet().removeRange(timePeriodKey, 0, removeCount - 1);
        }
        redisTemplate.expire(timePeriodKey, ttl);

        // 6. todo 将 offset 写入数据库

    }

    /**
     * 从数据库恢复
     *
     * @param datasetName datasetName
     * @param offsetKey   offsetKey
     * @return {@link Long}
     */
    private long restoreOffsetFromDb(String datasetName, String offsetKey) {
        String lockKey = "lock:offset:init:" + datasetName;
        Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", Duration.ofSeconds(10));
        if (Boolean.FALSE.equals(locked)) {
            throw new RuntimeException("Interrupted while waiting for offset init");
        }
        try {
            // 从 DB 读取最新 offset
            // Long dbOffset = offsetRepository.getCurrentOffsetFromDb(datasetName);
            // long nextOffset = dbOffset + 1;
            // redisTemplate.opsForValue().set(offsetKey, String.valueOf(nextOffset));
            // return nextOffset;
        } finally {
            redisTemplate.delete(lockKey);
        }
        return 1;
    }


}

package com.healerjean.proj.hotcache.incr;


import com.healerjean.proj.hotcache.config.DatasetIncrementalConfig;
import com.healerjean.proj.hotcache.config.IncrementalExecutionConfig;
import com.healerjean.proj.hotcache.config.SnapshotGlobalConfig;
import com.healerjean.proj.hotcache.enums.SnapshotPathEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;

/**
 * 消息队列消费者 - 处理增量数据写入
 *
 * @author zhangyujin
 * @date 2025/11/3
 */
@Slf4j
@Component
public class IncrementalLuaWriteService {


    private final static RedisScript<String> WRITE_INCREMENTAL_SCRIPT = RedisScript.of(
            "redis.call('ZADD', KEYS[1], ARGV[2], ARGV[1]);" +
                    "redis.call('EXPIRE', KEYS[1], ARGV[3]);" +
                    "redis.call('SET', KEYS[2], ARGV[2], 'EX', ARGV[3]);" +
                    "redis.call('ZADD', KEYS[3], ARGV[4], ARGV[4]);" +
                    "redis.call('EXPIRE', KEYS[3], ARGV[3]);" +
                    "local maxMembers = tonumber(ARGV[5]);" +
                    "local currentSize = redis.call('ZCARD', KEYS[3]);" +
                    "if currentSize > maxMembers then " +
                    "  local removeCount = currentSize - maxMembers; " +
                    "  redis.call('ZREMRANGEBYRANK', KEYS[3], 0, removeCount - 1); " +
                    "end; " +
                    "return 1;",
            Long.class
    );

    /***
     * snapshotGlobalConfig
     */
    @Resource
    private SnapshotGlobalConfig snapshotGlobalConfig;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 增量任务写入
     *
     * @param datasetName datasetName
     */
    public void write(String data, String datasetName) {
        IncrementalExecutionConfig executionConfig = snapshotGlobalConfig.instanceIncrementalConfig(datasetName);
        DatasetIncrementalConfig incrementalConfig = executionConfig.getIncrementalConfig();
        Long ttlSeconds = incrementalConfig.getTtlSeconds();
        Integer maxMembersSize = incrementalConfig.getMaxMembersSize();
        LocalDateTime now = LocalDateTime.now();

        // 1. 获取全局唯一 offset
        String offsetKey = SnapshotPathEnum.REDIS_INCR_LATEST_OFFSET_KEY.format(datasetName);
        Long offset = redisTemplate.opsForValue().increment(offsetKey);
        if (offset == null || offset <= 0) {
            offset = restoreOffsetFromDb(datasetName, offsetKey);
            // 注意：restore 后建议 SET offsetKey offset+1 避免重复，此处简化
        }else {
            // 6. todo 将 offset 写入数据库
        }

        // 2. 计算 shardId 和所有 keys
        long shardId = (offset - 1) / maxMembersSize;
        String shardKey = SnapshotPathEnum.REDIS_INCR_CURRENT_SHARD_KEY.format(datasetName, shardId);
        long incrTimeSecond = executionConfig.getIncrIntervalTimeSecond(now, 1);
        String timeOffsetKey = SnapshotPathEnum.REDIS_INCR_TIME_OFFSET_KEY.format(datasetName, String.valueOf(incrTimeSecond));
        String timePeriodKey = SnapshotPathEnum.REDIS_INCR_TIME_PERIOD_KEY.format(datasetName);

        // 3. 准备 Lua 参数
        List<String> keys = Arrays.asList(shardKey, timeOffsetKey, timePeriodKey);
        List<String> args = Arrays.asList(
                data,
                String.valueOf(offset),
                String.valueOf(ttlSeconds),
                String.valueOf(incrTimeSecond),
                String.valueOf(maxMembersSize)
        );

        // 4. 执行 Lua 脚本
        redisTemplate.execute(WRITE_INCREMENTAL_SCRIPT, keys, args);
    }

    /**
     * 从数据库恢复
     *
     * @param datasetName datasetName
     * @param offsetKey   offsetKey
     * @return {@link Long}
     */
    private long restoreOffsetFromDb(String datasetName, String offsetKey) {
        String lockKey = "lock:offset:init:" + datasetName;
        Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", Duration.ofSeconds(10));
        if (Boolean.FALSE.equals(locked)) {
            throw new RuntimeException("Interrupted while waiting for offset init");
        }
        try {
            // 从 DB 读取最新 offset
            // Long dbOffset = offsetRepository.getCurrentOffsetFromDb(datasetName);
            // long nextOffset = dbOffset + 1;
            // redisTemplate.opsForValue().set(offsetKey, String.valueOf(nextOffset));
            // return nextOffset;
        } finally {
            redisTemplate.delete(lockKey);
        }
        return 1;
    }


}

while (true) {
    Long latest = redis.get("latest_snapshot_version");
    if (latest > currentVersion) {
        downloadFromCDN(latest);
        break;
    }
    Thread.sleep(5000); // 每 5 秒查一次
}

package com.healerjean.proj.hotcache.service.pull;

import com.alibaba.fastjson.JSON;
import com.google.gson.Gson;
import com.healerjean.proj.hotcache.config.DatasetSnapshotConfig;
import com.healerjean.proj.hotcache.config.SnapshotGlobalConfig;
import com.healerjean.proj.hotcache.enums.SnapshotPathEnum;
import com.healerjean.proj.hotcache.factory.SnapshotFactory;
import com.healerjean.proj.hotcache.model.SnapshotMetadata;
import com.healerjean.proj.hotcache.model.UserTag;
import com.healerjean.proj.hotcache.service.cache.InMemoryUserTagCache;
import com.healerjean.proj.hotcache.service.serialization.DataSerializerStrategy;
import com.healerjean.proj.hotcache.service.storage.StorageServiceStrategy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;

/**
 * FullLoadToMemoryService
 *
 * @author zhangyujin
 * @date 2025/11/6
 */
@Service
@Slf4j
public class FullLoadToMemoryService {

    @Autowired
    private SnapshotGlobalConfig snapshotGlobalConfig;

    @Autowired
    private InMemoryUserTagCache userTagCache;

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    public void loadLatestFullSnapshotToMemory(String datasetName) throws Exception {
        // 1. 从 Redis 获取最新版本
        String latestKey = SnapshotPathEnum.REDIS_SNAPSHOT_LATEST_VERSION_KEY.format(datasetName);
        String latestVersion = redisTemplate.opsForValue().get(latestKey);
        if (latestVersion == null) {
            throw new IllegalStateException("No full snapshot found in Redis");
        }

        // 2. 获取元数据（含 manifestUrl）
        String metaKey = SnapshotPathEnum.REDIS_SNAPSHOT_VERSION_META_KEY.format(datasetName, latestVersion);
        String metaJson = redisTemplate.opsForValue().get(metaKey);
        SnapshotMetadata metadata = new Gson().fromJson(metaJson, SnapshotMetadata.class);

        // 3. 根据 storageStrategy 下载所有分片文件并反序列化
        DatasetSnapshotConfig snapshotConfig = metadata.getConfig();
        StorageServiceStrategy storage = SnapshotFactory.getInstance().getStorageServiceStrategy(datasetName, snapshotConfig.getStorageStrategy());
        DataSerializerStrategy dataSerializer = SnapshotFactory.getInstance().getDataSerializer(datasetName, snapshotConfig.getSerializerStrategy());
        for (SnapshotMetadata.FileInfo fileInfo : metadata.getFileInfoMap().values()) {
            try (InputStream is = storage.download(fileInfo.getFilename())) {
                BufferedReader reader = new BufferedReader(new InputStreamReader(is));
                String line;
                while ((line = reader.readLine()) != null) {
                    if (!line.trim().isEmpty()) {
                        try {
                            UserTag userTag = (UserTag)dataSerializer.deserialize(line.getBytes(StandardCharsets.UTF_8));
                            userTagCache.put(userTag.getUserId(), userTag);
                        }catch (Exception e){
                            log.error("line:{}, fileInfo:{}", line, JSON.toJSONString(fileInfo), e);
                            throw new RuntimeException(e.getMessage(), e);
                        }
                    }
                }
            }
        }
        log.info("✅ 全量加载完成，共 {} 条记录", userTagCache.getAll().size());
    }
}

private final static Map<String, AtomicLong> LOCAL_PROCESSED_OFFSET_MAP = new ConcurrentHashMap<>();

/**
 * 时间驱动的起始点定位
 */
public void pullIncrementalByLowerBound(String datasetName) {
    // 1、从 Redis 获取最新快照版本（latestVersion）
    String latestKey = SnapshotPathEnum.REDIS_SNAPSHOT_LATEST_VERSION_KEY.format(datasetName);
    String latestVersion = redisTemplate.opsForValue().get(latestKey);
    if (latestVersion == null) {
        throw new IllegalStateException("No full snapshot found in Redis");
    }

    // 2. 解析其元数据 SnapshotMetadata，获取 incrPullTimeLowerBound（时间戳）
    String metaKey = SnapshotPathEnum.REDIS_SNAPSHOT_VERSION_META_KEY.format(datasetName, latestVersion);
    String metaJson = redisTemplate.opsForValue().get(metaKey);
    SnapshotMetadata metadata = new Gson().fromJson(metaJson, SnapshotMetadata.class);
    long incrPullTimeLowerBound = metadata.getIncrPullTimeLowerBound();
    String timePeriodKey = SnapshotPathEnum.REDIS_INCR_TIME_PERIOD_KEY.format(datasetName);

    // 3、 查找 >= 时间下界的第一个时间点
    Set<String> candidates = redisTemplate.opsForZSet().rangeByScore(timePeriodKey, (double) incrPullTimeLowerBound, Double.MAX_VALUE, 0, 1);
    if (CollectionUtils.isEmpty(candidates)) {
        LOCAL_PROCESSED_OFFSET_MAP.put(datasetName, new AtomicLong(0));
        log.info("No time period found >= {} for dataset {}", incrPullTimeLowerBound, datasetName);
        return;
    }

    // 4、通过 REDIS_INCR_TIME_OFFSET_KEY 反查对应的 offset
    String firstTimeStr = candidates.iterator().next();
    String offsetKey = SnapshotPathEnum.REDIS_INCR_TIME_OFFSET_KEY.format(datasetName, firstTimeStr);
    String offsetValue = redisTemplate.opsForValue().get(offsetKey);
    long latestOffset = Long.parseLong(offsetValue);

    // 5、初始化 LOCAL_PROCESSED_OFFSET_MAP 并启动 pullIncrementalByOffset
    LOCAL_PROCESSED_OFFSET_MAP.put(datasetName, new AtomicLong(latestOffset));
    pullIncrementalByOffset(datasetName);
}

@Autowired
private SnapshotGlobalConfig snapshotGlobalConfig;

@Autowired
private InMemoryUserTagCache userTagCache;

@Resource
private RedisTemplate<String, String> redisTemplate;

private static final long PULL_BATCH_SIZE = 300L;

/**
 * 本地已处理的偏移量（必须先通过 pullIncrementalByLowerBound 初始化）
 */
private final static Map<String, AtomicLong> LOCAL_PROCESSED_OFFSET_MAP = new ConcurrentHashMap<>();


/**
 * 增量拉取流程
 *
 * @param datasetName datasetName
 */
public void pullIncrementalByOffset(String datasetName) {
    if (!LOCAL_PROCESSED_OFFSET_MAP.containsKey(datasetName)) {
        log.warn("Offset not initialized for dataset: {}. Skipping pull. Call pullIncrementalByLowerBound first.", datasetName);
        return;
    }
    IncrementalExecutionConfig executionConfig = snapshotGlobalConfig.instanceIncrementalConfig(datasetName);
    DatasetIncrementalConfig incrementalConfig = executionConfig.getIncrementalConfig();
    Integer maxMembersSize = incrementalConfig.getMaxMembersSize();
    AtomicLong atomicCurrentOffset = LOCAL_PROCESSED_OFFSET_MAP.get(datasetName);

    while (true) {
        // 1、计算当前应读取的 shardId，
        long currentOffset = atomicCurrentOffset.get();
        String latestKey = SnapshotPathEnum.REDIS_INCR_LATEST_OFFSET_KEY.format(datasetName);
        String latestOffsetStr = redisTemplate.opsForValue().get(latestKey);
        long latestOffset = Long.parseLong(latestOffsetStr);
        if (currentOffset >= latestOffset) {
            // 当前 shard 已追上最新 offset → 结束
            log.debug("Dataset {} caught up to latest offset {}, sleeping...", datasetName, latestOffset);
            break;
        }

        // 2、从 Redis ZSET 拉取一批数据
        long shardId = currentOffset / maxMembersSize;
        String shardKey = SnapshotPathEnum.REDIS_INCR_CURRENT_SHARD_KEY.format(datasetName, shardId);
        long nextOffset = currentOffset + 1;
        double maxScore = Math.min(nextOffset + PULL_BATCH_SIZE - 1, latestOffset);
        Set<ZSetOperations.TypedTuple<String>> tuples = redisTemplate.opsForZSet().rangeByScoreWithScores(shardKey, (double) nextOffset, maxScore);

        // 3、防止拉取数据为空，且当前 offset 未更新，则继续拉取
        if (CollectionUtils.isEmpty(tuples)) {
            // 当前 shard 无数据，尝试跳转到下一个 shard 起始位置
            long nextShardStart = (shardId + 1) * maxMembersSize;
            long jumpTo = Math.min(nextShardStart, latestOffset);
            // 如果跳转幅度很大（比如超过一个 batch），先小步试探
            if (jumpTo - currentOffset > PULL_BATCH_SIZE * 2) {
                // 只前进一小步，避免跨度过大跳过异常数据
                jumpTo = currentOffset + 1;
            }

            if (jumpTo > currentOffset) {
                atomicCurrentOffset.set(jumpTo);
                log.debug("Advanced offset to {} for dataset {}", jumpTo, datasetName);
                continue;
            }
            // 已到最新，但仍无数据 → 可能数据延迟，稍后重试
            log.debug("No data found in shard {} for dataset {}, but not caught up yet.", shardId, datasetName);
            break;
        }

        // 4、批量处理 & 更新偏移量
        for (ZSetOperations.TypedTuple<String> data : tuples) {
            Double score = data.getScore();
            String value = data.getValue();
        }

        // 5、更新本地 offset 为本次拉取的最大 score
        // 更新 offset 为本次最大 score（注意：不再 +1）
        long maxProcessedOffset = tuples.stream()
                .mapToLong(t -> Objects.requireNonNull(t.getScore()).longValue())
                .max()
                .orElse(currentOffset);

        atomicCurrentOffset.set(maxProcessedOffset);
        log.debug("Pulled {} records for dataset {}, offset updated to {}",
                tuples.size(), datasetName, maxProcessedOffset);
    }
}