POC: Early Exit 2 - sdf.collect(...).to_pandas()

[gwaramadze]

Introduces an ability to collect a small batch of data and inspect interactively in a Python interpreter or a tool like Jupyter Notebook.

1. Like POC 1 makes use of ExitManager, an internal class that facilitates stopping the Application once a certain number of messages has been processed or a timeout in seconds is reached.
2. Adds StreamingDataFrame.collect method:
   • Accepts number and timeout parameters and configures ExitManager.
   • Ensures that a single update operation is appended to the sdf. This operation will collect processed messages into a list. This step will not happen on consecutive sdf.collect calls.
   • Triggers Application.run and Application.reset.
   • The collect name is temporary. I don't like the clash with the windowed aggregation function named collect.
3. Adds to_pandas and to_polars methods, inspired by to_topic. These methods will return either Pandas or Polars dataframe, not our StreamingDataFrame.
4. Note: Application reset will not reset the consumer group, this means that consecutive runs will consume new messages (according to committed offset) instead of reprocessing the stream.
5. Note: There is no need for the user to do app.run()

Example:

from uuid import uuid4
from quixstreams import Application

app = Application(
    broker_address="localhost:19092",
    auto_offset_reset="earliest",
    consumer_group=str(uuid4()),
    use_changelog_topics=False,
)

topic = app.topic("1000000-numbers-100-keys")
sdf = app.dataframe(topic=topic)
sdf.collect(number=2)

# Inspect
sdf.to_pandas()
sdf.to_polars()

# Collect new data
sdf.collect(timeout=3)

# Inspect new data
sdf.to_pandas()
sdf.to_polars()

[daniil-quix]

I like the to_pandas() idea — it looks neat and returns a usable DataFrame (there is no need to initialize it upfront).
We could also pack the .collect() inside to_pandas() to simplify the API.