From 72c94edb5f460f88655b3a4642523588223c0bd8 Mon Sep 17 00:00:00 2001 From: Salomon Popp Date: Thu, 9 Apr 2026 16:37:44 +0200 Subject: [PATCH 1/8] feat(kafka-connect): reset connector using REST API, remove custom resetter --- ...aults_pipeline_component_dependencies.yaml | 1 - .../dependencies/kpops_structure.yaml | 6 - .../pipeline_component_dependencies.yaml | 3 - .../pipeline-components/kafka-connector.yaml | 4 - .../kafka-sink-connector.yaml | 4 - .../kafka-source-connector.yaml | 4 - .../pipeline-components/pipeline.yaml | 8 - .../defaults-kafka-connector.yaml | 4 - .../resources/pipeline-defaults/defaults.yaml | 4 - docs/docs/schema/defaults.json | 51 ----- docs/docs/schema/pipeline.json | 34 ---- .../kafka_connect/connect_wrapper.py | 16 ++ .../kafka_connect/kafka_connect_handler.py | 18 ++ .../base_components/kafka_connector.py | 155 +------------- tests/components/test_kafka_connector.py | 3 - tests/components/test_kafka_sink_connector.py | 191 +----------------- .../components/test_kafka_source_connector.py | 159 +-------------- tests/pipeline/test_generate.py | 26 --- 18 files changed, 54 insertions(+), 637 deletions(-) diff --git a/docs/docs/resources/pipeline-components/dependencies/defaults_pipeline_component_dependencies.yaml b/docs/docs/resources/pipeline-components/dependencies/defaults_pipeline_component_dependencies.yaml index 0e3be65f8..f27b30fb8 100644 --- a/docs/docs/resources/pipeline-components/dependencies/defaults_pipeline_component_dependencies.yaml +++ b/docs/docs/resources/pipeline-components/dependencies/defaults_pipeline_component_dependencies.yaml @@ -13,7 +13,6 @@ kafka-connector.yaml: - from_.yaml - to.yaml - config-kafka-connector.yaml -- resetter_values.yaml kafka-sink-connector.yaml: [] kafka-source-connector.yaml: - from_-kafka-source-connector.yaml diff --git a/docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml b/docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml index 5f922be36..e08aa777b 100644 --- a/docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml +++ b/docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml @@ -35,8 +35,6 @@ kpops_components_fields: - to - config - state - - resetter_namespace - - resetter_values kafka-sink-connector: - name - enabled @@ -45,8 +43,6 @@ kpops_components_fields: - to - config - state - - resetter_namespace - - resetter_values kafka-source-connector: - name - enabled @@ -55,8 +51,6 @@ kpops_components_fields: - to - config - state - - resetter_namespace - - resetter_values - offset_topic kubernetes-app: - name diff --git a/docs/docs/resources/pipeline-components/dependencies/pipeline_component_dependencies.yaml b/docs/docs/resources/pipeline-components/dependencies/pipeline_component_dependencies.yaml index c517d87e3..a15e7b3bc 100644 --- a/docs/docs/resources/pipeline-components/dependencies/pipeline_component_dependencies.yaml +++ b/docs/docs/resources/pipeline-components/dependencies/pipeline_component_dependencies.yaml @@ -23,19 +23,16 @@ kafka-connector.yaml: - from_.yaml - to.yaml - config-kafka-connector.yaml -- resetter_values.yaml kafka-sink-connector.yaml: - prefix.yaml - from_.yaml - to.yaml - config-kafka-connector.yaml -- resetter_values.yaml kafka-source-connector.yaml: - prefix.yaml - from_-kafka-source-connector.yaml - to.yaml - config-kafka-connector.yaml -- resetter_values.yaml - offset_topic-kafka-source-connector.yaml kubernetes-app.yaml: - prefix.yaml diff --git a/docs/docs/resources/pipeline-components/kafka-connector.yaml b/docs/docs/resources/pipeline-components/kafka-connector.yaml index 4a64cdf3a..dc809d45b 100644 --- a/docs/docs/resources/pipeline-components/kafka-connector.yaml +++ b/docs/docs/resources/pipeline-components/kafka-connector.yaml @@ -45,7 +45,3 @@ # Full documentation on connectors: https://kafka.apache.org/documentation/#connectconfigs config: # required tasks.max: 1 - # Overriding Kafka Connect Resetter Helm values. E.g. to override the - # Image Tag etc. - resetter_values: - imageTag: "1.2.3" diff --git a/docs/docs/resources/pipeline-components/kafka-sink-connector.yaml b/docs/docs/resources/pipeline-components/kafka-sink-connector.yaml index 9253db0e0..dae7ef1e4 100644 --- a/docs/docs/resources/pipeline-components/kafka-sink-connector.yaml +++ b/docs/docs/resources/pipeline-components/kafka-sink-connector.yaml @@ -46,7 +46,3 @@ # Full documentation on connectors: https://kafka.apache.org/documentation/#connectconfigs config: # required tasks.max: 1 - # Overriding Kafka Connect Resetter Helm values. E.g. to override the - # Image Tag etc. - resetter_values: - imageTag: "1.2.3" diff --git a/docs/docs/resources/pipeline-components/kafka-source-connector.yaml b/docs/docs/resources/pipeline-components/kafka-source-connector.yaml index 8fa6c51d5..e35ad39db 100644 --- a/docs/docs/resources/pipeline-components/kafka-source-connector.yaml +++ b/docs/docs/resources/pipeline-components/kafka-source-connector.yaml @@ -27,10 +27,6 @@ # Full documentation on connectors: https://kafka.apache.org/documentation/#connectconfigs config: # required tasks.max: 1 - # Overriding Kafka Connect Resetter Helm values. E.g. to override the - # Image Tag etc. - resetter_values: - imageTag: "1.2.3" # offset.storage.topic # https://kafka.apache.org/documentation/#connect_running offset_topic: offset_topic diff --git a/docs/docs/resources/pipeline-components/pipeline.yaml b/docs/docs/resources/pipeline-components/pipeline.yaml index b6ad9a726..566cfae61 100644 --- a/docs/docs/resources/pipeline-components/pipeline.yaml +++ b/docs/docs/resources/pipeline-components/pipeline.yaml @@ -248,10 +248,6 @@ # Full documentation on connectors: https://kafka.apache.org/documentation/#connectconfigs config: # required tasks.max: 1 - # Overriding Kafka Connect Resetter Helm values. E.g. to override the - # Image Tag etc. - resetter_values: - imageTag: "1.2.3" # Kafka source connector - type: kafka-source-connector # required name: kafka-source-connector # required @@ -281,10 +277,6 @@ # Full documentation on connectors: https://kafka.apache.org/documentation/#connectconfigs config: # required tasks.max: 1 - # Overriding Kafka Connect Resetter Helm values. E.g. to override the - # Image Tag etc. - resetter_values: - imageTag: "1.2.3" # offset.storage.topic # https://kafka.apache.org/documentation/#connect_running offset_topic: offset_topic diff --git a/docs/docs/resources/pipeline-defaults/defaults-kafka-connector.yaml b/docs/docs/resources/pipeline-defaults/defaults-kafka-connector.yaml index 9ace5173b..dc06b447d 100644 --- a/docs/docs/resources/pipeline-defaults/defaults-kafka-connector.yaml +++ b/docs/docs/resources/pipeline-defaults/defaults-kafka-connector.yaml @@ -48,7 +48,3 @@ kafka-connector: # Full documentation on connectors: https://kafka.apache.org/documentation/#connectconfigs config: # required tasks.max: 1 - # Overriding Kafka Connect Resetter Helm values. E.g. to override the - # Image Tag etc. - resetter_values: - imageTag: "1.2.3" diff --git a/docs/docs/resources/pipeline-defaults/defaults.yaml b/docs/docs/resources/pipeline-defaults/defaults.yaml index e1e16c56c..a16a0e880 100644 --- a/docs/docs/resources/pipeline-defaults/defaults.yaml +++ b/docs/docs/resources/pipeline-defaults/defaults.yaml @@ -172,10 +172,6 @@ kafka-connector: # Full documentation on connectors: https://kafka.apache.org/documentation/#connectconfigs config: # required tasks.max: 1 - # Overriding Kafka Connect Resetter Helm values. E.g. to override the - # Image Tag etc. - resetter_values: - imageTag: "1.2.3" # Kafka sink connector # # Child of: KafkaConnector diff --git a/docs/docs/schema/defaults.json b/docs/docs/schema/defaults.json index e18ff92d6..27f3b09f1 100644 --- a/docs/docs/schema/defaults.json +++ b/docs/docs/schema/defaults.json @@ -1371,23 +1371,6 @@ "title": "Prefix", "type": "string" }, - "resetter_namespace": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Kubernetes namespace in which the Kafka Connect resetter shall be deployed", - "title": "Resetter Namespace" - }, - "resetter_values": { - "$ref": "#/$defs/HelmAppValues", - "description": "Overriding Kafka Connect resetter Helm values, e.g. to override the image tag etc." - }, "state": { "anyOf": [ { @@ -1522,23 +1505,6 @@ "title": "Prefix", "type": "string" }, - "resetter_namespace": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Kubernetes namespace in which the Kafka Connect resetter shall be deployed", - "title": "Resetter Namespace" - }, - "resetter_values": { - "$ref": "#/$defs/HelmAppValues", - "description": "Overriding Kafka Connect resetter Helm values, e.g. to override the image tag etc." - }, "state": { "anyOf": [ { @@ -1628,23 +1594,6 @@ "title": "Prefix", "type": "string" }, - "resetter_namespace": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Kubernetes namespace in which the Kafka Connect resetter shall be deployed", - "title": "Resetter Namespace" - }, - "resetter_values": { - "$ref": "#/$defs/HelmAppValues", - "description": "Overriding Kafka Connect resetter Helm values, e.g. to override the image tag etc." - }, "state": { "anyOf": [ { diff --git a/docs/docs/schema/pipeline.json b/docs/docs/schema/pipeline.json index 779d3131c..0b3373880 100644 --- a/docs/docs/schema/pipeline.json +++ b/docs/docs/schema/pipeline.json @@ -1327,23 +1327,6 @@ "title": "Prefix", "type": "string" }, - "resetter_namespace": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Kubernetes namespace in which the Kafka Connect resetter shall be deployed", - "title": "Resetter Namespace" - }, - "resetter_values": { - "$ref": "#/$defs/HelmAppValues", - "description": "Overriding Kafka Connect resetter Helm values, e.g. to override the image tag etc." - }, "state": { "anyOf": [ { @@ -1433,23 +1416,6 @@ "title": "Prefix", "type": "string" }, - "resetter_namespace": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "default": null, - "description": "Kubernetes namespace in which the Kafka Connect resetter shall be deployed", - "title": "Resetter Namespace" - }, - "resetter_values": { - "$ref": "#/$defs/HelmAppValues", - "description": "Overriding Kafka Connect resetter Helm values, e.g. to override the image tag etc." - }, "state": { "anyOf": [ { diff --git a/kpops/component_handlers/kafka_connect/connect_wrapper.py b/kpops/component_handlers/kafka_connect/connect_wrapper.py index 97755b37b..794dd31ee 100644 --- a/kpops/component_handlers/kafka_connect/connect_wrapper.py +++ b/kpops/component_handlers/kafka_connect/connect_wrapper.py @@ -232,3 +232,19 @@ async def delete_connector(self, connector_name: str) -> None: await asyncio.sleep(1) return await self.delete_connector(connector_name) raise KafkaConnectError(response) + + async def reset_offset(self, connector_name: str) -> None: + """Reset the offsets for a connector; the connector must exist, and must be in the STOPPED state. + + API Reference: + https://docs.confluent.io/platform/current/connect/references/restapi.html#delete--connectors-connector-offsets + :param connector_name: Configuration parameters for the connector. + :raises ConnectorNotFoundException: Connector not found + """ + response = await self._client.delete(f"/connectors/{connector_name}/offsets") + if response.status_code == httpx.codes.OK: + log.info(f"Connector {connector_name} offsets reset.") + return + if response.status_code == httpx.codes.NOT_FOUND: + raise ConnectorNotFoundException + raise KafkaConnectError(response) diff --git a/kpops/component_handlers/kafka_connect/kafka_connect_handler.py b/kpops/component_handlers/kafka_connect/kafka_connect_handler.py index de1d1a22f..71fda5781 100644 --- a/kpops/component_handlers/kafka_connect/kafka_connect_handler.py +++ b/kpops/component_handlers/kafka_connect/kafka_connect_handler.py @@ -87,6 +87,24 @@ async def destroy_connector(self, connector_name: str, *, dry_run: bool) -> None f"Connector Destruction: the connector {connector_name} does not exist. Skipping." ) + async def reset_connector(self, connector_name: str, *, dry_run: bool) -> None: + """Reset connector offsets. + + :param connector_name: The connector name. + :param dry_run: Whether the connector reset should be run in dry run mode. + """ + if dry_run: + pass # TODO + else: + try: + await self._connect_wrapper.stop_connector(connector_name) + await self._connect_wrapper.reset_offset(connector_name) + + except ConnectorNotFoundException: + log.warning( + f"Connector reset: the connector {connector_name} does not exist. Skipping." + ) + async def __dry_run_connector_creation( self, connector_config: KafkaConnectorConfig, diff --git a/kpops/components/base_components/kafka_connector.py b/kpops/components/base_components/kafka_connector.py index fd68ffae2..7f9b1ccc0 100644 --- a/kpops/components/base_components/kafka_connector.py +++ b/kpops/components/base_components/kafka_connector.py @@ -2,101 +2,24 @@ import logging from abc import ABC -from functools import cached_property -from typing import Any, Literal, NoReturn, Self +from typing import Any, NoReturn -import pydantic -from pydantic import Field, PrivateAttr, ValidationInfo, field_validator +from pydantic import PrivateAttr, ValidationInfo, field_validator from typing_extensions import override from kpops.component_handlers import get_handlers -from kpops.component_handlers.helm_wrapper.model import ( - HelmRepoConfig, -) from kpops.component_handlers.kafka_connect.model import ( ConnectorNewState, KafkaConnectorConfig, KafkaConnectorType, ) -from kpops.components.base_components.cleaner import Cleaner -from kpops.components.base_components.helm_app import HelmAppValues from kpops.components.base_components.models.from_section import FromTopic from kpops.components.base_components.pipeline_component import PipelineComponent from kpops.components.common.topic import KafkaTopic -from kpops.config import get_config -from kpops.utils.colorify import magentaify -from kpops.utils.pydantic import CamelCaseConfigModel, SkipGenerate log = logging.getLogger("KafkaConnector") -class KafkaConnectorResetterConfig(CamelCaseConfigModel): - brokers: str - connector: str - delete_consumer_group: bool | None = None - offset_topic: str | None = None - - -class KafkaConnectorResetterValues(HelmAppValues): - connector_type: Literal["source", "sink"] - config: KafkaConnectorResetterConfig - - -class KafkaConnectorResetter(Cleaner, ABC): - """Helm app for resetting and cleaning a Kafka Connector. - - :param repo_config: Configuration of the Helm chart repo to be used for - deploying the component, defaults to kafka-connect-resetter Helm repo - :param version: Helm chart version, defaults to "1.0.4" - """ - - from_: None = None # pyright: ignore[reportIncompatibleVariableOverride] - to: None = None # pyright: ignore[reportIncompatibleVariableOverride] - values: KafkaConnectorResetterValues # pyright: ignore[reportIncompatibleVariableOverride] - repo_config: SkipGenerate[HelmRepoConfig] = HelmRepoConfig( # pyright: ignore[reportIncompatibleVariableOverride] - repository_name="bakdata-kafka-connect-resetter", - url="https://bakdata.github.io/kafka-connect-resetter/", - ) - version: str | None = "1.0.4" - - @property - @override - def helm_chart(self) -> str: - return f"{self.repo_config.repository_name}/kafka-connect-resetter" - - @override - async def reset(self, dry_run: bool) -> None: - """Reset connector. - - At first, it deletes the previous cleanup job (connector resetter) - to make sure that there is no running clean job in the cluster. Then it releases a cleanup job. - If retain_clean_jobs config is set to false the cleanup job will be deleted subsequently. - - :param dry_run: If the cleanup should be run in dry run mode or not - """ - log.info( - magentaify( - f"Connector Cleanup: uninstalling cleanup job Helm release from previous runs for {self.values.config.connector}" - ) - ) - await self.destroy(dry_run) - - log.info( - magentaify( - f"Connector Cleanup: deploy Connect {self.values.connector_type} resetter for {self.values.config.connector}" - ) - ) - await self.deploy(dry_run) - - if not get_config().retain_clean_jobs: - log.info(magentaify("Connector Cleanup: uninstall Kafka Resetter.")) - await self.destroy(dry_run) - - @override - async def clean(self, dry_run: bool) -> None: - await self.reset(dry_run) - - class KafkaConnector(PipelineComponent, ABC): """Base class for all Kafka connectors. @@ -104,15 +27,10 @@ class KafkaConnector(PipelineComponent, ABC): :param config: Connector config :param state: Connector state - :param resetter_namespace: Kubernetes namespace in which the Kafka Connect resetter shall be deployed - :param resetter_values: Overriding Kafka Connect resetter Helm values, e.g. to override the image tag etc., - defaults to empty HelmAppValues """ config: KafkaConnectorConfig state: ConnectorNewState | None = None - resetter_namespace: str | None = None - resetter_values: HelmAppValues = Field(default_factory=HelmAppValues) _connector_type: KafkaConnectorType = PrivateAttr() @field_validator("config", mode="before") @@ -132,35 +50,6 @@ def connector_config_should_have_component_name( config["name"] = component_name return KafkaConnectorConfig.model_validate(config) - @cached_property - def _resetter(self) -> KafkaConnectorResetter: - kwargs: dict[str, Any] = {} - if self.resetter_namespace: - kwargs["namespace"] = self.resetter_namespace - return KafkaConnectorResetter( - **kwargs, - **self.model_dump( - by_alias=True, - exclude={ - "_resetter", - "resetter_values", - "resetter_namespace", - "values", - "config", - "from_", - "to", - }, - ), - values=KafkaConnectorResetterValues( - connector_type=self._connector_type.value, - config=KafkaConnectorResetterConfig( - connector=self.full_name, - brokers=get_config().kafka_brokers, - ), - **self.resetter_values.model_dump(), - ), - ) - @override async def deploy(self, dry_run: bool) -> None: """Deploy Kafka Connector (Source/Sink). Create output topics and register schemas if configured.""" @@ -182,9 +71,17 @@ async def destroy(self, dry_run: bool) -> None: self.full_name, dry_run=dry_run ) + @override + async def reset(self, dry_run: bool) -> None: + """Reset connector offsets without deleting the connector.""" + await get_handlers().connector_handler.reset_connector( + self.full_name, dry_run=dry_run + ) + @override async def clean(self, dry_run: bool) -> None: """Delete Kafka Connector. If schema handler is enabled, then remove schemas. Delete all the output topics.""" + await self.reset(dry_run) await super().clean(dry_run) if self.to: if schema_handler := get_handlers().schema_handler: @@ -205,29 +102,11 @@ class KafkaSourceConnector(KafkaConnector): _connector_type: KafkaConnectorType = PrivateAttr(KafkaConnectorType.SOURCE) - @pydantic.model_validator(mode="after") - def populate_offset_topic(self) -> Self: - if self.offset_topic: - self._resetter.values.config.offset_topic = self.offset_topic - return self - @override def apply_from_inputs(self, name: str, topic: FromTopic) -> NoReturn: msg = "Kafka source connector doesn't support FromSection" raise NotImplementedError(msg) - @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 - async def clean(self, dry_run: bool) -> None: - """Delete connector and reset state.""" - await super().clean(dry_run) - await self._resetter.clean(dry_run) - class KafkaSinkConnector(KafkaConnector): """Kafka sink connector model.""" @@ -250,17 +129,3 @@ def set_input_pattern(self, name: str) -> None: @override def set_error_topic(self, topic: KafkaTopic) -> None: self.config.errors_deadletterqueue_topic_name = topic - - @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) - - @override - async def clean(self, dry_run: bool) -> None: - """Delete connector and consumer group.""" - await super().clean(dry_run) - self._resetter.values.config.delete_consumer_group = True - await self._resetter.clean(dry_run) diff --git a/tests/components/test_kafka_connector.py b/tests/components/test_kafka_connector.py index 237b18461..9e873ddd1 100644 --- a/tests/components/test_kafka_connector.py +++ b/tests/components/test_kafka_connector.py @@ -19,7 +19,6 @@ "${pipeline.name}-" + "test-connector-with-long-612f3-clean" ) CONNECTOR_CLASS = "com.bakdata.connect.TestConnector" -RESETTER_NAMESPACE = "test-namespace" @pytest.mark.usefixtures("mock_env") @@ -49,7 +48,6 @@ def connector(self, connector_config: KafkaConnectorConfig) -> KafkaConnector: return KafkaConnector( # HACK: not supposed to be instantiated, because ABC name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, ) def test_connector_config_name_override(self, connector: KafkaConnector): @@ -58,7 +56,6 @@ def test_connector_config_name_override(self, connector: KafkaConnector): connector = KafkaConnector( name=CONNECTOR_NAME, config={"connector.class": CONNECTOR_CLASS}, # pyright: ignore[reportArgumentType], gets enriched - resetter_namespace=RESETTER_NAMESPACE, ) assert connector.config.name == CONNECTOR_FULL_NAME diff --git a/tests/components/test_kafka_sink_connector.py b/tests/components/test_kafka_sink_connector.py index 4d586c501..2aab3be4b 100644 --- a/tests/components/test_kafka_sink_connector.py +++ b/tests/components/test_kafka_sink_connector.py @@ -1,21 +1,16 @@ -from unittest.mock import ANY, MagicMock, call +from unittest.mock import MagicMock, call import pytest from pytest_mock import MockerFixture from typing_extensions import override from kpops.component_handlers import get_handlers -from kpops.component_handlers.helm_wrapper.model import ( - HelmUpgradeInstallFlags, - RepoAuthFlags, -) from kpops.component_handlers.kafka_connect.model import ( ConnectorNewState, KafkaConnectorConfig, KafkaConnectorType, ) from kpops.components.base_components.kafka_connector import ( - KafkaConnectorResetter, KafkaSinkConnector, ) from kpops.components.base_components.models import TopicName @@ -34,12 +29,8 @@ ) from kpops.utils.colorify import magentaify from tests.components.test_kafka_connector import ( - CONNECTOR_CLEAN_FULL_NAME, - CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - CONNECTOR_CLEAN_RELEASE_NAME, CONNECTOR_FULL_NAME, CONNECTOR_NAME, - RESETTER_NAMESPACE, TestKafkaConnector, ) @@ -57,7 +48,6 @@ def connector(self, connector_config: KafkaConnectorConfig) -> KafkaSinkConnecto return KafkaSinkConnector( name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, to=ToSection( topics={ TopicName("${output_topic_name}"): TopicConfig( @@ -67,41 +57,6 @@ def connector(self, connector_config: KafkaConnectorConfig) -> KafkaSinkConnecto ), ) - def test_resetter(self, connector: KafkaSinkConnector): - resetter = connector._resetter - assert isinstance(resetter, KafkaConnectorResetter) - assert resetter.full_name == CONNECTOR_CLEAN_FULL_NAME - - def test_resetter_release_name(self, connector: KafkaSinkConnector): - assert connector.config.name == CONNECTOR_FULL_NAME - assert connector._resetter.helm_release_name == CONNECTOR_CLEAN_RELEASE_NAME - - def test_resetter_helm_name_override(self, connector: KafkaSinkConnector): - assert ( - connector._resetter.to_helm_values()["nameOverride"] - == CONNECTOR_CLEAN_HELM_NAMEOVERRIDE - ) - assert ( - connector._resetter.to_helm_values()["fullnameOverride"] - == CONNECTOR_CLEAN_HELM_NAMEOVERRIDE - ) - - def test_resetter_inheritance(self, connector: KafkaSinkConnector): - setattr(connector.resetter_values, "testKey", "foo") - resetter = connector._resetter - assert resetter - assert not hasattr(resetter, "_resetter") - - assert not hasattr(resetter, "resetter_namespace") - assert resetter.namespace == connector.resetter_namespace - - assert not hasattr(resetter, "resetter_values") - # check that resetter values are contained in resetter app values - assert ( - connector.resetter_values.model_dump().items() - <= resetter.values.model_dump().items() - ) - def test_connector_config_parsing(self, connector_config: KafkaConnectorConfig): topic_pattern = ".*" connector = KafkaSinkConnector( @@ -112,7 +67,6 @@ def test_connector_config_parsing(self, connector_config: KafkaConnectorConfig): "topics.regex": topic_pattern, } ), - resetter_namespace=RESETTER_NAMESPACE, ) assert connector.config.topics_regex == topic_pattern assert connector.config.model_dump()["topics.regex"] == topic_pattern @@ -125,7 +79,6 @@ def test_from_section_parsing_input_topic( connector = KafkaSinkConnector( name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, from_=FromSection( topics={ topic1: FromTopic(type=InputTopicTypes.INPUT), @@ -160,7 +113,6 @@ def test_from_section_parsing_input_pattern( connector = KafkaSinkConnector( name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, from_=FromSection( topics={topic_pattern: FromTopic(type=InputTopicTypes.PATTERN)} ), @@ -239,13 +191,11 @@ async def test_reset_when_dry_run_is_true( 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() + mock_destroy.assert_not_called() + dry_run_handler_mock.print_helm_diff.assert_not_called() async def test_reset_when_dry_run_is_false( self, @@ -261,7 +211,6 @@ async def test_reset_when_dry_run_is_false( mock_clean_connector = mocker.patch.object( get_handlers().connector_handler, "clean_connector" ) - mock_resetter_reset = mocker.spy(connector._resetter, "reset") mock = mocker.MagicMock() mock.attach_mock(mock_destroy, "destroy_connector") @@ -271,53 +220,8 @@ async def test_reset_when_dry_run_is_false( dry_run = False await connector.reset(dry_run=dry_run) - 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/", - RepoAuthFlags(), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ - mocker.call.helm.upgrade_install( - CONNECTOR_CLEAN_RELEASE_NAME, - "bakdata-kafka-connect-resetter/kafka-connect-resetter", - dry_run, - RESETTER_NAMESPACE, - { - "nameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "fullnameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "connectorType": CONNECTOR_TYPE, - "config": { - "brokers": "broker:9092", - "connector": CONNECTOR_FULL_NAME, - "deleteConsumerGroup": False, - }, - }, - HelmUpgradeInstallFlags( - version="1.0.4", - wait=True, - wait_for_jobs=True, - ), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ - ] - ) + mock.assert_has_calls([]) dry_run_handler_mock.print_helm_diff.assert_not_called() mock_delete_topic.assert_not_called() @@ -379,46 +283,6 @@ async def test_clean_when_dry_run_is_false( mocker.call.mock_delete_topic(topic, dry_run=dry_run) for topic in connector.to.kafka_topics ), - mocker.call.helm.add_repo( - "bakdata-kafka-connect-resetter", - "https://bakdata.github.io/kafka-connect-resetter/", - RepoAuthFlags(), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ - mocker.call.helm.upgrade_install( - CONNECTOR_CLEAN_RELEASE_NAME, - "bakdata-kafka-connect-resetter/kafka-connect-resetter", - dry_run, - RESETTER_NAMESPACE, - { - "nameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "fullnameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "connectorType": CONNECTOR_TYPE, - "config": { - "brokers": "broker:9092", - "connector": CONNECTOR_FULL_NAME, - "deleteConsumerGroup": True, - }, - }, - HelmUpgradeInstallFlags( - version="1.0.4", - wait=True, - wait_for_jobs=True, - ), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ ] dry_run_handler_mock.print_helm_diff.assert_not_called() @@ -430,7 +294,6 @@ async def test_clean_without_to_when_dry_run_is_true( connector = KafkaSinkConnector( name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, ) dry_run = True @@ -448,7 +311,6 @@ async def test_clean_without_to_when_dry_run_is_false( connector = KafkaSinkConnector( name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, ) mock_destroy = mocker.patch.object(connector, "destroy") @@ -470,51 +332,6 @@ async def test_clean_without_to_when_dry_run_is_false( 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/", - RepoAuthFlags( - username=None, - password=None, - ca_file=None, - insecure_skip_tls_verify=False, - ), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ - mocker.call.helm.upgrade_install( - CONNECTOR_CLEAN_RELEASE_NAME, - "bakdata-kafka-connect-resetter/kafka-connect-resetter", - dry_run, - RESETTER_NAMESPACE, - { - "nameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "fullnameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "connectorType": CONNECTOR_TYPE, - "config": { - "brokers": "broker:9092", - "connector": CONNECTOR_FULL_NAME, - "deleteConsumerGroup": True, - }, - }, - HelmUpgradeInstallFlags( - version="1.0.4", - wait=True, - wait_for_jobs=True, - ), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ ] dry_run_handler_mock.print_helm_diff.assert_not_called() diff --git a/tests/components/test_kafka_source_connector.py b/tests/components/test_kafka_source_connector.py index af89bcb8b..243cc24d8 100644 --- a/tests/components/test_kafka_source_connector.py +++ b/tests/components/test_kafka_source_connector.py @@ -1,21 +1,16 @@ -from unittest.mock import ANY, MagicMock +from unittest.mock import MagicMock import pytest from pytest_mock import MockerFixture from typing_extensions import override from kpops.component_handlers import get_handlers -from kpops.component_handlers.helm_wrapper.model import ( - HelmUpgradeInstallFlags, - RepoAuthFlags, -) from kpops.component_handlers.kafka_connect.model import ( ConnectorNewState, KafkaConnectorConfig, KafkaConnectorType, ) from kpops.components.base_components.kafka_connector import ( - KafkaConnectorResetter, KafkaSourceConnector, ) from kpops.components.base_components.models import TopicName @@ -28,13 +23,9 @@ ToSection, ) from kpops.components.common.topic import OutputTopicTypes, TopicConfig -from kpops.utils.environment import ENV from tests.components.test_kafka_connector import ( - CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - CONNECTOR_CLEAN_RELEASE_NAME, CONNECTOR_FULL_NAME, CONNECTOR_NAME, - RESETTER_NAMESPACE, TestKafkaConnector, ) @@ -53,7 +44,6 @@ def connector( return KafkaSourceConnector( name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, to=ToSection( topics={ TopicName("${output_topic_name}"): TopicConfig( @@ -64,15 +54,6 @@ def connector( offset_topic=OFFSETS_TOPIC, ) - def test_resetter_release_name(self, connector: KafkaSourceConnector): - assert connector.config.name == CONNECTOR_FULL_NAME - resetter = connector._resetter - assert isinstance(resetter, KafkaConnectorResetter) - assert connector._resetter.helm_release_name == CONNECTOR_CLEAN_RELEASE_NAME - - def test_resetter_offset_topic(self, connector: KafkaSourceConnector): - assert connector._resetter.values.config.offset_topic == OFFSETS_TOPIC - def test_from_section_raises_exception( self, connector_config: KafkaConnectorConfig, @@ -81,7 +62,6 @@ def test_from_section_raises_exception( KafkaSourceConnector( name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, from_=FromSection( topics={ TopicName("connector-topic"): FromTopic( @@ -147,7 +127,6 @@ async def test_destroy( connector: KafkaSourceConnector, mocker: MockerFixture, ): - ENV["KPOPS_KAFKA_CONNECT_RESETTER_OFFSET_TOPIC"] = OFFSETS_TOPIC assert get_handlers().connector_handler mock_destroy_connector = mocker.patch.object( @@ -167,13 +146,11 @@ async def test_reset_when_dry_run_is_true( 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() + mock_destroy.assert_not_called() + dry_run_handler_mock.print_helm_diff.assert_not_called() async def test_reset_when_dry_run_is_false( self, @@ -199,49 +176,7 @@ async def test_reset_when_dry_run_is_false( dry_run = False await connector.reset(dry_run) - 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/", - RepoAuthFlags(), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ - mocker.call.helm.upgrade_install( - CONNECTOR_CLEAN_RELEASE_NAME, - "bakdata-kafka-connect-resetter/kafka-connect-resetter", - dry_run, - RESETTER_NAMESPACE, - { - "connectorType": CONNECTOR_TYPE, - "config": { - "brokers": "broker:9092", - "connector": CONNECTOR_FULL_NAME, - "offsetTopic": OFFSETS_TOPIC, - }, - "nameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "fullnameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - }, - HelmUpgradeInstallFlags( - version="1.0.4", - wait=True, - wait_for_jobs=True, - ), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ - ] + assert mock.mock_calls == [] mock_delete_topic.assert_not_called() dry_run_handler_mock.print_helm_diff.assert_not_called() @@ -252,7 +187,7 @@ async def test_clean_when_dry_run_is_true( ): await connector.clean(dry_run=True) - dry_run_handler_mock.print_helm_diff.assert_called_once() + dry_run_handler_mock.print_helm_diff.assert_not_called() async def test_clean_when_dry_run_is_false( self, @@ -286,46 +221,6 @@ async def test_clean_when_dry_run_is_false( mocker.call.mock_delete_topic(topic, dry_run=dry_run) for topic in connector.to.kafka_topics ), - mocker.call.helm.add_repo( - "bakdata-kafka-connect-resetter", - "https://bakdata.github.io/kafka-connect-resetter/", - RepoAuthFlags(), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ - mocker.call.helm.upgrade_install( - CONNECTOR_CLEAN_RELEASE_NAME, - "bakdata-kafka-connect-resetter/kafka-connect-resetter", - dry_run, - RESETTER_NAMESPACE, - { - "nameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "fullnameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "connectorType": CONNECTOR_TYPE, - "config": { - "brokers": "broker:9092", - "connector": CONNECTOR_FULL_NAME, - "offsetTopic": OFFSETS_TOPIC, - }, - }, - HelmUpgradeInstallFlags( - version="1.0.4", - wait=True, - wait_for_jobs=True, - ), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ ] dry_run_handler_mock.print_helm_diff.assert_not_called() @@ -340,7 +235,6 @@ async def test_clean_without_to_when_dry_run_is_false( connector = KafkaSourceConnector( name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, offset_topic=OFFSETS_TOPIC, ) assert connector.to is None @@ -367,46 +261,6 @@ async def test_clean_without_to_when_dry_run_is_false( 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/", - RepoAuthFlags(), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ - mocker.call.helm.upgrade_install( - CONNECTOR_CLEAN_RELEASE_NAME, - "bakdata-kafka-connect-resetter/kafka-connect-resetter", - dry_run, - RESETTER_NAMESPACE, - { - "nameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "fullnameOverride": CONNECTOR_CLEAN_HELM_NAMEOVERRIDE, - "connectorType": CONNECTOR_TYPE, - "config": { - "brokers": "broker:9092", - "connector": CONNECTOR_FULL_NAME, - "offsetTopic": OFFSETS_TOPIC, - }, - }, - HelmUpgradeInstallFlags( - version="1.0.4", - wait=True, - wait_for_jobs=True, - ), - ), - mocker.call.helm.uninstall( - RESETTER_NAMESPACE, - CONNECTOR_CLEAN_RELEASE_NAME, - dry_run, - ), - ANY, # __bool__ - ANY, # __str__ ] mock_delete_topic.assert_not_called() @@ -420,7 +274,6 @@ async def test_clean_without_to_when_dry_run_is_true( connector = KafkaSourceConnector( name=CONNECTOR_NAME, config=connector_config, - resetter_namespace=RESETTER_NAMESPACE, offset_topic=OFFSETS_TOPIC, ) assert connector.to is None @@ -429,4 +282,4 @@ async def test_clean_without_to_when_dry_run_is_true( await connector.clean(dry_run=True) - dry_run_handler_mock.print_helm_diff.assert_called_once() + dry_run_handler_mock.print_helm_diff.assert_not_called() diff --git a/tests/pipeline/test_generate.py b/tests/pipeline/test_generate.py index 57d3ddf7a..00c7cd5f1 100644 --- a/tests/pipeline/test_generate.py +++ b/tests/pipeline/test_generate.py @@ -877,32 +877,6 @@ def test_temp_trim_release_name(self): == "in-order-to-have-len-fifty-two-name-should-end--here" ) - def test_substitution_in_inflated_component(self): - pipeline = kpops.generate(RESOURCE_PATH / "resetter_values" / PIPELINE_YAML) - assert isinstance(pipeline.components[1], KafkaSinkConnector) - assert ( - pipeline.components[1]._resetter.values.label == "inflated-connector-name" # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType] - ) - assert ( - pipeline.components[1]._resetter.values.imageTag # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType] - == "override-default-image-tag" - ) - - def test_substitution_in_resetter(self): - pipeline = kpops.generate( - RESOURCE_PATH - / "resetter_values" - / KpopsFileType.PIPELINE.as_yaml_file(suffix="_connector_only"), - ) - assert isinstance(pipeline.components[0], KafkaSinkConnector) - assert pipeline.components[0].name == "es-sink-connector" - assert pipeline.components[0]._resetter.name == "es-sink-connector" - assert pipeline.components[0]._resetter.values.label == "es-sink-connector" # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType] - assert ( - pipeline.components[0]._resetter.values.imageTag # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType] - == "override-default-image-tag" - ) - def test_streams_bootstrap(self, snapshot: Snapshot): pipeline = kpops.generate( RESOURCE_PATH / "streams-bootstrap" / PIPELINE_YAML, From 8e076ba2b3e20bf3f71150025c8dd628299cfab5 Mon Sep 17 00:00:00 2001 From: Salomon Popp Date: Wed, 1 Jul 2026 14:25:36 +0200 Subject: [PATCH 2/8] test(pipeline): update snapshots --- .../test_example/test_generate/atm-fraud/pipeline.yaml | 1 - .../test_example/test_generate/word-count/pipeline.yaml | 3 ++- .../test_generate/test_inflate_pipeline/pipeline.yaml | 1 - .../test_kafka_connect_sink_weave_from_topics/pipeline.yaml | 1 - .../test_generate/test_read_from_component/pipeline.yaml | 2 -- .../test_generate/test_with_env_defaults/pipeline.yaml | 1 - 6 files changed, 2 insertions(+), 7 deletions(-) diff --git a/tests/pipeline/snapshots/test_example/test_generate/atm-fraud/pipeline.yaml b/tests/pipeline/snapshots/test_example/test_generate/atm-fraud/pipeline.yaml index 7acef86c9..67bd46f73 100644 --- a/tests/pipeline/snapshots/test_example/test_generate/atm-fraud/pipeline.yaml +++ b/tests/pipeline/snapshots/test_example/test_generate/atm-fraud/pipeline.yaml @@ -224,4 +224,3 @@ errors.deadletterqueue.topic.replication.factor: '1' errors.tolerance: all pk.mode: record_value - resetter_values: {} diff --git a/tests/pipeline/snapshots/test_example/test_generate/word-count/pipeline.yaml b/tests/pipeline/snapshots/test_example/test_generate/word-count/pipeline.yaml index 056264a29..99436418a 100644 --- a/tests/pipeline/snapshots/test_example/test_generate/word-count/pipeline.yaml +++ b/tests/pipeline/snapshots/test_example/test_generate/word-count/pipeline.yaml @@ -22,6 +22,7 @@ enabled: false replicaCount: 1 version: 3.1.0 + timeout: 1 helm_release_name: word-count-data-producer helm_name_override: word-count-data-producer - type: streams-app @@ -58,6 +59,7 @@ labels: pipeline: word-count version: 3.1.0 + timeout: 1 helm_release_name: word-count-word-count-app helm_name_override: word-count-word-count-app - type: kafka-sink-connector @@ -73,4 +75,3 @@ tasks.max: '1' key.converter: org.apache.kafka.connect.storage.StringConverter value.converter: org.apache.kafka.connect.storage.StringConverter - resetter_values: {} diff --git a/tests/pipeline/snapshots/test_generate/test_inflate_pipeline/pipeline.yaml b/tests/pipeline/snapshots/test_generate/test_inflate_pipeline/pipeline.yaml index 1a9f20984..cb8db80f6 100644 --- a/tests/pipeline/snapshots/test_generate/test_inflate_pipeline/pipeline.yaml +++ b/tests/pipeline/snapshots/test_generate/test_inflate_pipeline/pipeline.yaml @@ -149,7 +149,6 @@ max.buffered.records: '20000' read.timeout.ms: '120000' tasks.max: '1' - resetter_values: {} - type: streams-app name: should-inflate-inflated-streams-app enabled: true diff --git a/tests/pipeline/snapshots/test_generate/test_kafka_connect_sink_weave_from_topics/pipeline.yaml b/tests/pipeline/snapshots/test_generate/test_kafka_connect_sink_weave_from_topics/pipeline.yaml index 2f055caee..fba490286 100644 --- a/tests/pipeline/snapshots/test_generate/test_kafka_connect_sink_weave_from_topics/pipeline.yaml +++ b/tests/pipeline/snapshots/test_generate/test_kafka_connect_sink_weave_from_topics/pipeline.yaml @@ -52,5 +52,4 @@ max.buffered.records: '20000' read.timeout.ms: '120000' tasks.max: '1' - resetter_values: {} diff --git a/tests/pipeline/snapshots/test_generate/test_read_from_component/pipeline.yaml b/tests/pipeline/snapshots/test_generate/test_read_from_component/pipeline.yaml index 71474a107..66a1beac0 100644 --- a/tests/pipeline/snapshots/test_generate/test_read_from_component/pipeline.yaml +++ b/tests/pipeline/snapshots/test_generate/test_read_from_component/pipeline.yaml @@ -107,7 +107,6 @@ max.buffered.records: '20000' read.timeout.ms: '120000' tasks.max: '1' - resetter_values: {} - type: streams-app name: inflate-step-inflated-streams-app enabled: true @@ -209,7 +208,6 @@ max.buffered.records: '20000' read.timeout.ms: '120000' tasks.max: '1' - resetter_values: {} - type: streams-app name: inflate-step-without-prefix-inflated-streams-app enabled: true diff --git a/tests/pipeline/snapshots/test_generate/test_with_env_defaults/pipeline.yaml b/tests/pipeline/snapshots/test_generate/test_with_env_defaults/pipeline.yaml index cf59070dd..3150480d3 100644 --- a/tests/pipeline/snapshots/test_generate/test_with_env_defaults/pipeline.yaml +++ b/tests/pipeline/snapshots/test_generate/test_with_env_defaults/pipeline.yaml @@ -52,5 +52,4 @@ max.buffered.records: '20000' read.timeout.ms: '120000' tasks.max: '1' - resetter_values: {} From b11835ed534c8f9e79a5dee3a4124410b13e7bdd Mon Sep 17 00:00:00 2001 From: Salomon Popp Date: Wed, 1 Jul 2026 18:02:01 +0200 Subject: [PATCH 3/8] test: update --- .../kafka_connect/test_connect_handler.py | 32 +++++++++++ tests/components/test_kafka_sink_connector.py | 57 +++++++------------ .../components/test_kafka_source_connector.py | 46 +++++++-------- 3 files changed, 72 insertions(+), 63 deletions(-) diff --git a/tests/component_handlers/kafka_connect/test_connect_handler.py b/tests/component_handlers/kafka_connect/test_connect_handler.py index f8f0282ed..fe60fe1a8 100644 --- a/tests/component_handlers/kafka_connect/test_connect_handler.py +++ b/tests/component_handlers/kafka_connect/test_connect_handler.py @@ -460,3 +460,35 @@ async def test_print_correct_warning_log_when_destroying_connector_and_connector log_warning_mock.assert_called_once_with( f"Connector Destruction: the connector {CONNECTOR_NAME} does not exist. Skipping." ) + + async def test_reset_connector_dry_run( + self, + connect_wrapper: AsyncMock, + handler: KafkaConnectHandler, + ): + await handler.reset_connector(CONNECTOR_NAME, dry_run=True) + connect_wrapper.stop_connector.assert_not_called() + connect_wrapper.reset_offset.assert_not_called() + + async def test_reset_connector( + self, + connect_wrapper: AsyncMock, + handler: KafkaConnectHandler, + ): + await handler.reset_connector(CONNECTOR_NAME, dry_run=False) + assert connect_wrapper.mock_calls == [ + mock.call.stop_connector(CONNECTOR_NAME), + mock.call.reset_offset(CONNECTOR_NAME), + ] + + async def test_reset_connector_not_found( + self, + connect_wrapper: AsyncMock, + handler: KafkaConnectHandler, + log_warning_mock: MagicMock, + ): + connect_wrapper.stop_connector.side_effect = ConnectorNotFoundException() + await handler.reset_connector(CONNECTOR_NAME, dry_run=False) + log_warning_mock.assert_called_once_with( + f"Connector reset: the connector {CONNECTOR_NAME} does not exist. Skipping." + ) diff --git a/tests/components/test_kafka_sink_connector.py b/tests/components/test_kafka_sink_connector.py index 2aab3be4b..222451b47 100644 --- a/tests/components/test_kafka_sink_connector.py +++ b/tests/components/test_kafka_sink_connector.py @@ -1,4 +1,4 @@ -from unittest.mock import MagicMock, call +from unittest.mock import MagicMock import pytest from pytest_mock import MockerFixture @@ -27,7 +27,6 @@ OutputTopicTypes, TopicConfig, ) -from kpops.utils.colorify import magentaify from tests.components.test_kafka_connector import ( CONNECTOR_FULL_NAME, CONNECTOR_NAME, @@ -191,40 +190,41 @@ async def test_reset_when_dry_run_is_true( mocker: MockerFixture, ): mock_destroy = mocker.patch.object(connector, "destroy") + mock_reset_connector = mocker.patch.object( + get_handlers().connector_handler, "reset_connector" + ) dry_run = True await connector.reset(dry_run=dry_run) mock_destroy.assert_not_called() dry_run_handler_mock.print_helm_diff.assert_not_called() + mock_reset_connector.assert_called_once_with( + CONNECTOR_FULL_NAME, dry_run=dry_run + ) async def test_reset_when_dry_run_is_false( self, connector: KafkaSinkConnector, dry_run_handler_mock: MagicMock, - helm_mock: MagicMock, mocker: MockerFixture, ): mock_destroy = mocker.patch.object(connector, "destroy") mock_delete_topic = mocker.patch.object( get_handlers().topic_handler, "delete_topic" ) - mock_clean_connector = mocker.patch.object( - get_handlers().connector_handler, "clean_connector" + mock_reset_connector = mocker.patch.object( + get_handlers().connector_handler, "reset_connector" ) - mock = mocker.MagicMock() - mock.attach_mock(mock_destroy, "destroy_connector") - mock.attach_mock(mock_clean_connector, "mock_clean_connector") - mock.attach_mock(helm_mock, "helm") - dry_run = False - await connector.reset(dry_run=dry_run) - mock.assert_has_calls([]) - - dry_run_handler_mock.print_helm_diff.assert_not_called() + mock_reset_connector.assert_called_once_with( + CONNECTOR_FULL_NAME, dry_run=dry_run + ) + mock_destroy.assert_not_called() mock_delete_topic.assert_not_called() + dry_run_handler_mock.print_helm_diff.assert_not_called() async def test_clean_when_dry_run_is_true( self, @@ -234,50 +234,33 @@ async def test_clean_when_dry_run_is_true( dry_run = True await connector.clean(dry_run=dry_run) - dry_run_handler_mock.print_helm_diff.assert_called_once() + dry_run_handler_mock.print_helm_diff.assert_not_called() async def test_clean_when_dry_run_is_false( self, connector: KafkaSinkConnector, - helm_mock: MagicMock, - log_info_mock: MagicMock, dry_run_handler_mock: MagicMock, mocker: MockerFixture, ): mock_destroy = mocker.patch.object(connector, "destroy") - mock_delete_topic = mocker.patch.object( get_handlers().topic_handler, "delete_topic" ) - mock_clean_connector = mocker.patch.object( - get_handlers().connector_handler, "clean_connector" + mock_reset_connector = mocker.patch.object( + get_handlers().connector_handler, "reset_connector" ) mock = mocker.MagicMock() + mock.attach_mock(mock_reset_connector, "mock_reset_connector") mock.attach_mock(mock_destroy, "destroy_connector") mock.attach_mock(mock_delete_topic, "mock_delete_topic") - mock.attach_mock(mock_clean_connector, "mock_clean_connector") - mock.attach_mock(helm_mock, "helm") dry_run = False await connector.clean(dry_run=dry_run) - assert log_info_mock.mock_calls == [ - call.log_info( - magentaify( - f"Connector Cleanup: uninstalling cleanup job Helm release from previous runs for {CONNECTOR_FULL_NAME}" - ) - ), - call.log_info( - magentaify( - f"Connector Cleanup: deploy Connect {KafkaConnectorType.SINK.value} resetter for {CONNECTOR_FULL_NAME}" - ) - ), - call.log_info(magentaify("Connector Cleanup: uninstall Kafka Resetter.")), - ] - assert connector.to assert mock.mock_calls == [ + mocker.call.mock_reset_connector(CONNECTOR_FULL_NAME, dry_run=dry_run), mocker.call.destroy_connector(dry_run), *( mocker.call.mock_delete_topic(topic, dry_run=dry_run) @@ -299,7 +282,7 @@ async def test_clean_without_to_when_dry_run_is_true( dry_run = True await connector.clean(dry_run) - dry_run_handler_mock.print_helm_diff.assert_called_once() + dry_run_handler_mock.print_helm_diff.assert_not_called() async def test_clean_without_to_when_dry_run_is_false( self, diff --git a/tests/components/test_kafka_source_connector.py b/tests/components/test_kafka_source_connector.py index 243cc24d8..c15c3f42d 100644 --- a/tests/components/test_kafka_source_connector.py +++ b/tests/components/test_kafka_source_connector.py @@ -146,37 +146,39 @@ async def test_reset_when_dry_run_is_true( mocker: MockerFixture, ): mock_destroy = mocker.patch.object(connector, "destroy") + mock_reset_connector = mocker.patch.object( + get_handlers().connector_handler, "reset_connector" + ) dry_run = True await connector.reset(dry_run=dry_run) mock_destroy.assert_not_called() dry_run_handler_mock.print_helm_diff.assert_not_called() + mock_reset_connector.assert_called_once_with( + CONNECTOR_FULL_NAME, dry_run=dry_run + ) async def test_reset_when_dry_run_is_false( self, connector: KafkaSourceConnector, dry_run_handler_mock: MagicMock, - helm_mock: MagicMock, mocker: MockerFixture, ): mock_destroy = mocker.patch.object(connector, "destroy") - mock_delete_topic = mocker.patch.object( get_handlers().topic_handler, "delete_topic" ) - mock_clean_connector = mocker.spy( - get_handlers().connector_handler, "clean_connector" + mock_reset_connector = mocker.patch.object( + get_handlers().connector_handler, "reset_connector" ) - mock = mocker.MagicMock() - mock.attach_mock(mock_destroy, "destroy_connector") - mock.attach_mock(mock_clean_connector, "mock_clean_connector") - mock.attach_mock(helm_mock, "helm") - dry_run = False await connector.reset(dry_run) - assert mock.mock_calls == [] + mock_reset_connector.assert_called_once_with( + CONNECTOR_FULL_NAME, dry_run=dry_run + ) + mock_destroy.assert_not_called() mock_delete_topic.assert_not_called() dry_run_handler_mock.print_helm_diff.assert_not_called() @@ -192,42 +194,38 @@ async def test_clean_when_dry_run_is_true( async def test_clean_when_dry_run_is_false( self, connector: KafkaSourceConnector, - helm_mock: MagicMock, dry_run_handler_mock: MagicMock, mocker: MockerFixture, ): mock_destroy = mocker.patch.object(connector, "destroy") - mock_delete_topic = mocker.patch.object( get_handlers().topic_handler, "delete_topic" ) - mock_clean_connector = mocker.spy( - get_handlers().connector_handler, "clean_connector" + mock_reset_connector = mocker.patch.object( + get_handlers().connector_handler, "reset_connector" ) mock = mocker.MagicMock() + mock.attach_mock(mock_reset_connector, "mock_reset_connector") mock.attach_mock(mock_destroy, "destroy_connector") mock.attach_mock(mock_delete_topic, "mock_delete_topic") - mock.attach_mock(mock_clean_connector, "mock_clean_connector") - mock.attach_mock(helm_mock, "helm") dry_run = False await connector.clean(dry_run) assert connector.to assert mock.mock_calls == [ + mocker.call.mock_reset_connector(CONNECTOR_FULL_NAME, dry_run=dry_run), mocker.call.destroy_connector(dry_run), *( mocker.call.mock_delete_topic(topic, dry_run=dry_run) for topic in connector.to.kafka_topics ), ] - dry_run_handler_mock.print_helm_diff.assert_not_called() async def test_clean_without_to_when_dry_run_is_false( self, - helm_mock: MagicMock, dry_run_handler_mock: MagicMock, mocker: MockerFixture, connector_config: KafkaConnectorConfig, @@ -239,30 +237,26 @@ async def test_clean_without_to_when_dry_run_is_false( ) assert connector.to is None - assert get_handlers().connector_handler - mock_destroy = mocker.patch.object(connector, "destroy") - mock_delete_topic = mocker.patch.object( get_handlers().topic_handler, "delete_topic" ) - mock_clean_connector = mocker.spy( - get_handlers().connector_handler, "clean_connector" + mock_reset_connector = mocker.patch.object( + get_handlers().connector_handler, "reset_connector" ) mock = mocker.MagicMock() + mock.attach_mock(mock_reset_connector, "mock_reset_connector") mock.attach_mock(mock_destroy, "destroy_connector") mock.attach_mock(mock_delete_topic, "mock_delete_topic") - mock.attach_mock(mock_clean_connector, "mock_clean_connector") - mock.attach_mock(helm_mock, "helm") dry_run = False await connector.clean(dry_run) assert mock.mock_calls == [ + mocker.call.mock_reset_connector(CONNECTOR_FULL_NAME, dry_run=dry_run), mocker.call.destroy_connector(dry_run), ] - mock_delete_topic.assert_not_called() dry_run_handler_mock.print_helm_diff.assert_not_called() From 3c06aa415c78ada5656186dc915bc3f5cf98eba9 Mon Sep 17 00:00:00 2001 From: Salomon Popp Date: Wed, 1 Jul 2026 18:04:13 +0200 Subject: [PATCH 4/8] chore(dprint): ignore changelog --- dprint.jsonc | 1 + 1 file changed, 1 insertion(+) diff --git a/dprint.jsonc b/dprint.jsonc index 6bba58965..1dd951346 100644 --- a/dprint.jsonc +++ b/dprint.jsonc @@ -12,6 +12,7 @@ ".mypy_cache/**", "**/.venv/**", "CHANGELOG.md", + "docs/docs/user/changelog.md", "**/cli-commands.md", "**/cli_env_vars.md", "**/config_env_vars.md", From 46d4527f50d96a1b40931a180de6e472455f5f5b Mon Sep 17 00:00:00 2001 From: Salomon Popp Date: Wed, 1 Jul 2026 18:06:08 +0200 Subject: [PATCH 5/8] test(word-count): update snapshot --- .../test_example/test_generate/word-count/pipeline.yaml | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/pipeline/snapshots/test_example/test_generate/word-count/pipeline.yaml b/tests/pipeline/snapshots/test_example/test_generate/word-count/pipeline.yaml index 99436418a..6628995c0 100644 --- a/tests/pipeline/snapshots/test_example/test_generate/word-count/pipeline.yaml +++ b/tests/pipeline/snapshots/test_example/test_generate/word-count/pipeline.yaml @@ -22,7 +22,6 @@ enabled: false replicaCount: 1 version: 3.1.0 - timeout: 1 helm_release_name: word-count-data-producer helm_name_override: word-count-data-producer - type: streams-app @@ -59,7 +58,6 @@ labels: pipeline: word-count version: 3.1.0 - timeout: 1 helm_release_name: word-count-word-count-app helm_name_override: word-count-word-count-app - type: kafka-sink-connector From a2d4966de0a455c5823ea8df5bfa1b1054c6b5b8 Mon Sep 17 00:00:00 2001 From: Salomon Popp Date: Wed, 1 Jul 2026 18:14:33 +0200 Subject: [PATCH 6/8] feat(kafka-connect): implement dry-run for reset connector --- .../kafka_connect/kafka_connect_handler.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/kpops/component_handlers/kafka_connect/kafka_connect_handler.py b/kpops/component_handlers/kafka_connect/kafka_connect_handler.py index 71fda5781..ede5d7606 100644 --- a/kpops/component_handlers/kafka_connect/kafka_connect_handler.py +++ b/kpops/component_handlers/kafka_connect/kafka_connect_handler.py @@ -94,7 +94,7 @@ async def reset_connector(self, connector_name: str, *, dry_run: bool) -> None: :param dry_run: Whether the connector reset should be run in dry run mode. """ if dry_run: - pass # TODO + await self.__dry_run_connector_reset(connector_name) else: try: await self._connect_wrapper.stop_connector(connector_name) @@ -160,6 +160,23 @@ async def __dry_run_connector_creation( f"Connector Creation: connector config for {connector_name} is valid!" ) + async def __dry_run_connector_reset(self, connector_name: str) -> None: + try: + await self._connect_wrapper.get_connector(connector_name) + log.info( + magentaify( + f"Connector reset: connector {connector_name} exists. Resetting offsets." + ) + ) + log.debug(f"PUT /connectors/{connector_name}/stop HTTP/1.1") + log.debug(f"HOST: {self._connect_wrapper.url}") + log.debug(f"DELETE /connectors/{connector_name}/offsets HTTP/1.1") + log.debug(f"HOST: {self._connect_wrapper.url}") + except ConnectorNotFoundException: + log.warning( + f"Connector reset: the connector {connector_name} does not exist. Skipping." + ) + async def __dry_run_connector_deletion(self, connector_name: str) -> None: try: await self._connect_wrapper.get_connector(connector_name) From 37544219164d72e3d3905acdcf4d72622b17b869 Mon Sep 17 00:00:00 2001 From: Salomon Popp Date: Wed, 1 Jul 2026 18:16:57 +0200 Subject: [PATCH 7/8] test: update --- .../kafka_connect/test_connect_handler.py | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/component_handlers/kafka_connect/test_connect_handler.py b/tests/component_handlers/kafka_connect/test_connect_handler.py index fe60fe1a8..2306f8256 100644 --- a/tests/component_handlers/kafka_connect/test_connect_handler.py +++ b/tests/component_handlers/kafka_connect/test_connect_handler.py @@ -465,10 +465,31 @@ async def test_reset_connector_dry_run( self, connect_wrapper: AsyncMock, handler: KafkaConnectHandler, + log_info_mock: MagicMock, ): await handler.reset_connector(CONNECTOR_NAME, dry_run=True) + connect_wrapper.get_connector.assert_called_once_with(CONNECTOR_NAME) connect_wrapper.stop_connector.assert_not_called() connect_wrapper.reset_offset.assert_not_called() + log_info_mock.assert_called_once_with( + magentaify( + f"Connector reset: connector {CONNECTOR_NAME} exists. Resetting offsets." + ) + ) + + async def test_reset_connector_dry_run_connector_not_found( + self, + connect_wrapper: AsyncMock, + handler: KafkaConnectHandler, + log_warning_mock: MagicMock, + ): + connect_wrapper.get_connector.side_effect = ConnectorNotFoundException() + await handler.reset_connector(CONNECTOR_NAME, dry_run=True) + connect_wrapper.stop_connector.assert_not_called() + connect_wrapper.reset_offset.assert_not_called() + log_warning_mock.assert_called_once_with( + f"Connector reset: the connector {CONNECTOR_NAME} does not exist. Skipping." + ) async def test_reset_connector( self, From e429e8cae451ac08b776d1d51cfa53835c8101df Mon Sep 17 00:00:00 2001 From: Salomon Popp Date: Thu, 2 Jul 2026 09:47:38 +0200 Subject: [PATCH 8/8] test: cleanup leftover resources --- .../sections/resetter_values.yaml | 4 ---- .../resources/resetter_values/defaults.yaml | 13 ------------- .../resources/resetter_values/pipeline.yaml | 1 - .../resetter_values/pipeline_connector_only.yaml | 4 ---- 4 files changed, 22 deletions(-) delete mode 100644 docs/docs/resources/pipeline-components/sections/resetter_values.yaml delete mode 100644 tests/pipeline/resources/resetter_values/defaults.yaml delete mode 100644 tests/pipeline/resources/resetter_values/pipeline.yaml delete mode 100644 tests/pipeline/resources/resetter_values/pipeline_connector_only.yaml diff --git a/docs/docs/resources/pipeline-components/sections/resetter_values.yaml b/docs/docs/resources/pipeline-components/sections/resetter_values.yaml deleted file mode 100644 index 376c472d3..000000000 --- a/docs/docs/resources/pipeline-components/sections/resetter_values.yaml +++ /dev/null @@ -1,4 +0,0 @@ - # Overriding Kafka Connect Resetter Helm values. E.g. to override the - # Image Tag etc. - resetter_values: - imageTag: "1.2.3" diff --git a/tests/pipeline/resources/resetter_values/defaults.yaml b/tests/pipeline/resources/resetter_values/defaults.yaml deleted file mode 100644 index ee97e5e1b..000000000 --- a/tests/pipeline/resources/resetter_values/defaults.yaml +++ /dev/null @@ -1,13 +0,0 @@ -helm-app: - name: "${component.type}" - namespace: "namespace" - values: - label: ${component.name} - kafka: - bootstrapServers: ${config.kafka_brokers} - -kafka-sink-connector: - config: - connector.class: "io.confluent.connect.jdbc.JdbcSinkConnector" - resetter_values: - imageTag: override-default-image-tag diff --git a/tests/pipeline/resources/resetter_values/pipeline.yaml b/tests/pipeline/resources/resetter_values/pipeline.yaml deleted file mode 100644 index ef9d05c76..000000000 --- a/tests/pipeline/resources/resetter_values/pipeline.yaml +++ /dev/null @@ -1 +0,0 @@ -- type: simple-inflate-connectors diff --git a/tests/pipeline/resources/resetter_values/pipeline_connector_only.yaml b/tests/pipeline/resources/resetter_values/pipeline_connector_only.yaml deleted file mode 100644 index af131751d..000000000 --- a/tests/pipeline/resources/resetter_values/pipeline_connector_only.yaml +++ /dev/null @@ -1,4 +0,0 @@ -- type: kafka-sink-connector - name: es-sink-connector - config: - connector.class: io.confluent.connect.elasticsearch.ElasticsearchSinkConnector