@Override
    public long resetOffset(final MessageStore store, final String group,
            final String topic, int partitionId,
            long reSetOffset, final String modifier) {
        long oldOffset = -1;
        if (store != null) {
            long indexMaxOffset = store.getIndexMaxOffset();
            reSetOffset = MixedUtils.mid(reSetOffset, store.getIndexMinOffset(), indexMaxOffset);
            String offsetCacheKey = OffsetStgInfo.buildOffsetKey(topic, partitionId);
            getAndResetTmpOffset(group, offsetCacheKey);
            OffsetStorageInfo regInfo =
                    loadOrCreateOffset(group, topic, partitionId, offsetCacheKey, 0);
            oldOffset = regInfo.getAndSetOffset(reSetOffset);
            long currentOffset = regInfo.getOffset();
            long offsetDelta = indexMaxOffset - currentOffset;
            logger.info(new StringBuilder(512)
                    .append("[Offset Manager] Manual update offset by modifier=")
                    .append(modifier).append(",resetOffset=").append(reSetOffset)
                    .append(",loadedOffset=").append(oldOffset)
                    .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());
        }
        return oldOffset;
    }
Reset offset.

@param store          the message store instance
@param group          the consume group name
@param topic          the consumed topic name
@param partitionId    the consumed partition id
@param reSetOffset    the reset offset
@param modifier       the modifier
@return               the current offset
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/DefaultOffsetManager.java