feat(kafka-connect): set connector initial state

docs/docs/resources/pipeline-components/dependencies/kpops_structure.yaml

@@ -20,6 +20,7 @@ kpops_components_fields:
   - from_
   - to
   - config
+  - initial_state
   - resetter_namespace
   - resetter_values
   kafka-sink-connector:
@@ -28,6 +29,7 @@ kpops_components_fields:
   - from_
   - to
   - config
+  - initial_state
   - resetter_namespace
   - resetter_values
   kafka-source-connector:
@@ -36,6 +38,7 @@ kpops_components_fields:
   - from_
   - to
   - config
+  - initial_state
   - resetter_namespace
   - resetter_values
   - offset_topic

docs/docs/schema/defaults.json

@@ -309,6 +309,15 @@
             "title": "ImagePullPolicy",
             "type": "string"
         },
+        "InitialState": {
+            "enum": [
+                "running",
+                "paused",
+                "stopped"
+            ],
+            "title": "InitialState",
+            "type": "string"
+        },
         "InputTopicTypes": {
             "description": "Input topic types.\n\n- INPUT: input topic\n- PATTERN: extra-topic-pattern or input-topic-pattern",
             "enum": [
@@ -654,6 +663,11 @@
                     "description": "Topic(s) and/or components from which the component will read input",
                     "title": "From"
                 },
+                "initial_state": {
+                    "$ref": "#/$defs/InitialState",
+                    "default": "running",
+                    "description": "Initial state for newly created connector"
+                },
                 "name": {
                     "description": "Component name",
                     "title": "Name",
@@ -787,6 +801,11 @@
                     "description": "Topic(s) and/or components from which the component will read input",
                     "title": "From"
                 },
+                "initial_state": {
+                    "$ref": "#/$defs/InitialState",
+                    "default": "running",
+                    "description": "Initial state for newly created connector"
+                },
                 "name": {
                     "description": "Component name",
                     "title": "Name",
@@ -862,6 +881,11 @@
                     "description": "Topic(s) and/or components from which the component will read input",
                     "title": "From"
                 },
+                "initial_state": {
+                    "$ref": "#/$defs/InitialState",
+                    "default": "running",
+                    "description": "Initial state for newly created connector"
+                },
                 "name": {
                     "description": "Component name",
                     "title": "Name",

docs/docs/schema/pipeline.json

@@ -309,6 +309,15 @@
             "title": "ImagePullPolicy",
             "type": "string"
         },
+        "InitialState": {
+            "enum": [
+                "running",
+                "paused",
+                "stopped"
+            ],
+            "title": "InitialState",
+            "type": "string"
+        },
         "InputTopicTypes": {
             "description": "Input topic types.\n\n- INPUT: input topic\n- PATTERN: extra-topic-pattern or input-topic-pattern",
             "enum": [
@@ -621,6 +630,11 @@
                     "description": "Topic(s) and/or components from which the component will read input",
                     "title": "From"
                 },
+                "initial_state": {
+                    "$ref": "#/$defs/InitialState",
+                    "default": "running",
+                    "description": "Initial state for newly created connector"
+                },
                 "name": {
                     "description": "Component name",
                     "title": "Name",
@@ -696,6 +710,11 @@
                     "description": "Topic(s) and/or components from which the component will read input",
                     "title": "From"
                 },
+                "initial_state": {
+                    "$ref": "#/$defs/InitialState",
+                    "default": "running",
+                    "description": "Initial state for newly created connector"
+                },
                 "name": {
                     "description": "Component name",
                     "title": "Name",

kpops/component_handlers/kafka_connect/connect_wrapper.py

@@ -11,8 +11,10 @@
     KafkaConnectError,
 )
 from kpops.component_handlers.kafka_connect.model import (
+    InitialState,
     KafkaConnectConfigErrorResponse,
     KafkaConnectorConfig,
+    KafkaConnectRequest,
     KafkaConnectResponse,
 )
 
@@ -21,7 +23,6 @@
 
     from kpops.config import KafkaConnectConfig
 
-HEADERS = {"Accept": "application/json", "Content-Type": "application/json"}
 
 log = logging.getLogger("KafkaConnectAPI")
 
@@ -32,29 +33,30 @@ class ConnectWrapper:
 
     def __init__(self, config: KafkaConnectConfig) -> None:
         self._config: KafkaConnectConfig = config
-        self._client = httpx.AsyncClient(timeout=config.timeout)
+        self._client = httpx.AsyncClient(
+            base_url=str(config.url),
+            headers={"Accept": "application/json", "Content-Type": "application/json"},
+            timeout=config.timeout,
+        )
 
     @property
     def url(self) -> AnyHttpUrl:
         return self._config.url
 
     async def create_connector(
-        self, connector_config: KafkaConnectorConfig
+        self, connector_config: KafkaConnectorConfig, initial_state: InitialState
     ) -> KafkaConnectResponse:
         """Create a new connector.
 
         API Reference: https://docs.confluent.io/platform/current/connect/references/restapi.html#post--connectors
         :param connector_config: The config of the connector
         :return: The current connector info if successful.
         """
-        config_json: dict[str, Any] = connector_config.model_dump()
-        connect_data: dict[str, Any] = {
-            "name": connector_config.name,
-            "config": config_json,
-        }
-        response = await self._client.post(
-            url=f"{self.url}connectors", headers=HEADERS, json=connect_data
+        payload = KafkaConnectRequest(
+            config=connector_config,
+            initial_state=initial_state,
         )
+        response = await self._client.post("/connectors", json=payload.model_dump())
         if response.status_code == httpx.codes.CREATED:
             log.info(f"Connector {connector_config.name} created.")
             log.debug(response.json())
@@ -63,25 +65,19 @@ async def create_connector(
             log.warning(
                 "Rebalancing in progress while creating a connector... Retrying..."
             )
-
             await asyncio.sleep(1)
-            await self.create_connector(connector_config)
+            await self.create_connector(connector_config, initial_state)
 
         raise KafkaConnectError(response)
 
-    async def get_connector(self, connector_name: str | None) -> KafkaConnectResponse:
-        """Get information about the connector.
+    async def get_connector(self, connector_name: str) -> KafkaConnectResponse:
+        """Get information about a connector.
 
         API Reference: https://docs.confluent.io/platform/current/connect/references/restapi.html#get--connectors-(string-name)
-        :param connector_name: Nameof the crated connector
+        :param connector_name: Name of the connector
         :return: Information about the connector.
         """
-        if connector_name is None:
-            msg = "Connector name not set"
-            raise Exception(msg)
-        response = await self._client.get(
-            url=f"{self.url}connectors/{connector_name}", headers=HEADERS
-        )
+        response = await self._client.get(f"/connectors/{connector_name}")
         if response.status_code == httpx.codes.OK:
             log.info(f"Connector {connector_name} exists.")
             log.debug(response.json())
@@ -111,8 +107,7 @@ async def update_connector_config(
 
         config_json = connector_config.model_dump()
         response = await self._client.put(
-            url=f"{self.url}connectors/{connector_name}/config",
-            headers=HEADERS,
+            f"/connectors/{connector_name}/config",
             json=config_json,
         )
 
@@ -143,8 +138,7 @@ async def validate_connector_config(
         :return: List of all found errors
         """
         response = await self._client.put(
-            url=f"{self.url}connector-plugins/{connector_config.class_name}/config/validate",
-            headers=HEADERS,
+            f"/connector-plugins/{connector_config.class_name}/config/validate",
             json=connector_config.model_dump(),
         )
 
@@ -172,9 +166,7 @@ async def delete_connector(self, connector_name: str) -> None:
         :param connector_name: Configuration parameters for the connector.
         :raises ConnectorNotFoundException: Connector not found
         """
-        response = await self._client.delete(
-            url=f"{self.url}connectors/{connector_name}", headers=HEADERS
-        )
+        response = await self._client.delete(f"/connectors/{connector_name}")
         if response.status_code == httpx.codes.NO_CONTENT:
             log.info(f"Connector {connector_name} deleted.")
             return

kpops/component_handlers/kafka_connect/kafka_connect_handler.py

@@ -8,6 +8,7 @@
     ConnectorNotFoundException,
     ConnectorStateException,
 )
+from kpops.component_handlers.kafka_connect.model import InitialState
 from kpops.utils.colorify import magentaify
 from kpops.utils.dict_differ import render_diff
 
@@ -24,7 +25,11 @@ def __init__(self, connect_wrapper: ConnectWrapper):
         self._connect_wrapper = connect_wrapper
 
     async def create_connector(
-        self, connector_config: KafkaConnectorConfig, *, dry_run: bool
+        self,
+        connector_config: KafkaConnectorConfig,
+        *,
+        initial_state: InitialState,
+        dry_run: bool,
     ) -> None:
         """Create a connector.
 
@@ -41,7 +46,9 @@ async def create_connector(
                 await self._connect_wrapper.update_connector_config(connector_config)
 
             except ConnectorNotFoundException:
-                await self._connect_wrapper.create_connector(connector_config)
+                await self._connect_wrapper.create_connector(
+                    connector_config, initial_state
+                )
 
     async def destroy_connector(self, connector_name: str, *, dry_run: bool) -> None:
         """Delete a connector resource from the cluster.
@@ -69,9 +76,12 @@ async def __dry_run_connector_creation(
             connector = await self._connect_wrapper.get_connector(connector_name)
 
             log.info(f"Connector Creation: connector {connector_name} already exists.")
-            if diff := render_diff(connector.config, connector_config.model_dump()):
+            if diff := render_diff(
+                connector.config.model_dump(), connector_config.model_dump()
+            ):
                 log.info(f"Updating config:\n{diff}")
 
+            # TODO: refactor, this should not be here
             log.debug(connector_config.model_dump())
             log.debug(f"PUT /connectors/{connector_name}/config HTTP/1.1")
             log.debug(f"HOST: {self._connect_wrapper.url}")

kpops/component_handlers/kafka_connect/model.py

@@ -1,10 +1,12 @@
-from enum import StrEnum
+from enum import StrEnum, auto
 from typing import Any, ClassVar
 
 import pydantic
 from pydantic import (
     BaseModel,
     ConfigDict,
+    computed_field,
+    field_serializer,
     field_validator,
     model_serializer,
 )
@@ -95,14 +97,38 @@ def serialize_model(
         return {by_alias(self, name): to_str(value) for name, value in result.items()}
 
 
+class InitialState(StrEnum):
+    RUNNING = auto()
+    PAUSED = auto()
+    STOPPED = auto()
+
+    @property
+    def api_value(self) -> str:
+        return self.value.upper()
+
+
 class ConnectorTask(BaseModel):
     connector: str
     task: int
 
 
+class KafkaConnectRequest(BaseModel):
+    config: KafkaConnectorConfig
+    initial_state: InitialState
+
+    @computed_field
+    @property
+    def name(self) -> str:
+        return self.config.name
+
+    @field_serializer("initial_state")
+    def serialize_initial_state(self, initial_state: InitialState) -> str:
+        return initial_state.api_value
+
+
 class KafkaConnectResponse(BaseModel):
     name: str
-    config: dict[str, str]
+    config: KafkaConnectorConfig
     tasks: list[ConnectorTask]
     type: str | None = None
 

kpops/components/base_components/kafka_connector.py

@@ -14,6 +14,7 @@
     HelmRepoConfig,
 )
 from kpops.component_handlers.kafka_connect.model import (
+    InitialState,
     KafkaConnectorConfig,
     KafkaConnectorType,
 )
@@ -107,6 +108,7 @@ class KafkaConnector(PipelineComponent, ABC):
     Should only be used to set defaults
 
     :param config: Connector config
+    :param initial_state: Initial state for newly created connector
     :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
@@ -115,6 +117,10 @@ class KafkaConnector(PipelineComponent, ABC):
     config: KafkaConnectorConfig = Field(
         description=describe_attr("config", __doc__),
     )
+    initial_state: InitialState = Field(
+        default=InitialState.RUNNING,
+        description=describe_attr("initial_state", __doc__),
+    )
     resetter_namespace: str | None = Field(
         default=None, description=describe_attr("resetter_namespace", __doc__)
     )
@@ -181,7 +187,7 @@ async def deploy(self, dry_run: bool) -> None:
                 await schema_handler.submit_schemas(to_section=self.to, dry_run=dry_run)
 
         await get_handlers().connector_handler.create_connector(
-            self.config, dry_run=dry_run
+            self.config, initial_state=self.initial_state, dry_run=dry_run
         )
 
     @override