private long ingest() {
        synchronized (ingestThreadLock) {
            if (!isIngestState()) {
                if (Thread.currentThread() == ingestThread) {
                    ingestThread = null;
                }
                return -1;
            }
            if (!setting.ingest) {
                if (Thread.currentThread() == ingestThread) {
                    ingestThread = null;
                }
                state = State.STOP;
                return -1;
            }
            if (Thread.currentThread() != ingestThread) {
                return -1;
            }
        }
        if (!rtResources.reserveMemory(4 << 20)) {
            logger.info("Not enough memory, wait... [table: {}]", tableName);
            return ingestRetryPeriod;
        }

        RTSGroup ingestRTSG = latestRTSG();
        if (ingestRTSG == null || !isRTSGIngestable(ingestRTSG)) {
            if (ingestRTSG != null) {
                // Make the last rtsg upload faster.
                ingestRTSG.disallowAddSegment();
            }
            logger.info("Create new rts group. [table: {}]", tableName);

            ZonedDateTime now = ZonedDateTime.now(setting.timeZone.toZoneId());
            String rtsgPathStr = String.format("rtsg.%s.%s.seg", now.format(timeFormatter), UUID.randomUUID());
            Path rtsgPath = localDir.resolve(rtsgPathStr);
            String rtsgName = "rt/" + rtsgPathStr;
            RTSGroup rtsGroup = Try.on(
                    () -> RTSGroup.open(
                            rtsgPath,
                            tableName,
                            rtsgName,
                            setting.schema,
                            setting.dims,
                            setting.metrics,
                            setting.nameToAlias,
                            setting.grouping,
                            setting.mode),
                    1, logger,
                    String.format("Create new rts group failed. [table: %s, rtsg: %s]", tableName, rtsgName));
            if (rtsGroup == null) {
                return ingestRetryPeriod;
            }

            logger.info("New rts group. [table: {}, rtsg: {}]", tableName, rtsgName);

            addRTSGroup(rtsGroup);
            ingestRTSG = rtsGroup;
        }
        return doIngest(ingestRTSG, setting);
    }
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