add tests for shutdown behavior

logprep/connector/confluent_kafka/output.py

@@ -150,16 +150,20 @@ class Config(Output.Config):
         """The topic into which the processed events should be written to."""
         error_topic: str = field(validator=validators.instance_of(str))
         """The topic into which the failed events should be written to."""
-        flush_timeout: float = field(validator=validators.instance_of(float), default=0.1)
+        flush_timeout: float = field(
+            validator=validators.instance_of(float), converter=float, default=0
+        )
         """The maximum time in seconds to wait for the producer to flush the messages
         to kafka broker. If the buffer is full, the producer will block until the buffer
         is empty or the timeout is reached. This implies that the producer does not
         wait for all messages to be send to the broker, if the timeout is reached
-        before the buffer is empty. Default is :code:`0.1`.
+        before the buffer is empty. Default is :code:`0`.
         """
-        send_timeout: float = field(validator=validators.instance_of(float), default=0.1)
+        send_timeout: float = field(
+            validator=validators.instance_of(float), converter=float, default=0
+        )
         """The maximum time in seconds to wait for an answer from the broker on polling.
-        Default is :code:`0.1`."""
+        Default is :code:`0`."""
         kafka_config: Optional[MappingProxyType] = field(
             validator=[
                 validators.instance_of(MappingProxyType),
@@ -338,9 +342,19 @@ def shut_down(self) -> None:
         """ensures that all messages are flushed. According to
         https://confluent-kafka-python.readthedocs.io/en/latest/#confluent_kafka.Producer.flush
         flush without the timeout parameter will block until all messages are delivered.
+        This ensures no messages will get lost on shutdown.
         """
-        if self._producer is not None:
-            self._producer.flush()
+        if self._producer is None:
+            return
+        remaining_messages = self._producer.flush()
+        if remaining_messages:
+            self.metrics.number_of_errors += 1
+            logger.error(
+                "Flushing producer timed out. %s messages are still in the buffer.",
+                remaining_messages,
+            )
+        else:
+            logger.info("Producer flushed successfully. %s messages remaining.", remaining_messages)
 
     def health(self) -> bool:
         """Check the health of kafka producer."""

tests/unit/connector/test_confluent_kafka_output.py

@@ -195,3 +195,12 @@ def test_health_counts_metrics_on_kafka_exception(self):
         self.object._producer.list_topics.side_effect = KafkaException("test error")
         assert not self.object.health()
         assert self.object.metrics.number_of_errors == 1
+
+    def test_shutdown_logs_and_counts_error_if_queue_not_fully_flushed(self):
+        self.object.metrics.number_of_errors = 0
+        self.object._producer = mock.MagicMock()
+        self.object._producer.flush.return_value = 1
+        with mock.patch("logging.Logger.error") as mock_error:
+            self.object.shut_down()
+        mock_error.assert_called()
+        self.object.metrics.number_of_errors = 1