private void loadSegments(SegmentType segType, long offsetIfCreate,
            StringBuilder sBuilder) throws IOException {
        String segTypeStr = "Data";
        File segListDir = this.dataDir;
        String fileSuffix = DataStoreUtils.DATA_FILE_SUFFIX;
        if (segType == SegmentType.INDEX) {
            segTypeStr = "Index";
            segListDir = this.indexDir;
            fileSuffix = DataStoreUtils.INDEX_FILE_SUFFIX;
        }
        logger.info(sBuilder.append("[File Store] begin Load ")
                .append(segTypeStr).append(" segments ")
                .append(segListDir.getAbsolutePath()).toString());
        sBuilder.delete(0, sBuilder.length());
        final List<Segment> accum = new ArrayList<>();
        final File[] ls = segListDir.listFiles();
        if (ls != null) {
            for (final File file : ls) {
                if (file == null) {
                    continue;
                }
                if (file.isFile() && file.toString().endsWith(fileSuffix)) {
                    if (!file.canRead()) {
                        throw new IOException(new StringBuilder(512)
                                .append("Could not read ").append(segTypeStr)
                                .append(" file ").append(file).toString());
                    }
                    final String filename = file.getName();
                    final long start =
                            Long.parseLong(filename.substring(0, filename.length() - fileSuffix.length()));
                    accum.add(new FileSegment(start, file, false, segType));
                }
            }
        }
        if (accum.size() == 0) {
            final File newFile =
                    new File(segListDir,
                            DataStoreUtils.nameFromOffset(offsetIfCreate, fileSuffix));
            logger.info(sBuilder.append("[File Store] Created ").append(segTypeStr)
                    .append(" segment ").append(newFile.getAbsolutePath()).toString());
            sBuilder.delete(0, sBuilder.length());
            accum.add(new FileSegment(offsetIfCreate, newFile, segType));
        } else {
            // The list of segments is required to be arranged continuously from low to high
            accum.sort(new Comparator<Segment>() {

                @Override
                public int compare(final Segment o1, final Segment o2) {
                    return Long.compare(o1.getStart(), o2.getStart());
                }
            });
            validateSegments(segTypeStr, accum);
            Segment last = accum.get(accum.size() - 1);
            if ((last.getCachedSize() > 0)
                    && (System.currentTimeMillis()
                            - last.getFile().lastModified() >= DataStoreUtils.MAX_FILE_NO_WRITE_DURATION)) {
                // If the last segment is not written for a long time, it will be aged at startup
                final long newOffset = last.getCommitLast();
                final File newFile =
                        new File(segListDir,
                                DataStoreUtils.nameFromOffset(newOffset, fileSuffix));
                logger.info(sBuilder.append("[File Store] Created time roll").append(segTypeStr)
                        .append(" segment ").append(newFile.getAbsolutePath()).toString());
                sBuilder.delete(0, sBuilder.length());
                accum.add(new FileSegment(newOffset, newFile, segType));
            } else {
                last = accum.remove(accum.size() - 1);
                last.close();
                logger.info(sBuilder
                        .append("[File Store] Loading the last ").append(segTypeStr)
                        .append(" segment in mutable mode and running recover on ")
                        .append(last.getFile().getAbsolutePath()).toString());
                sBuilder.delete(0, sBuilder.length());
                final FileSegment mutable =
                        new FileSegment(last.getStart(), last.getFile(), segType, Long.MAX_VALUE);
                accum.add(mutable);
            }
        }
        if (segType == SegmentType.DATA) {
            this.dataSegments = new FileSegmentList(accum.toArray(new Segment[accum.size()]));
        } else {
            this.indexSegments = new FileSegmentList(accum.toArray(new Segment[accum.size()]));
        }
        logger.info(sBuilder.append("[File Store] Loaded ")
                .append(segTypeStr).append(" ").append(accum.size()).append(" segments from ")
                .append(segListDir.getAbsolutePath()).toString());
        sBuilder.delete(0, sBuilder.length());
    }
Flush data to disk at interval.

@throws IOException the exception during processing
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/disk/MsgFileStore.java