move logic to parent

kpops/components/base_components/kafka_app.py

@@ -117,6 +117,14 @@ async def clean(self, dry_run: bool) -> None:
             log.info(f"Uninstall cleanup job for {self.helm_release_name}")
             await self.destroy(dry_run)
 
+    async def update_cleaner_with_cluster_values(self) -> None:
+        """Update cleaner with cluster values if the release exists."""
+        cluster_values = await self.fetch_values_from_cluster()
+        if cluster_values:
+            self.app = self.app.__class__.model_validate(cluster_values)
+            self.app.name_override = self.helm_name_override
+            log.debug("Updated cleaner with cluster values")
+
 
 class KafkaApp(PipelineComponent, ABC):
     """Base component for Kafka-based components.

kpops/components/streams_bootstrap/producer/producer_app.py

@@ -37,14 +37,6 @@ async def clean(self, dry_run: bool) -> None:
         self.app.streams.delete_output = True
         await super().clean(dry_run)
 
-    async def update_cleaner_with_cluster_values(self):
-        """Update cleaner with cluster values if the release exists."""
-        cluster_values = await self.fetch_values_from_cluster()
-        if cluster_values:
-            self.app = ProducerAppValues.model_validate(cluster_values)
-            self.app.name_override = self.helm_name_override
-            log.debug("Updated producer cleaner with cluster values")
-
 
 class ProducerApp(KafkaApp, StreamsBootstrap):
     """Producer component.

kpops/components/streams_bootstrap/streams/streams_app.py

@@ -48,14 +48,6 @@ async def clean(self, dry_run: bool) -> None:
         if self.app.stateful_set and self.app.persistence.enabled:
             await self.clean_pvcs(dry_run)
 
-    async def update_cleaner_with_cluster_values(self) -> None:
-        """Update cleaner with cluster values if the release exists."""
-        cluster_values = await self.fetch_values_from_cluster()
-        if cluster_values:
-            self.app = StreamsAppValues.model_validate(cluster_values)
-            self.app.name_override = self.helm_name_override
-            log.debug("Updated streams cleaner with cluster values")
-
     async def clean_pvcs(self, dry_run: bool) -> None:
         app_full_name = super(HelmApp, self).full_name
         pvc_handler = await PVCHandler.create(app_full_name, self.namespace)