Refactor RowConsumer and RowProducer (#861)

* Rename `RowConsumer` -> `InternalConsumer`
* Rename `RowProducer` -> `InternalProducer`
* Rename internal consumer test folder
* Merge TransactionalProducer to Producer

Author: daniil-quix

quixstreams/app.py

@@ -31,6 +31,8 @@
     ProducerErrorCallback,
     default_on_processing_error,
 )
+from .internal_consumer import InternalConsumer
+from .internal_producer import InternalProducer
 from .kafka import AutoOffsetReset, ConnectionConfig, Consumer, Producer
 from .logging import LogLevel, configure_logging
 from .models import (
@@ -50,8 +52,6 @@
     is_quix_deployment,
 )
 from .processing import ProcessingContext
-from .rowconsumer import RowConsumer
-from .rowproducer import RowProducer
 from .runtracker import RunTracker
 from .sinks import SinkManager
 from .sources import BaseSource, SourceException, SourceManager
@@ -66,7 +66,7 @@
 ProcessingGuarantee = Literal["at-least-once", "exactly-once"]
 MessageProcessedCallback = Callable[[str, int, int], None]
 
-# Enforce idempotent producing for the internal RowProducer
+# Enforce idempotent producing for InternalProducer
 _default_producer_extra_config = {"enable.idempotence": True}
 
 # Default config for the internal consumer
