private void refreshLocalRT() throws Exception {
        // Declare local segments to zk
        List<RTSGroupInfo> localInfos = localRealtimeTable
                .rtsGroups()
                .values()
                .stream()
                // We keep the uploaded segments and wait for deletion delay exceed.
                //.filter(rtsg -> !rtsg.hasUploaded())
                .map(rtsg -> {
                    try {
                        SegmentSchema schema = rtsg.schema();
                        ColumnNode[] columnNodes = new ColumnNode[schema.columns.size()];
                        for (int i = 0; i < columnNodes.length; i++) {
                            columnNodes[i] = rtsg.columnNode(i);
                        }
                        return new RTSGroupInfo(rtsg.version(), rtsg.mode().name(), rtsg.name(), schema, rtsg.rowCount(), columnNodes, hostName);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                })
                .collect(Collectors.toList());
        String json = JsonUtil.toJson(new HostSegmentInfo(localInfos));
        String zkPath = declarePath + "/" + hostName;

        byte[] rtsgInfo = GenericCompression.compress(json.getBytes("utf-8"));
        if (lastRTSGInfo != null
                && rtsgInfo.length == lastRTSGInfo.length
                && ByteArrayMethods.arrayEquals(
                rtsgInfo, Platform.BYTE_ARRAY_OFFSET,
                lastRTSGInfo, Platform.BYTE_ARRAY_OFFSET, rtsgInfo.length)) {
            // No need to update
            return;
        }
        // Replace.
        try {
            if (zkClient.checkExists().forPath(zkPath) == null) {
                zkClient.create().withMode(CreateMode.EPHEMERAL).forPath(zkPath, rtsgInfo);
            } else {
                zkClient.setData().forPath(zkPath, rtsgInfo);
            }
        } catch (KeeperException.NodeExistsException e) {
            zkClient.setData().forPath(zkPath, rtsgInfo);
        }
        lastRTSGInfo = rtsgInfo;
    }
This segment pool cache all realtime segment fds.
indexr-server/src/main/java/io/indexr/server/rt/RealtimeSegmentPool.java