Skip to content

Migrate from V6 to V7

Automatic loading of namespaced custom components

KPOps is now distributed as a Python namespace package (as defined by PEP 420). This allows us to standardize the namespace kpops.components for both builtin and custom pipeline components.

As a result of the restructure, some imports need to be adjusted:

KPOps Python API

1
2
- import kpops
+ import kpops.api as kpops

builtin KPOps components

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
- from kpops.components import (
-     HelmApp,
-     KafkaApp,
-     KafkaConnector,
-     KafkaSinkConnector,
-     KafkaSourceConnector,
-     KubernetesApp,
-     StreamsBootstrap,
-     ProducerApp,
-     StreamsApp,
-     PipelineComponent,
-     StreamsApp,
-     ProducerApp,
- )
+ from kpops.components.base_components import (
+     HelmApp,
+     KafkaApp,
+     KafkaConnector,
+     KafkaSinkConnector,
+     KafkaSourceConnector,
+     KubernetesApp,
+     PipelineComponent,
+ )
+ from kpops.components.streams_bootstrap import (
+     StreamsBootstrap,
+     StreamsApp,
+     ProducerApp,
+ )

your custom KPOps components

config.yaml

1
- components_module: components

Python module

1
2
- components/__init__.py
+ kpops/components/custom/__init__.py

Rename app field

The app attribute of the builtin KPOps components has been renamed to better differentiate them. Both your pipeline.yaml and defaults.yaml files have to be updated, e.g.:

defaults.yaml

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
  kubernetes-app:
-   app: {}
+   values: {}

  helm-app:
-   app: {}
+   values: {}

  kafka-app:
-   app: {}
+   values: {}

  streams-app:
-   app: {}
+   values: {}

  producer-app:
-   app: {}
+   values: {}

  kafka-connector:
-   app: {}
+   config: {}

  kafka-source-connector:
-   app: {}
+   config: {}

  kafka-sink-connector:
-   app: {}
+   config: {}

Call destroy from inside of reset or clean

Before v7, the KPOps CLI executed destroy before running reset/clean to ensure the component was destroyed.

This logic has changed. The destroy method is now called within the PipelineComponent's reset/clean.

During migrating to v7, you should check your custom components and see if they override the reset/clean methods. If so, you need to call the supermethod reset/clean to trigger the destroy inside the parent class. Alternatively, if you are implementing the PipelineComponent class, you need to call the destroy method at the beginning of the method.

components.py

For example, when creating a custom StreamsApp or ProducerApp (or any other custom component), you must call the supermethod reset/clean to execute the destroy in the parent class. Otherwise, the logic of destroy will not be executed!

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class MyStreamsApp(StreamsApp):

    @override
    async def clean(self, dry_run: bool) -> None:
+       await super().clean(dry_run)
        # Some custom clean logic
        # ...
        ```diff
        
 
class MyCustomComponent(PipelineComponent):
    
    @override
    async def destroy(self, dry_run: bool) -> None:
        # Some custom destroy logic
        # ...

    @override
    async def clean(self, dry_run: bool) -> None:
+       await super().clean(dry_run)
        # Some custom clean logic
        # ...