fix revoke callback error handling (#797)

* fix revoke callback error handling
* Update CHANGELOG.md
* refactored member_id

Co-authored-by: kmeinerz1 <kaithomas.meinerz@bwi.de>
Co-authored-by: Jörg Zimmermann <101292599+ekneg54@users.noreply.github.com>
Co-authored-by: Jörg Zimmermann <joerg.zimmermann@bwi.de>

Author: 3 people

CHANGELOG.md

@@ -9,6 +9,8 @@
 ### Improvements
 ### Bugfix
 
+- Fixed logging error in _revoke_callback() by adding error handling
+
 ## 16.1.0
 ### Deprecations
 

logprep/connector/confluent_kafka/input.py

@@ -130,11 +130,16 @@ class Metrics(Input.Metrics):
 
         librdkafka_replyq: GaugeMetric = field(
             factory=lambda: GaugeMetric(
-                description="Number of ops (callbacks, events, etc) waiting in queue for application to serve with rd_kafka_poll()",
+                description=(
+                    "Number of ops (callbacks, events, etc) waiting in "
+                    "queue for application to serve with rd_kafka_poll()"
+                ),
                 name="confluent_kafka_input_librdkafka_replyq",
             )
         )
-        """Number of ops (callbacks, events, etc) waiting in queue for application to serve with rd_kafka_poll()"""
+        """Number of ops (callbacks, events, etc) waiting in queue for application
+           to serve with rd_kafka_poll()
+        """
         librdkafka_tx: GaugeMetric = field(
             factory=lambda: GaugeMetric(
                 description="Total number of requests sent to Kafka brokers",
@@ -166,14 +171,22 @@ class Metrics(Input.Metrics):
         """Total number of bytes received from Kafka brokers"""
         librdkafka_rxmsgs: GaugeMetric = field(
             factory=lambda: GaugeMetric(
-                description="Total number of messages consumed, not including ignored messages (due to offset, etc), from Kafka brokers.",
+                description=(
+                    "Total number of messages consumed, not including ignored messages"
+                    "(due to offset, etc), from Kafka brokers."
+                ),
                 name="confluent_kafka_input_librdkafka_rxmsgs",
             )
         )
-        """Total number of messages consumed, not including ignored messages (due to offset, etc), from Kafka brokers."""
+        """Total number of messages consumed, not including ignored messages
+           (due to offset, etc), from Kafka brokers.
+        """
         librdkafka_rxmsg_bytes: GaugeMetric = field(
             factory=lambda: GaugeMetric(
-                description="Total number of message bytes (including framing) received from Kafka brokers",
+                description=(
+                    "Total number of message bytes (including framing)"
+                    "received from Kafka brokers"
+                ),
                 name="confluent_kafka_input_librdkafka_rxmsg_bytes",
             )
         )
@@ -195,11 +208,11 @@ class Metrics(Input.Metrics):
         """Time elapsed since last rebalance (assign or revoke) (milliseconds)."""
         librdkafka_cgrp_rebalance_cnt: GaugeMetric = field(
             factory=lambda: GaugeMetric(
-                description="Total number of rebalances (assign or revoke).",
+                description="Total number of rebalance (assign or revoke).",
                 name="confluent_kafka_input_librdkafka_cgrp_rebalance_cnt",
             )
         )
-        """Total number of rebalances (assign or revoke)."""
+        """Total number of rebalance (assign or revoke)."""
         librdkafka_cgrp_assignment_size: GaugeMetric = field(
             factory=lambda: GaugeMetric(
                 description="Current assignment's partition count.",
@@ -316,7 +329,7 @@ def _error_callback(self, error: KafkaException) -> None:
             the error that occurred
         """
         self.metrics.number_of_errors += 1
-        logger.error(f"{self.describe()}: {error}")
+        logger.error("%s: %s", self.describe(), error)
 
     def _stats_callback(self, stats_raw: str) -> None:
         """Callback for statistics data. This callback is triggered by poll()
@@ -487,12 +500,13 @@ def batch_finished_callback(self) -> None:
         except KafkaException as error:
             raise InputWarning(self, f"{error}, {self._last_valid_record}") from error
 
-    def _assign_callback(self, consumer, topic_partitions):
+    def _assign_callback(self, topic_partitions: list[TopicPartition]) -> None:
         for topic_partition in topic_partitions:
             offset, partition = topic_partition.offset, topic_partition.partition
+            member_id = self._get_memberid()
             logger.info(
                 "%s was assigned to topic: %s | partition %s",
-                consumer.memberid(),
+                member_id,
                 topic_partition.topic,
                 partition,
             )
@@ -503,27 +517,38 @@ def _assign_callback(self, consumer, topic_partitions):
             self.metrics.committed_offsets.add_with_labels(offset, labels)
             self.metrics.current_offsets.add_with_labels(offset, labels)
 
-    def _revoke_callback(self, consumer, topic_partitions):
+    def _revoke_callback(self, topic_partitions: list[TopicPartition]) -> None:
+
         for topic_partition in topic_partitions:
             self.metrics.number_of_warnings += 1
+            member_id = self._get_memberid()
             logger.warning(
                 "%s to be revoked from topic: %s | partition %s",
-                consumer.memberid(),
+                member_id,
                 topic_partition.topic,
                 topic_partition.partition,
             )
         self.batch_finished_callback()
 
-    def _lost_callback(self, consumer, topic_partitions):
+    def _lost_callback(self, topic_partitions: list[TopicPartition]) -> None:
         for topic_partition in topic_partitions:
             self.metrics.number_of_warnings += 1
+            member_id = self._get_memberid()
             logger.warning(
                 "%s has lost topic: %s | partition %s - try to reassign",
-                consumer.memberid(),
+                member_id,
                 topic_partition.topic,
                 topic_partition.partition,
             )
 
+    def _get_memberid(self) -> str | None:
+        member_id = None
+        try:
+            member_id = self._consumer.memberid()
+        except RuntimeError as error:
+            logger.error("Failed to retrieve member ID: %s", error)
+        return member_id
+
     def shut_down(self) -> None:
         """Close consumer, which also commits kafka offsets."""
         self._consumer.close()

logprep/connector/confluent_kafka/output.py

@@ -232,7 +232,7 @@ def _error_callback(self, error: KafkaException):
             the error that occurred
         """
         self.metrics.number_of_errors += 1
-        logger.error(f"{self.describe()}: {error}")  # pylint: disable=logging-fstring-interpolation
+        logger.error("%s: %s", self.describe(), error)
 
     def _stats_callback(self, stats_raw: str) -> None:
         """Callback for statistics data. This callback is triggered by poll()

tests/unit/connector/test_confluent_kafka_common.py

@@ -34,7 +34,7 @@ def test_error_callback_logs_error(self):
             test_error = Exception("test error")
             self.object._error_callback(test_error)
             mock_error.assert_called()
-            mock_error.assert_called_with(f"{self.object.describe()}: {test_error}")
+            mock_error.assert_called_with("%s: %s", self.object.describe(), test_error)
         assert self.object.metrics.number_of_errors == 1
 
     def test_stats_callback_sets_metric_objetc_attributes(self):

tests/unit/connector/test_confluent_kafka_input.py

@@ -4,12 +4,13 @@
 # pylint: disable=wrong-import-order
 # pylint: disable=attribute-defined-outside-init
 import os
+import re
 import socket
 from copy import deepcopy
 from unittest import mock
 
 import pytest
-from confluent_kafka import OFFSET_BEGINNING, KafkaError, KafkaException
+from confluent_kafka import OFFSET_BEGINNING, KafkaError, KafkaException  # type: ignore
 
 from logprep.abc.input import (
     CriticalInputError,
@@ -345,7 +346,7 @@ def test_lost_callback_counts_warnings_and_logs(self, mock_consumer):
         self.object.metrics.number_of_warnings = 0
         mock_partitions = [mock.MagicMock()]
         with mock.patch("logging.Logger.warning") as mock_warning:
-            self.object._lost_callback(mock_consumer, mock_partitions)
+            self.object._lost_callback(mock_partitions)
         mock_warning.assert_called()
         assert self.object.metrics.number_of_warnings == 1
 
@@ -360,39 +361,44 @@ def test_commit_callback_sets_offset_to_0_for_special_offsets(self, _):
         }
         self.object.metrics.committed_offsets.add_with_labels.assert_called_with(0, expected_labels)
 
-    @mock.patch("logprep.connector.confluent_kafka.input.Consumer")
-    def test_assign_callback_sets_offsets_and_logs_info(self, mock_consumer):
+    def test_assign_callback_sets_offsets_and_logs_info(self):
         self.object.metrics.committed_offsets.add_with_labels = mock.MagicMock()
         self.object.metrics.current_offsets.add_with_labels = mock.MagicMock()
         mock_partitions = [mock.MagicMock()]
         mock_partitions[0].offset = OFFSET_BEGINNING
         with mock.patch("logging.Logger.info") as mock_info:
-            self.object._assign_callback(mock_consumer, mock_partitions)
+            self.object._assign_callback(mock_partitions)
         expected_labels = {
             "description": f"topic: test_input_raw - partition: {mock_partitions[0].partition}"
         }
         mock_info.assert_called()
         self.object.metrics.committed_offsets.add_with_labels.assert_called_with(0, expected_labels)
         self.object.metrics.current_offsets.add_with_labels.assert_called_with(0, expected_labels)
 
-    @mock.patch("logprep.connector.confluent_kafka.input.Consumer")
-    def test_revoke_callback_logs_warning_and_counts(self, mock_consumer):
+    def test_revoke_callback_logs_warning_and_counts(self):
         self.object.metrics.number_of_warnings = 0
         self.object.output_connector = mock.MagicMock()
         mock_partitions = [mock.MagicMock()]
         with mock.patch("logging.Logger.warning") as mock_warning:
-            self.object._revoke_callback(mock_consumer, mock_partitions)
+            self.object._revoke_callback(mock_partitions)
         mock_warning.assert_called()
         assert self.object.metrics.number_of_warnings == 1
 
-    @mock.patch("logprep.connector.confluent_kafka.input.Consumer")
-    def test_revoke_callback_calls_batch_finished_callback(self, mock_consumer):
+    def test_revoke_callback_calls_batch_finished_callback(self):
         self.object.output_connector = mock.MagicMock()
         self.object.batch_finished_callback = mock.MagicMock()
         mock_partitions = [mock.MagicMock()]
-        self.object._revoke_callback(mock_consumer, mock_partitions)
+        self.object._revoke_callback(mock_partitions)
         self.object.batch_finished_callback.assert_called()
 
+    def test_revoke_callback_logs_error_if_consumer_closed(self, caplog):
+        with mock.patch.object(self.object, "_consumer") as mock_consumer:
+            mock_consumer.memberid = mock.MagicMock()
+            mock_consumer.memberid.side_effect = RuntimeError("Consumer is closed")
+            mock_partitions = [mock.MagicMock()]
+            self.object._revoke_callback(mock_partitions)
+            assert re.search(r"ERROR.*Consumer is closed", caplog.text)
+
     def test_health_returns_true_if_no_error(self):
         self.object._admin = mock.MagicMock()
         metadata = mock.MagicMock()