@@ -198,8 +198,8 @@ def __init__(
             Default - `"state"`.
         :param rocksdb_options: RocksDB options.
             If `None`, the default options will be used.
-        :param consumer_poll_timeout: timeout for `RowConsumer.poll()`. Default - `1.0`s
-        :param producer_poll_timeout: timeout for `RowProducer.poll()`. Default - `0`s.
+        :param consumer_poll_timeout: timeout for `InternalConsumer.poll()`. Default - `1.0`s
+        :param producer_poll_timeout: timeout for `InternalProducer.poll()`. Default - `0`s.
         :param on_message_processed: a callback triggered when message is successfully
             processed.
         :param loglevel: a log level for "quixstreams" logger.
@@ -227,11 +227,11 @@ def __init__(
             exceptions occur on different stages of stream processing. If the callback
             returns `True`, the exception will be ignored. Otherwise, the exception
             will be propagated and the processing will eventually stop.
-        :param on_consumer_error: triggered when internal `RowConsumer` fails
+        :param on_consumer_error: triggered when internal `InternalConsumer` fails
             to poll Kafka or cannot deserialize a message.
         :param on_processing_error: triggered when exception is raised within
             `StreamingDataFrame.process()`.
-        :param on_producer_error: triggered when `RowProducer` fails to serialize
+        :param on_producer_error: triggered when `InternalProducer` fails to serialize
             or to produce a message to Kafka.
         <br><br>***Quix Cloud Parameters***<br>
         :param quix_config_builder: instance of `QuixKafkaConfigsBuilder` to be used
@@ -335,11 +335,11 @@ def __init__(
         self._on_message_processed = on_message_processed
         self._on_processing_error = on_processing_error or default_on_processing_error
 
-        self._consumer = self._get_rowconsumer(
+        self._consumer = self._get_internal_consumer(
             on_error=on_consumer_error,
             extra_config_overrides=consumer_extra_config_overrides,
         )
-        self._producer = self._get_rowproducer(on_error=on_producer_error)
+        self._producer = self._get_internal_producer(on_error=on_producer_error)
         self._running = False
         self._failed = False
 
@@ -577,21 +577,21 @@ def stop(self, fail: bool = False):
         if self._state_manager.using_changelogs:
             self._state_manager.stop_recovery()
 
-    def _get_rowproducer(
+    def _get_internal_producer(
         self,
         on_error: Optional[ProducerErrorCallback] = None,
         transactional: Optional[bool] = None,
-    ) -> RowProducer:
+    ) -> InternalProducer:
         """
-        Create a RowProducer using the application config
+        Create InternalProducer using the application config
 
         Used to create the application producer as well as the sources producers
         """
 
         if transactional is None:
             transactional = self._config.exactly_once
 
-        return RowProducer(
+        return InternalProducer(
             broker_address=self._config.broker_address,
             extra_config=self._config.producer_extra_config,
             flush_timeout=self._config.flush_timeout,
@@ -628,13 +628,13 @@ def get_producer(self) -> Producer:
             extra_config=self._config.producer_extra_config,
         )
 
-    def _get_rowconsumer(
+    def _get_internal_consumer(
         self,
         on_error: Optional[ConsumerErrorCallback] = None,
         extra_config_overrides: Optional[dict] = None,
-    ) -> RowConsumer:
+    ) -> InternalConsumer:
         """
-        Create a RowConsumer using the application config
+        Create an InternalConsumer using the application config
 
         Used to create the application consumer as well as the sources consumers
         """
@@ -644,7 +644,7 @@ def _get_rowconsumer(
             **self._config.consumer_extra_config,
             **extra_config_overrides,
         }
-        return RowConsumer(
+        return InternalConsumer(
             broker_address=self._config.broker_address,
             consumer_group=self._config.consumer_group,
             auto_offset_reset=self._config.auto_offset_reset,
@@ -734,8 +734,8 @@ def add_source(self, source: BaseSource, topic: Optional[Topic] = None) -> Topic
         self._source_manager.register(
             source,
             topic,
-            self._get_rowproducer(transactional=False),
-            self._get_rowconsumer(
+            self._get_internal_producer(transactional=False),
+            self._get_internal_consumer(
                 extra_config_overrides=consumer_extra_config_overrides
             ),
             self._get_topic_manager(),

quixstreams/checkpointing/checkpoint.py

@@ -6,8 +6,8 @@
 from confluent_kafka import KafkaException, TopicPartition
 
 from quixstreams.dataframe import DataFrameRegistry
-from quixstreams.rowconsumer import RowConsumer
-from quixstreams.rowproducer import RowProducer
+from quixstreams.internal_consumer import InternalConsumer
+from quixstreams.internal_producer import InternalProducer
 from quixstreams.sinks import SinkManager
 from quixstreams.sinks.base import SinkBackpressureError
 from quixstreams.state import (
@@ -122,8 +122,8 @@ class Checkpoint(BaseCheckpoint):
     def __init__(
         self,
         commit_interval: float,
-        producer: RowProducer,
-        consumer: RowConsumer,
+        producer: InternalProducer,
+        consumer: InternalConsumer,
         state_manager: StateStoreManager,
         sink_manager: SinkManager,
         dataframe_registry: DataFrameRegistry,

quixstreams/internal_consumer/__init__.py

@@ -0,0 +1 @@
+from .consumer import InternalConsumer as InternalConsumer

quixstreams/rowconsumer/buffering.py renamed to quixstreams/internal_consumer/buffering.py

@@ -20,7 +20,7 @@
 
 logger = logging.getLogger(__name__)
 
-__all__ = ("RowConsumer",)
+__all__ = ("InternalConsumer",)
 
 
 def _validate_message_batch(
@@ -37,10 +37,9 @@ def _validate_message_batch(
             raise
 
 
-class RowConsumer(BaseConsumer):
+class InternalConsumer(BaseConsumer):
     _backpressure_resume_at: float
 
-    # TODO: Rename it to "InternalConsumer"? Or "InternalBufferedConsumer"?
     def __init__(
         self,
         broker_address: Union[str, ConnectionConfig],
@@ -55,14 +54,12 @@ def __init__(
         max_partition_buffer_size: int = 10000,
     ):
         """
-
         A consumer class that is capable of deserializing Kafka messages to Rows
         according to the Topics deserialization settings.
 
         It overrides `.subscribe()` method of Consumer class to accept `Topic`
         objects instead of strings.
 
-
         :param broker_address: Connection settings for Kafka.
             Accepts string with Kafka broker host and port formatted as `<host>:<port>`,
             or a ConnectionConfig object if authentication is required.
@@ -79,7 +76,8 @@ def __init__(
         :param extra_config: A dictionary with additional options that
             will be passed to `confluent_kafka.Consumer` as is.
             Note: values passed as arguments override values in `extra_config`.
-        :param on_error: a callback triggered when `RowConsumer.poll_row` fails.
+        :param on_error: a callback triggered when InternalConsumer fails
+            to get and deserialize a new message.
             If consumer fails and the callback returns `True`, the exception
             will be logged but not propagated.
             The default callback logs an exception and returns `False`.

quixstreams/rowconsumer/consumer.py renamed to quixstreams/internal_consumer/consumer.py

@@ -15,7 +15,6 @@
     PRODUCER_ON_ERROR_RETRIES,
     PRODUCER_POLL_TIMEOUT,
     Producer,
-    TransactionalProducer,
 )
 from .models import Headers, Row, Topic
 
@@ -74,7 +73,7 @@ def wrapper(*args, **kwargs):
 class KafkaProducerTransactionCommitFailed(QuixException): ...
 
 
-class RowProducer:
+class InternalProducer:
     """
     A producer class that is capable of serializing Rows to bytes and send them to Kafka.
     The serialization is performed according to the Topic serialization settings.
@@ -85,8 +84,8 @@ class RowProducer:
     :param extra_config: A dictionary with additional options that
         will be passed to `confluent_kafka.Producer` as is.
         Note: values passed as arguments override values in `extra_config`.
-    :param on_error: a callback triggered when `RowProducer.produce_row()`
-        or `RowProducer.poll()` fail`.
+    :param on_error: a callback triggered when `InternalProducer.produce_row()`
+        or `InternalProducer.poll()` fail`.
         If producer fails and the callback returns `True`, the exception
         will be logged but not propagated.
         The default callback logs an exception and returns `False`.
@@ -103,18 +102,12 @@ def __init__(
         flush_timeout: Optional[float] = None,
         transactional: bool = False,
     ):
-        if transactional:
-            self._producer: Producer = TransactionalProducer(
-                broker_address=broker_address,
-                extra_config=extra_config,
-                flush_timeout=flush_timeout,
-            )
-        else:
-            self._producer = Producer(
-                broker_address=broker_address,
-                extra_config=extra_config,
-                flush_timeout=flush_timeout,
-            )
+        self._producer = Producer(
+            broker_address=broker_address,
+            extra_config=extra_config,
+            flush_timeout=flush_timeout,
+            transactional=transactional,
+        )
 
         self._on_error: ProducerErrorCallback = on_error or default_on_producer_error
         self._tp_offsets: Dict[Tuple[str, int], int] = {}

quixstreams/rowproducer.py renamed to quixstreams/internal_producer.py

@@ -30,3 +30,6 @@ class KafkaConsumerException(BaseKafkaException): ...
 
 
 class KafkaProducerDeliveryError(BaseKafkaException): ...
+
+
+class InvalidProducerConfigError(QuixException): ...