Build a collect strategy that support batching

[lynnagara]

Build a processing strategy that collects messages into batches and periodically submits the batch to the next step. This is useful in scenarios where we need to batch messages, e.g. to bulk write on a storage. We want to provide something equivalent to what Python Arroyo does here: https://github.com/getsentry/arroyo/blob/main/arroyo/processing/strategies/streaming/collect.py