Feature: dynamic target topic routing (#976)

Co-authored-by: Daniil Gusev <[email protected]>

Author: gwaramadze

docs/processing.md

@@ -385,7 +385,65 @@ To publish the current value of the `StreamingDataFrame` to a topic, simply call
 from `Application.topic()`
 as an argument.
 
-Similarly to other methods, `.to_topic()` creates a new `StreamingDataFrame` object.
+Similarly to other methods, `.to_topic()` creates a new `StreamingDataFrame` object:
+
+```python
+from quixstreams import Application
+
+app = Application(broker_address='localhost:9092')
+
+# Declare input and output topics
+input_topic = app.topic('input', value_deserializer='json')
+output_topic = app.topic('output', value_serializer='json')
+
+sdf = app.dataframe(input_topic)
+ 
+# Produce data to the output topic
+sdf = sdf.to_topic(topic=output_topic)
+```
+
+### Splitting data into multiple topics
+
+To dynamically route messages to different topics based on the message content, you can provide a callable that returns a `Topic` object to the `StreamingDataFrame.to_topic()` method:
+
+```python
+from quixstreams import Application
+from typing import Any
+
+app = Application(broker_address='localhost:9092')
+
+# Declare topics
+input_topic = app.topic('sensor-data', value_deserializer='json')
+normal_topic = app.topic('normal-readings', value_serializer='json')
+alert_topic = app.topic('high-temp-alerts', value_serializer='json')
+
+sdf = app.dataframe(input_topic)
+ 
+def route_by_temperature(value: Any, key: Any, timestamp: int, headers):
+    """
+    Send messages to different topics based on temperature sensor value
+    """
+    if value.get('temperature', 0) > 80:
+        return alert_topic
+    else:
+        return normal_topic
+
+sdf = sdf.to_topic(topic=route_by_temperature)
+
+# You can also combine it together with `key` transformation to change the message key
+sdf = sdf.to_topic(
+    topic=route_by_temperature,
+    key=lambda value: f"sensor-{value['sensor_id']}"
+)
+```
+
+!!! warning "Important Considerations"
+    
+    We recommend declaring all `Topic` instances before staring the application instead of creating them dynamically within the passed callback.
+    
+    Creating topics dynamically can lead to accidentally creating numerous topics and saturating the broker's partitions limits.  
+    Also, each `app.topic()` calls checks if the topic exists in the cluster and attempts to create the missing one when `auto_create_topics=True` is passed to the `Application`.
+
 
 ### Changing Message Key Before Producing
 

quixstreams/dataframe/dataframe.py

@@ -667,8 +667,24 @@ def callback(value, *_):
 
         return StreamingSeries.from_apply_callback(callback, sdf_id=id(self))
 
+    @overload
     def to_topic(
-        self, topic: Topic, key: Optional[Callable[[Any], Any]] = None
+        self,
+        topic: Topic,
+        key: Optional[Callable[[Any], Any]] = None,
+    ) -> "StreamingDataFrame": ...
+
+    @overload
+    def to_topic(
+        self,
+        topic: Callable[[Any, Any, int, Any], Topic],
+        key: Optional[Callable[[Any], Any]] = None,
+    ) -> "StreamingDataFrame": ...
+
+    def to_topic(
+        self,
+        topic: Union[Topic, Callable[[Any, Any, int, Any], Topic]],
+        key: Optional[Callable[[Any], Any]] = None,
     ) -> "StreamingDataFrame":
         """
         Produce current value to a topic. You can optionally specify a new key.
@@ -692,18 +708,40 @@ def to_topic(
         sdf = sdf.to_topic(output_topic_0)
         # does not require reassigning
         sdf.to_topic(output_topic_1, key=lambda data: data["a_field"])
+
+        # Dynamic topic selection based on message content
+        def select_topic(value, key, timestamp, headers):
+            if value.get("priority") == "high":
+                return output_topic_0
+            else:
+                return output_topic_1
+
+        sdf = sdf.to_topic(select_topic)
         ```
 
-        :param topic: instance of `Topic`
+        :param topic: instance of `Topic` or a callable that returns a `Topic`.
+            If a callable is provided, it will receive four arguments:
+            value, key, timestamp, and headers of the current message.
+            The callable must return a `Topic` object.
+            **Important**: We recommend declaring all `Topic` instances before
+            staring the application instead of creating them dynamically
+            within the passed callback. Creating topics dynamically can lead
+            to accidentally creating numerous topics and
+            saturating the broker's partitions limits.
         :param key: a callable to generate a new message key, optional.
             If passed, the return type of this callable must be serializable
             by `key_serializer` defined for this Topic object.
             By default, the current message key will be used.
         :return: the updated StreamingDataFrame instance (reassignment NOT required).
         """
+        if isinstance(topic, Topic):
+            topic_callback = lambda value, orig_key, timestamp, headers: topic
+        else:
+            topic_callback = topic
+
         return self._add_update(
             lambda value, orig_key, timestamp, headers: self._produce(
-                topic=topic,
+                topic=topic_callback(value, orig_key, timestamp, headers),
                 value=value,
                 key=orig_key if key is None else key(value),
                 timestamp=timestamp,

tests/test_quixstreams/test_dataframe/test_dataframe.py

@@ -17,6 +17,7 @@
 from quixstreams.dataframe.windows.base import WindowResult
 from quixstreams.models import TopicConfig
 from quixstreams.models.topics.exceptions import TopicPartitionsMismatch
+from quixstreams.models.topics.topic import Topic
 from quixstreams.utils.stream_id import stream_id_from_strings
 from tests.utils import DummySink
 
@@ -779,6 +780,48 @@ def test_to_topic_multiple_topics_out(
             assert consumed_row.key == key
             assert consumed_row.value == value
 
+    def test_to_topic_dynamic_topic(
+        self,
+        dataframe_factory,
+        internal_consumer_factory,
+        internal_producer_factory,
+        topic_manager_topic_factory,
+        message_context_factory,
+    ):
+        topic_0 = topic_manager_topic_factory(
+            value_serializer="json", value_deserializer="json"
+        )
+        topic_1 = topic_manager_topic_factory(
+            value_serializer="json", value_deserializer="json"
+        )
+
+        def topic_callback(value: Any, key: Any, timestamp: int, headers: Any) -> Topic:
+            return topic_0 if value == 0 else topic_1
+
+        producer = internal_producer_factory()
+
+        sdf = dataframe_factory(producer=producer)
+        sdf = sdf.to_topic(topic_callback)
+
+        key, timestamp = b"key", 10
+        ctx = message_context_factory()
+
+        with producer:
+            sdf.test(value=0, key=key, timestamp=timestamp, ctx=ctx)
+            sdf.test(value=1, key=key, timestamp=timestamp, ctx=ctx)
+
+        with internal_consumer_factory(auto_offset_reset="earliest") as consumer:
+            consumer.subscribe([topic_0])
+            row_0 = consumer.poll_row(timeout=5.0)
+            assert row_0.topic == topic_0.name
+            assert row_0.value == 0
+
+        with internal_consumer_factory(auto_offset_reset="earliest") as consumer:
+            consumer.subscribe([topic_1])
+            row_1 = consumer.poll_row(timeout=5.0)
+            assert row_1.topic == topic_1.name
+            assert row_1.value == 1
+
 
 class TestStreamingDataframeStateful:
     def test_apply_stateful(