Actually make use of Error Monitor and add configuration option for it

elastic · Jul 23, 2024 · 19d9fcc · 19d9fcc
1 parent fda81d2
commit 19d9fcc
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 1 deletion.
diff --git a/connectors/config.py b/connectors/config.py
@@ -92,6 +92,7 @@ def _default_config():
                 "max_consecutive_errors": 10,
                 "max_error_rate": 0.15,
                 "error_window_size": 100,
+                "error_queue_size": 10,
             },
             "max_concurrent_content_syncs": 1,
             "max_concurrent_access_control_syncs": 1,

diff --git a/connectors/es/sink.py b/connectors/es/sink.py
@@ -917,7 +917,9 @@ def _extractor_task_running(self):
 
     async def cancel(self):
         if self._sink_task_running():
-            self._logger.info(f"Canceling the Sink task: {self._sink_task.get_name()}")
+            self._logger.info(
+                f"Canceling the Sink task: {self._sink_task.get_name()}"  # pyright: ignore
+            )
             self._sink_task.cancel()
         else:
             self._logger.debug(

diff --git a/tests/test_sink.py b/tests/test_sink.py
@@ -1349,7 +1349,16 @@ async def test_extractor_put_doc():
     doc = {"id": 123}
     queue = Mock()
     queue.put = AsyncMock()
+<<<<<<< HEAD
     extractor = Extractor(None, queue, INDEX, error_monitor=Mock())
+=======
+    extractor = Extractor(
+        None,
+        queue,
+        INDEX,
+        error_monitor=Mock()
+    )
+>>>>>>> b47bf686 (Actually make use of Error Monitor and add configuration option for it)
 
     await extractor.put_doc(doc)
     queue.put.assert_awaited_once_with(doc)
@@ -1388,7 +1397,16 @@ async def test_force_canceled_extractor_put_doc():
     doc = {"id": 123}
     queue = Mock()
     queue.put = AsyncMock()
+<<<<<<< HEAD
     extractor = Extractor(None, queue, INDEX, error_monitor=Mock())
+=======
+    extractor = Extractor(
+        None,
+        queue,
+        INDEX,
+        error_monitor=Mock()
+    )
+>>>>>>> b47bf686 (Actually make use of Error Monitor and add configuration option for it)
 
     extractor.force_cancel()
     with pytest.raises(ForceCanceledError):
@@ -1595,7 +1613,16 @@ def _put_side_effect(value):
     # instances
     doc_generator = AsyncIterator([(doc, None, "") for doc in docs_from_source])
 
+<<<<<<< HEAD
     extractor = Extractor(es_client, queue, INDEX, error_monitor=Mock())
+=======
+    extractor = Extractor(
+        es_client,
+        queue,
+        INDEX,
+        error_monitor=Mock()
+    )
+>>>>>>> b47bf686 (Actually make use of Error Monitor and add configuration option for it)
 
     await extractor.run(doc_generator, JobType.FULL)