feat: added batching to feature server /push to offline store (#5683)

Signed-off-by: Jacob Weinhold <[email protected]>

Author: jfw-ppi

sdk/python/feast/feature_server.py

@@ -215,28 +215,35 @@ def get_app(
     fs_cfg = getattr(store.config, "feature_server", None)
     batching_cfg = None
     if fs_cfg is not None:
-        enabled = getattr(fs_cfg, "offline_push_batching_enabled", None)
+        enabled = getattr(fs_cfg, "offline_push_batching_enabled", False)
         batch_size = getattr(fs_cfg, "offline_push_batching_batch_size", None)
         batch_interval_seconds = getattr(
             fs_cfg, "offline_push_batching_batch_interval_seconds", None
         )
 
-        if (
-            enabled is not None
-            or batch_size is not None
-            or batch_interval_seconds is not None
-        ):
-            batching_cfg = SimpleNamespace(
-                enabled=bool(enabled) if enabled is not None else False,
-                batch_size=batch_size if batch_size is not None else 100,
-                batch_interval_seconds=batch_interval_seconds
-                if batch_interval_seconds is not None
-                else 1,
+        if enabled is True:
+            size_ok = isinstance(batch_size, int) and not isinstance(
+                batch_size, bool
+            )
+            interval_ok = isinstance(batch_interval_seconds, int) and not isinstance(
+                batch_interval_seconds, bool
             )
+            if size_ok and interval_ok:
+                batching_cfg = SimpleNamespace(
+                    enabled=True,
+                    batch_size=batch_size,
+                    batch_interval_seconds=batch_interval_seconds,
+                )
+            else:
+                logger.warning(
+                    "Offline write batching enabled but missing or invalid numeric values; "
+                    "disabling batching (batch_size=%r, batch_interval_seconds=%r)",
+                    batch_size,
+                    batch_interval_seconds,
+                )
 
     offline_batcher: Optional[OfflineWriteBatcher] = None
-    enabled = getattr(batching_cfg, "enabled", False)
-    if batching_cfg is not None and isinstance(enabled, bool) and enabled is True:
+    if batching_cfg is not None and batching_cfg.enabled is True:
         offline_batcher = OfflineWriteBatcher(store=store, cfg=batching_cfg)
         logger.debug("Offline write batching is ENABLED")
     else: