@Override
    public OffsetStorageInfo loadOffset(final MessageStore msgStore, final String group,
            final String topic, int partitionId, int readStatus,
            long reqOffset, final StringBuilder sBuilder) {
        OffsetStorageInfo regInfo;
        long indexMaxOffset = msgStore.getIndexMaxOffset();
        long indexMinOffset = msgStore.getIndexMinOffset();
        long defOffset =
                (readStatus == TBaseConstants.CONSUME_MODEL_READ_NORMAL)
                        ? indexMinOffset
                        : indexMaxOffset;
        String offsetCacheKey = OffsetStgInfo.buildOffsetKey(topic, partitionId);
        regInfo = loadOrCreateOffset(group, topic, partitionId, offsetCacheKey, defOffset);
        getAndResetTmpOffset(group, offsetCacheKey);
        final long curOffset = regInfo.getOffset();
        final boolean isFirstCreate = regInfo.isFirstCreate();
        if ((reqOffset >= 0)
                || (readStatus == TBaseConstants.CONSUME_MODEL_READ_FROM_MAX_ALWAYS)) {
            long adjOffset = indexMaxOffset;
            if (readStatus != TBaseConstants.CONSUME_MODEL_READ_FROM_MAX_ALWAYS) {
                adjOffset = MixedUtils.mid(reqOffset, indexMinOffset, indexMaxOffset);
            }
            regInfo.getAndSetOffset(adjOffset);
        }
        sBuilder.append("[Offset Manager]");
        switch (readStatus) {
            case TBaseConstants.CONSUME_MODEL_READ_FROM_MAX_ALWAYS:
                sBuilder.append(" Consume From Max Offset Always");
                break;
            case TBaseConstants.CONSUME_MODEL_READ_FROM_MAX:
                sBuilder.append(" Consume From Max Offset");
                break;
            default: {
                sBuilder.append(" Normal Offset");
            }
        }
        if (!isFirstCreate) {
            sBuilder.append(",Continue");
        }
        long currentOffset = regInfo.getOffset();
        long offsetDelta = indexMaxOffset - currentOffset;
        logger.info(sBuilder.append(",requestOffset=").append(reqOffset)
                .append(",loadedOffset=").append(curOffset)
                .append(",currentOffset=").append(currentOffset)
                .append(",maxOffset=").append(indexMaxOffset)
                .append(",offsetDelta=").append(offsetDelta)
                .append(",lagLevel=").append(getLagLevel(offsetDelta))
                .append(",group=").append(group)
                .append(",topic-part=").append(offsetCacheKey).toString());
        sBuilder.delete(0, sBuilder.length());
        return regInfo;
    }
Load offset.

@param msgStore       the message store instance
@param group          the consume group name
@param topic          the consumed topic name
@param partitionId    the consumed partition id
@param readStatus     the consume behavior of the consumer group
@param reqOffset      the bootstrap offset
@param sBuilder       the string buffer
@return               the loaded offset information
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/DefaultOffsetManager.java