private long doIngest(RTSGroup rtsGroup, RealtimeSetting setting) {
        Boolean ok = Try.on(
                () -> setting.fetcher.ensure(setting.schema),
                1, logger,
                String.format("Ensure ingest fecther stream failed. [table: %s]", tableName));
        if (ok == null || !ok) {
            logger.warn("Fetcher not ready yet, wait. [table: {}, fetcher: {}]", tableName, setting.fetcher);
            return ingestRetryPeriod;
        }
        // The open/close state should match.
        if (!isIngestState()) {
            Try.on(setting.fetcher::close, logger);
            return ingestRetryPeriod;
        }
        Boolean hasNext = Try.on(setting.fetcher::hasNext, 1, logger);
        if (hasNext == null) {
            return ingestRetryPeriod;
        }
        if (!hasNext) {
            return ingestRetryPeriod;
        }
        RealtimeSegment rts = rtsGroup.addSegment(() -> Try.on(
                () -> {
                    ZonedDateTime now = ZonedDateTime.now(setting.timeZone.toZoneId());
                    String segmentName = String.format("rts.%s.%s.seg", now.format(timeFormatter), UUID.randomUUID());
                    Path segmentPath = rtsGroup.path().resolve(segmentName);
                    return RealtimeSegment.open(segmentPath, tableName, segmentName, setting.schema);
                },
                1, logger,
                String.format("Create new segment failed. [table: %s, rtsg: %s]", tableName, rtsGroup.name())));
        if (rts == null) {
            // If rts group is not allow to add new segment any more, then jump out and create a new one.
            return rtsGroup.allowAddSegment() ? ingestRetryPeriod : 0;
        }

        IngestThread thread = (IngestThread) Thread.currentThread();
        Runnable notifyTask = new Runnable() {
            long phase = ingestPhase.get();

            @Override
            public void run() {
                // Make sure the notify task only fire to current phase.
                if (phase == ingestPhase.get() && !thread.isInterrupted() && thread.isInterrupValid()) {
                    thread.interrupt();
                }
            }
        };

        ScheduledFuture f = notifyService.schedule(
                syncRun(notifyTask),
                setting.savePeriodMS + 1000, // Wait a little time before interrupt the thread. It is better that the ingest process could return itself.
                TimeUnit.MILLISECONDS);
        long rt = 0;
        try {
            logger.info("Start ingest new segment. [table: {}, rtsg: {}, segment: {}]", tableName, rtsGroup.name(), rts.name());
            long rowCount = rts.realtimeIngest(
                    setting.dims,
                    setting.metrics,
                    setting.nameToAlias,
                    setting.tagSetting,
                    setting.ignoreStrategy,
                    setting.grouping,
                    setting.fetcher,
                    setting.maxRowInMemory,
                    setting.savePeriodMS,
                    rtResources);
            logger.info("Finish ingest segment. [ingested: {}, got: {} ({})]. [table: {}, rtsg: {}, segment: {}]",
                    rowCount, rts.rowCount(), String.format("%.2f%%", (double) rts.rowCount() / rowCount * 100), tableName, rtsGroup.name(), rts.name());
        } catch (Exception e) {
            logger.error("Ingest segment failed. [table: {}, rtsg: {}, segment: {}]", tableName, rtsGroup.name(), rts.name(), e);
            rt = ingestRetryPeriod;
        } finally {
            f.cancel(false);
            // Clear interrupt flag.
            syncRun(() -> {
                ingestPhase.set(ingestPhase.get() + 1);
                Thread.interrupted();
            }).run();
            Try.on(() -> rts.commitState(RealtimeSegment.State.RTIFinished),
                    1, logger,
                    String.format("Create new segment failed. [table: %s, segment: %s]", tableName, rtsGroup.name()));
        }
        return rt;
    }
Close this realtime table. Job is done.
This method only called before process exit.
indexr-segment/src/main/java/io/indexr/segment/rt/RealtimeTable.java