@Override
    public Map<String, OffsetHistoryInfo> getOnlineGroupOffsetInfo() {
        Long tmpOffset;
        OffsetHistoryInfo recordInfo;
        ConcurrentHashMap<String, Long> fetchOffsetMap;
        Map<String, OffsetHistoryInfo> result = new HashMap<>();
        for (Map.Entry<String, ConcurrentHashMap<String, OffsetStorageInfo>> entry : cfmOffsetMap.entrySet()) {
            if (entry == null || entry.getKey() == null || entry.getValue() == null) {
                continue;
            }
            // get fetch offset map
            fetchOffsetMap = tmpOffsetMap.get(entry.getKey());
            // read confirm offset map information
            for (Map.Entry<String, OffsetStorageInfo> partEntry : entry.getValue().entrySet()) {
                if (partEntry == null
                        || partEntry.getKey() == null
                        || partEntry.getValue() == null) {
                    continue;
                }
                recordInfo = result.get(entry.getKey());
                if (recordInfo == null) {
                    recordInfo = new OffsetHistoryInfo(
                            brokerConfig.getBrokerId(), entry.getKey());
                    result.put(entry.getKey(), recordInfo);
                }
                tmpOffset = 0L;
                if (fetchOffsetMap != null) {
                    tmpOffset = fetchOffsetMap.get(partEntry.getKey());
                    if (tmpOffset == null) {
                        tmpOffset = 0L;
                    }
                }
                recordInfo.addCsmOffsets(partEntry.getValue().getTopic(),
                        partEntry.getValue().getPartitionId(),
                        partEntry.getValue().getOffset(), tmpOffset);
            }
        }
        return result;
    }
Get online groups' offset information

@return group offset info in memory
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/DefaultOffsetManager.java