Update Application.run() behavior

[daniil-quix]

Main changes

1. Application.run() with the count param now counts the outputs instead of messages.
   It makes the behavior easier to reason about for the users, and also simplifies the inner logic.
   For example, we no longer need to handle messages from the repartition topics separately.

2. Application.run() with count or timeout parameters can now accumulate and return the processed outputs for further debugging.
   This approach is less manual than using a ListSink.

3. StreamingDataFrame.group_by() now filters the data after sending it to the repartition topic to avoid double-counting