fix: use the closed window in the eof response

[vaibhavtiwari33]

Based off of numaproj/numaflow-rs#177

Testing

Updated the stream sorter example's handler to the following:

    async def handler(
        self,
        datums: AsyncIterable[Datum],
        output: NonBlockingIterator,
    ):
        _LOGGER.info("StreamSorter handler started")
        async for datum in datums:
            _LOGGER.info(
                f"Received datum with event time: {datum.event_time}, "
                f"Current latest watermark: {self.latest_wm}, "
                f"Datum watermark: {datum.watermark}"
            )

Created a jitter source, which pauses emission of events with certain keys for 45 seconds. Total 5 keys in the system. Timeout set for the accumulator is 30s

Logged into stream sorter pod to track disk usage across 4 hours (I should've probably installed prometheus and metrics server to gather this detail in my local cluster but I was too lazy to do that for some reason) :

Disk usage

root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
15M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
11M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
14M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
17M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
17M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
29M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
29M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
29M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
23M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
11M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
30M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
27M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
31M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
31M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
18M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
23M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
27M     .
root@jitter-stream-sorter-sorter-0:/var/numaflow/pbq# du -sh
13M     .

Memory usage

root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
29995008
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
27566080
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
33546240
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
35954688
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
45420544
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
45256704
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
38699008
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
39211008
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
44568576
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
50184192
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
51757056
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
52154368
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
41869312
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
46641152
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
46379008
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
51281920
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
33849344
root@jitter-stream-sorter-sorter-0:/# cat /sys/fs/cgroup/memory.current
37580800

Overall values seem mostly consistent across time.

[codecov]

Codecov Report

❌ Patch coverage is 94.44444% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 92.74%. Comparing base (e3d3afd) to head (962ff44).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
.../pynumaflow/accumulator/servicer/async_servicer.py	77.77%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #360      +/-   ##
==========================================
- Coverage   92.74%   92.74%   -0.01%     
==========================================
  Files          67       67              
  Lines        3514     3540      +26     
  Branches      228      232       +4     
==========================================
+ Hits         3259     3283      +24     
- Misses        190      191       +1     
- Partials       65       66       +1     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[BulkBeing]

Don't ignore these errors. If there's an error test should fail right?

[vaibhavtiwari33]

I'll mark the test as fail in the newly added tests.

[vaibhavtiwari33]

done

[BulkBeing]

I think you can do like this:

_close_window: KeyedWindow | None = field(default=None, init=False)

This value is not known/set at initialization of AccumulatorResult right? With above, it's not part of the constructor.

[vaibhavtiwari33]

updated ty!

[BulkBeing]

close_window is already a read-only property. Let's make this as well property setter ?

@close_window.setter
def close_window(self, window: KeyedWindow):
    if not isinstance(window, KeyedWindow):
        raise TypeError("window must be a KeyedWindow object")
    self._close_window = window

[vaibhavtiwari33]

updated ty!