Add commit_every configuration parameter (#416)

* Fix the validation in Checkpoint.store_offset()`
* Add a `commit_every` config parameter

Author: daniil-quix

docs/advanced/checkpointing.md

@@ -6,7 +6,14 @@ We call this process a “checkpointing”.
 
 The goal of checkpointing is to ensure that applications can recover after failures and reprocess records from Kafka producing the same results as if the failure never happened.
 
-Currently, Quix Streams provides *At-Least-Once* processing guarantees, which means that each incoming message will be processed, but it may happen multiple times and generate duplicated outputs.
+
+Quix Streams supports both *At-Least-Once* and *Exactly-Once* processing guarantees, which can be changed using the `processing_guarantee` parameter.
+
+- When At-Least-Once guarantee is enabled (the default), each incoming message is guaranteed to be processed, but it may happen multiple times and generate duplicated outputs in case of failure
+- If Exactly-Once guarantee is enabled, the outputs are guaranteed to be unique for every message at the cost of increased latency.
+
+See the [Configuration](../configuration.md#processing-guarantees) page to learn more about processing guarantees.  
+ 
 
 ## Under the Hood
 
@@ -61,9 +68,9 @@ During recovery, the app will apply changelog updates to the local state stores.
 
 ## Configuring the Checkpointing
 
-Users may configure how often the checkpoints are committed by passing the `commit_interval` parameter to the `Application` class.
+Users may configure how often the checkpoints are committed by passing the `commit_interval` and `commit_every` parameters to the `Application` class.
 
-The default commit interval is 5 seconds.
+By default, the `commit_interval` is 5 seconds, and the `commit_every` is 0, and only the commit interval is taken into account. 
 
 Changing the commit interval will have several implications for the application:
 
@@ -80,8 +87,12 @@ Changing the commit interval will have several implications for the application:
     However, it will reduce memory usage and limit the number of potentially reprocessed messages, reducing duplicates.
 
 - If `commit_interval` is set to `0`, the application will commit a checkpoint for every processed Kafka message.
+- If `commit_every` is set, the application will commit after processing N messages across all assigned partitions.
+  - You may use `commit_interval` to get more granular control over the commit schedule.  
+  - For example, if `commit_every=1000` and `commit_interval=5.0`, the application will commit the checkpoint as soon as 1000 messages are processed or 5s interval is elapsed. 
+
+When configuring the `commit_interval` and `commit_every`, take into account such factors as the number of unique keys in the input topics, hardware, and infrastructure.
 
-When configuring the commit interval, take into account such factors as the number of unique keys in the input topics, hardware, and infrastructure.
 
 ## Limitations
 

docs/configuration.md

@@ -26,9 +26,14 @@ Consumer group is also used in the state directory path and as a prefix for chan
 
 - **`commit_interval`** - How often to commit the processed offsets and state in seconds.  
 **Default** - `5.0`.    
-See the [Checkpointing](advanced/checkpointing.md) page for more information about 
+See the [Checkpointing](advanced/checkpointing.md#configuring-the-checkpointing) page for more information about 
 the `commit_interval` parameter.
 
+- **`commit_every`** - Commit the checkpoint after processing N offsets.  
+**Default** - `0`.  
+See the [Checkpointing](advanced/checkpointing.md#configuring-the-checkpointing) page for more information about 
+the `commit_every` parameter.
+
 - **`auto_offset_reset`** - Consumer `auto.offset.reset` setting.  
 It determines where the consumer should start reading messages from.  
 See more `auto.offset.reset` in this [article](https://www.quix.io/blog/kafka-auto-offset-reset-use-cases-and-pitfalls#the-auto-offset-reset-configuration).  

quixstreams/app.py

@@ -98,6 +98,7 @@ def __init__(
         consumer_group: Optional[str] = None,
         auto_offset_reset: AutoOffsetReset = "latest",
         commit_interval: float = 5.0,
+        commit_every: int = 0,
         consumer_extra_config: Optional[dict] = None,
         producer_extra_config: Optional[dict] = None,
         state_dir: str = "state",
@@ -138,6 +139,15 @@ def __init__(
               >***NOTE:*** Quix Applications will prefix it with the Quix workspace id.
         :param commit_interval: How often to commit the processed messages in seconds.
             Default - 5.0.
+        :param commit_every: Commit the checkpoint after processing N messages.
+            Use this parameter for more granular control of the commit schedule.
+            If the value is > 0, the application will commit the checkpoint after
+            processing the specified number of messages across all the assigned
+            partitions.
+            If the value is <= 0, only the `commit_interval` will be considered.
+            Default - 0.
+                >***NOTE:*** Only input offsets are counted, and the application
+                > may produce more results than the number of incoming messages.
         :param auto_offset_reset: Consumer `auto.offset.reset` setting
         :param consumer_extra_config: A dictionary with additional options that
             will be passed to `confluent_kafka.Consumer` as is.
@@ -256,6 +266,7 @@ def __init__(
         self._consumer_group = consumer_group
         self._auto_offset_reset = auto_offset_reset
         self._commit_interval = commit_interval
+        self._commit_every = commit_every
         self._producer_extra_config = producer_extra_config
         self._consumer_extra_config = consumer_extra_config
         self._processing_guarantee = processing_guarantee
@@ -312,6 +323,7 @@ def __init__(
         )
         self._processing_context = ProcessingContext(
             commit_interval=self._commit_interval,
+            commit_every=commit_every,
             producer=self._producer,
             consumer=self._consumer,
             state_manager=self._state_manager,
@@ -730,6 +742,7 @@ def run(
             f'consumer_group="{self._consumer_group}" '
             f'auto_offset_reset="{self._auto_offset_reset}" '
             f"commit_interval={self._commit_interval}s "
+            f"commit_every={self._commit_every} "
             f'processing_guarantee="{self._processing_guarantee}"'
         )
         if self.is_quix_app:

quixstreams/checkpointing/checkpoint.py

@@ -34,6 +34,7 @@ def __init__(
         consumer: Consumer,
         state_manager: StateStoreManager,
         exactly_once: bool = False,
+        commit_every: int = 0,
     ):
         self._created_at = time.monotonic()
         # A mapping of <(topic, partition): processed offset>
@@ -47,15 +48,21 @@ def __init__(
         self._consumer = consumer
         self._producer = producer
         self._exactly_once = exactly_once
+        self._commit_every = commit_every
+        self._total_offsets_processed = 0
 
         if self._exactly_once:
             self._producer.begin_transaction()
 
     def expired(self) -> bool:
         """
-        Returns `True` if checkpoint deadline has expired.
+        Returns `True` if checkpoint deadline has expired OR
+        if the total number of processed offsets exceeded the "commit_every" limit
+        when it's defined.
         """
-        return (time.monotonic() - self._commit_interval) >= self._created_at
+        return (time.monotonic() - self._commit_interval) >= self._created_at or (
+            0 < self._commit_every <= self._total_offsets_processed
+        )
 
     def empty(self) -> bool:
         """
@@ -77,12 +84,13 @@ def store_offset(self, topic: str, partition: int, offset: int):
         # same checkpoint.
         # It shouldn't normally happen, but a lot of logic relies on it,
         # and it's better to be safe.
-        if offset < stored_offset:
+        if offset <= stored_offset:
             raise InvalidStoredOffset(
                 f"Cannot store offset smaller or equal than already processed"
                 f" one: {offset} <= {stored_offset}"
             )
         self._tp_offsets[(topic, partition)] = offset
+        self._total_offsets_processed += 1
 
     def get_store_transaction(
         self, topic: str, partition: int, store_name: str = DEFAULT_STATE_STORE_NAME

quixstreams/processing_context.py

@@ -29,6 +29,7 @@ class ProcessingContext:
     consumer: RowConsumer
     state_manager: StateStoreManager
     exactly_once: bool = False
+    commit_every: int = 0
     _checkpoint: Optional[Checkpoint] = dataclasses.field(
         init=False, repr=False, default=None
     )
@@ -53,9 +54,10 @@ def init_checkpoint(self):
         """
         Initialize a new checkpoint
         """
-        logger.debug(f"Starting a checkpoint...")
+        logger.debug(f"Initializing a checkpoint...")
         self._checkpoint = Checkpoint(
             commit_interval=self.commit_interval,
+            commit_every=self.commit_every,
             state_manager=self.state_manager,
             producer=self.producer,
             consumer=self.consumer,

tests/test_quixstreams/test_checkpointing.py

@@ -21,13 +21,15 @@
 def checkpoint_factory(state_manager, consumer, row_producer_factory):
     def factory(
         commit_interval: float = 1,
+        commit_every: int = 0,
         consumer_: Optional[Consumer] = None,
         producer_: Optional[RowProducer] = None,
         state_manager_: Optional[StateStoreManager] = None,
         exactly_once: bool = False,
     ):
         return Checkpoint(
             commit_interval=commit_interval,
+            commit_every=commit_every,
             producer=producer_ or row_producer_factory(transactional=exactly_once),
             consumer=consumer_ or consumer,
             state_manager=state_manager_ or state_manager,
@@ -60,15 +62,30 @@ def test_exactly_once_init(self, checkpoint_factory):
         mock_producer.begin_transaction.assert_called()
 
     @pytest.mark.parametrize("commit_interval, expired", [(0, True), (999, False)])
-    def test_expired(self, commit_interval, expired, checkpoint_factory):
+    def test_expired_with_commit_interval(
+        self, commit_interval, expired, checkpoint_factory
+    ):
         checkpoint = checkpoint_factory(commit_interval=commit_interval)
         assert checkpoint.expired() == expired
 
+    def test_expired_with_commit_every(self, checkpoint_factory):
+        checkpoint = checkpoint_factory(commit_interval=999, commit_every=2)
+        checkpoint.store_offset("topic", 0, 0)
+        assert not checkpoint.expired()
+
+        checkpoint.store_offset("topic", 0, 1)
+        assert checkpoint.expired()
+
+    def test_expired_with_commit_every_and_commit_interval(self, checkpoint_factory):
+        checkpoint = checkpoint_factory(commit_interval=0, commit_every=10)
+        checkpoint.store_offset("topic", 0, 0)
+        assert checkpoint.expired()
+
     def test_store_already_processed_offset_fails(self, checkpoint_factory):
         checkpoint = checkpoint_factory()
         checkpoint.store_offset("topic", 0, 10)
         with pytest.raises(InvalidStoredOffset):
-            checkpoint.store_offset("topic", 0, 9)
+            checkpoint.store_offset("topic", 0, 10)
 
     @pytest.mark.parametrize("exactly_once", [False, True])
     def test_commit_no_state_success(