fix(kafka-connector): destroy connector on reset (#615)

fix #614
bakdata · Mar 6, 2025 · 477b6c6 · 477b6c6
1 parent 0b3d80d
commit 477b6c6
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 3 deletions.
diff --git a/kpops/components/base_components/kafka_connector.py b/kpops/components/base_components/kafka_connector.py
@@ -231,6 +231,7 @@ def apply_from_inputs(self, name: str, topic: FromTopic) -> NoReturn:
     @override
     async def reset(self, dry_run: bool) -> None:
         """Reset state. Keep connector."""
+        await super().reset(dry_run)
         await self._resetter.reset(dry_run)
 
     @override
@@ -265,6 +266,7 @@ def set_error_topic(self, topic: KafkaTopic) -> None:
     @override
     async def reset(self, dry_run: bool) -> None:
         """Reset state. Keep consumer group and connector."""
+        await super().reset(dry_run)
         self._resetter.values.config.delete_consumer_group = False
         await self._resetter.reset(dry_run)
 

diff --git a/tests/components/test_kafka_sink_connector.py b/tests/components/test_kafka_sink_connector.py
@@ -206,10 +206,15 @@ async def test_reset_when_dry_run_is_true(
         self,
         connector: KafkaSinkConnector,
         dry_run_handler_mock: MagicMock,
+        mocker: MockerFixture,
     ):
+        mock_destroy = mocker.patch.object(connector, "destroy")
+        mock_resetter_reset = mocker.spy(connector._resetter, "reset")
         dry_run = True
         await connector.reset(dry_run=dry_run)
 
+        mock_destroy.assert_called_once_with(dry_run)
+        mock_resetter_reset.assert_called_once_with(dry_run)
         dry_run_handler_mock.print_helm_diff.assert_called_once()
 
     async def test_reset_when_dry_run_is_false(
@@ -236,11 +241,11 @@ async def test_reset_when_dry_run_is_false(
         dry_run = False
 
         await connector.reset(dry_run=dry_run)
-        mock_destroy.assert_not_called()
         mock_resetter_reset.assert_called_once_with(dry_run)
 
         mock.assert_has_calls(
             [
+                mocker.call.destroy_connector(dry_run),
                 mocker.call.helm.add_repo(
                     "bakdata-kafka-connect-resetter",
                     "https://bakdata.github.io/kafka-connect-resetter/",

diff --git a/tests/components/test_kafka_source_connector.py b/tests/components/test_kafka_source_connector.py
@@ -138,9 +138,15 @@ async def test_reset_when_dry_run_is_true(
         self,
         connector: KafkaSourceConnector,
         dry_run_handler_mock: MagicMock,
+        mocker: MockerFixture,
     ):
-        await connector.reset(dry_run=True)
+        mock_destroy = mocker.patch.object(connector, "destroy")
+        mock_resetter_reset = mocker.spy(connector._resetter, "reset")
+        dry_run = True
+        await connector.reset(dry_run=dry_run)
 
+        mock_destroy.assert_called_once_with(dry_run)
+        mock_resetter_reset.assert_called_once_with(dry_run)
         dry_run_handler_mock.print_helm_diff.assert_called_once()
 
     async def test_reset_when_dry_run_is_false(
@@ -166,9 +172,9 @@ async def test_reset_when_dry_run_is_false(
 
         dry_run = False
         await connector.reset(dry_run)
-        mock_destroy.assert_not_called()
 
         assert mock.mock_calls == [
+            mocker.call.destroy_connector(dry_run),
             mocker.call.helm.add_repo(
                 "bakdata-kafka-connect-resetter",
                 "https://bakdata.github.io/kafka-connect-resetter/",