Method _run

sdk/python/feast/feature_server.py:1063–1098

Background loop: periodically checks for buffers that should be flushed based on time since last flush.

(self)

    # ---------- Internal helpers ----------

    def _run(self) -> None:
        """
        Background loop: periodically checks for buffers that should be flushed
        based on time since last flush.
        """
        interval = max(1, int(getattr(self._cfg, "batch_interval_seconds", 30)))
        logger.debug(
            "OfflineWriteBatcher background loop started with check interval=%s",
            interval,
        )

        while not self._stop_event.wait(timeout=interval):
            now = time.time()
            try:
                with self._lock:
                    keys_to_flush: List[_OfflineBatchKey] = []
                    for key, dfs in list(self._buffers.items()):
                        if not dfs:
                            continue
                        last = self._last_flush[
                            key
                        ]  # this will also init the default timestamp
                        age = now - last
                        if age >= self._cfg.batch_interval_seconds:
                            logger.debug(
                                "OfflineWriteBatcher time threshold reached for %s: age=%s",
                                key,
                                age,
                            )
                            keys_to_flush.append(key)
                for key in keys_to_flush:
                    self._flush(key)
            except Exception:
                logger.exception("Error in OfflineWriteBatcher background loop")

        logger.debug("OfflineWriteBatcher background loop exiting")

    def _drain_locked(self, key: _OfflineBatchKey) -> Optional[List[pd.DataFrame]]:
        """