Fix kafka input (#811)

* fixed assign callback
* added changelog entry
* fix revoke callback

---------

Co-authored-by: kmeinerz1 <kaithomas.meinerz@bwi.de>
Co-authored-by: Jörg Zimmermann <joerg.zimmermann@bwi.de>

Author: 3 people

CHANGELOG.md

@@ -13,6 +13,7 @@
 - Fixed endless loading in logprep test config
 - prevent the auto rule tester from loading rules directly defined inside the config, since they break the auto rule tester and can't have tests anyways
 - Fixed typo and broken link in documentation
+- Fixed assign_callback error in confluentkafka input
 
 ## 16.1.0
 ### Deprecations

logprep/connector/confluent_kafka/input.py

@@ -33,7 +33,7 @@
 from functools import cached_property, partial
 from socket import getfqdn
 from types import MappingProxyType
-from typing import Optional, Tuple, Union
+from typing import Any, Optional, Tuple, Union
 
 import msgspec
 from attrs import define, field, validators
@@ -500,7 +500,7 @@ 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, topic_partitions: list[TopicPartition]) -> None:
+    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()
@@ -517,7 +517,7 @@ def _assign_callback(self, topic_partitions: list[TopicPartition]) -> None:
             self.metrics.committed_offsets.add_with_labels(offset, labels)
             self.metrics.current_offsets.add_with_labels(offset, labels)
 
-    def _revoke_callback(self, topic_partitions: list[TopicPartition]) -> None:
+    def _revoke_callback(self, _: Any, topic_partitions: list[TopicPartition]) -> None:
 
         for topic_partition in topic_partitions:
             self.metrics.number_of_warnings += 1

tests/unit/connector/test_confluent_kafka_input.py

@@ -247,10 +247,10 @@ def test_commit_callback_counts_commit_success(self):
     def test_commit_callback_sets_committed_offsets(self):
         mock_add = mock.MagicMock()
         self.object.metrics.committed_offsets.add_with_labels = mock_add
-        topic_partion = mock.MagicMock()
-        topic_partion.partition = 99
-        topic_partion.offset = 666
-        self.object._commit_callback(None, [topic_partion])
+        topic_partition = mock.MagicMock()
+        topic_partition.partition = 99
+        topic_partition.offset = 666
+        self.object._commit_callback(None, [topic_partition])
         call_args = 666, {"description": "topic: test_input_raw - partition: 99"}
         mock_add.assert_called_with(*call_args)
 
@@ -361,13 +361,14 @@ 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)
 
-    def test_assign_callback_sets_offsets_and_logs_info(self):
+    @mock.patch("logprep.connector.confluent_kafka.input.Consumer")
+    def test_assign_callback_sets_offsets_and_logs_info(self, mock_consumer):
         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_partitions)
+            self.object._assign_callback(mock_consumer, mock_partitions)
         expected_labels = {
             "description": f"topic: test_input_raw - partition: {mock_partitions[0].partition}"
         }
@@ -380,23 +381,23 @@ def test_revoke_callback_logs_warning_and_counts(self):
         self.object.output_connector = mock.MagicMock()
         mock_partitions = [mock.MagicMock()]
         with mock.patch("logging.Logger.warning") as mock_warning:
-            self.object._revoke_callback(mock_partitions)
+            self.object._revoke_callback(None, mock_partitions)
         mock_warning.assert_called()
         assert self.object.metrics.number_of_warnings == 1
 
     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_partitions)
+        self.object._revoke_callback(None, 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)
+            self.object._revoke_callback(None, mock_partitions)
             assert re.search(r"ERROR.*Consumer is closed", caplog.text)
 
     def test_health_returns_true_if_no_error(self):