private void cfmOffset(StringBuilder sb, String group,
            Collection<OffsetStorageInfo> infoList) throws OffsetStoreException {
        sb.delete(0, sb.length());
        for (final OffsetStorageInfo info : infoList) {
            long newOffset = -1;
            long msgId = -1;
            synchronized (info) {
                if (!info.isModified()) {
                    continue;
                }
                newOffset = info.getOffset();
                msgId = info.getMessageId();
                info.setModified(false);
            }
            final String topic = info.getTopic();
            String offsetPath = sb.append(this.consumerZkDir).append("/")
                    .append(group).append("/offsets/").append(topic).append("/")
                    .append(info.getBrokerId()).append(TokenConstants.HYPHEN)
                    .append(info.getPartitionId()).toString();
            sb.delete(0, sb.length());
            String offsetData =
                    sb.append(msgId).append(TokenConstants.HYPHEN).append(newOffset).toString();
            sb.delete(0, sb.length());
            try {
                ZKUtil.updatePersistentPath(this.zkw, offsetPath, offsetData);
            } catch (final Throwable t) {
                BrokerSrvStatsHolder.incZKExcCnt();
                logger.error("Exception during commit offsets to ZooKeeper", t);
                throw new OffsetStoreException(t);
            }
            if (logger.isDebugEnabled()) {
                logger.debug(sb.append("Committed offset, path=")
                        .append(offsetPath).append(", data=").append(offsetData).toString());
                sb.delete(0, sb.length());
            }
        }
    }
Initial ZooKeeper offset storage object

@param zkConfig   the ZooKeeper configure
@param isBroker   whether used in broker node
@param brokerId   the broker id
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/offset/offsetstorage/ZkOffsetStorage.java