Add proactive window timeout functionality

[tomas-quix]

Summary

This PR implements comprehensive proactive window timeout functionality that allows windows to expire based on wall-clock time even when no new messages arrive.

Key Features

• Proactive Expiration: Windows can expire based on wall-clock time without requiring new messages
• Application Integration: Timeout checking integrated into the consumer polling loop
• Key Tracking: Efficient tracking of active window keys and their last activity times
• Memory Safety: Automatic cleanup of expired keys to prevent memory leaks
• Configurable: Enable/disable timeout checking and adjust check intervals

Changes Made

Core Implementation

• Application timeout infrastructure (quixstreams/app.py)

  • Added enable_window_timeout_checking and window_timeout_check_interval configuration
  • Implemented key tracking system with track_window_key() and untrack_window_key()
  • Added _check_window_timeouts() method integrated into consumer polling loop
  • Added automatic cleanup of stale keys to prevent memory leaks

• Window timeout methods (quixstreams/dataframe/windows/time_based.py)

  • Enhanced expire_timeouts_for_key() for proactive timeout checking
  • Improved timeout logic to use last activity time instead of creation time
  • Added proper window state cleanup when timeouts occur

• Processing integration (quixstreams/dataframe/windows/base.py, quixstreams/processing/context.py)

  • Added callback infrastructure for window registration and key tracking
  • Hooked into window processing pipeline to automatically track active keys

• Window registration (quixstreams/dataframe/windows/definitions.py, quixstreams/dataframe/dataframe.py)

  • Fixed window registration to happen after aggregation when windows have proper names
  • Added automatic registration of timeout-enabled windows

Fixes

• Fixed AttributeError: Resolved 'DataFrameRegistry' object has no attribute 'items' error
• Proper timeout semantics: Changed from creation time to last activity time for timeout calculations
• Window name handling: Fixed registration timing so windows have proper names for store access
• Result processing: Fixed MessageContext creation and window result processing

Test Plan

• Unit tests for Application timeout configuration and key tracking
• Integration tests for timeout checking logic and window expiration
• End-to-end testing with real applications
• Memory leak prevention testing with stale key cleanup
• Multiple topic and partition scenarios

Usage Example

from quixstreams import Application

# Create application with timeout checking enabled
app = Application(
    broker_address="localhost:9092",
    consumer_group="my-group",
    enable_window_timeout_checking=True,  # Enable timeout checking (default: True)
    window_timeout_check_interval=5.0,    # Check every 5 seconds (default: 5.0)
)

# Create a window with 30-second timeout
topic = app.topic("input-topic")
sdf = app.dataframe(topic)
sdf = sdf.tumbling_window(
    duration_ms=60000,      # 1-minute windows
    timeout_ms=30000        # Expire after 30 seconds of inactivity
).sum()

# Windows will now expire proactively after 30 seconds even without new messages
app.run()

Compatibility

• ✅ Fully backward compatible - existing code works unchanged
• ✅ Timeout functionality is opt-in via timeout_ms parameter
• ✅ No performance impact when timeouts are not used
• ✅ Configurable timeout checking can be disabled if needed

Files Changed:

The PR includes changes to the following files:

• quixstreams/app.py - Main timeout infrastructure and Application integration
• quixstreams/dataframe/windows/time_based.py - Proactive expiration methods
• quixstreams/dataframe/windows/base.py - Key tracking hooks
• quixstreams/processing/context.py - Callback infrastructure
• quixstreams/dataframe/dataframe.py - Window registration integration
• quixstreams/dataframe/windows/definitions.py - Window creation and registration
• tests/test_quixstreams/test_dataframe/test_windows/test_tumbling.py - Comprehensive tests