Skip to content

Commit

Permalink
Populate offset topic to resetter during instantiation
Browse files Browse the repository at this point in the history
remove custom logic
  • Loading branch information
disrupted committed Jun 12, 2024
1 parent 80cf303 commit 54a2581
Showing 1 changed file with 12 additions and 4 deletions.
16 changes: 12 additions & 4 deletions kpops/components/base_components/kafka_connector.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from functools import cached_property
from typing import Any, NoReturn

import pydantic
from pydantic import Field, PrivateAttr, ValidationInfo, computed_field, field_validator
from typing_extensions import override

Expand All @@ -25,6 +26,11 @@
from kpops.utils.colorify import magentaify
from kpops.utils.docstring import describe_attr

try:
from typing import Self # pyright: ignore[reportAttributeAccessIssue]
except ImportError:
from typing_extensions import Self

log = logging.getLogger("KafkaConnector")


Expand Down Expand Up @@ -205,6 +211,12 @@ 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.app.config.offset_topic = self.offset_topic
return self

@computed_field
@cached_property
def _resetter(self) -> KafkaConnectorResetter:
Expand All @@ -217,15 +229,11 @@ def apply_from_inputs(self, name: str, topic: FromTopic) -> NoReturn:

@override
async def reset(self, dry_run: bool) -> None:
if self.offset_topic:
self._resetter.app.config.offset_topic = self.offset_topic
await self._resetter.reset(dry_run)

@override
async def clean(self, dry_run: bool) -> None:
await super().clean(dry_run)
if self.offset_topic:
self._resetter.app.config.offset_topic = self.offset_topic
await self._resetter.clean(dry_run)


Expand Down

0 comments on commit 54a2581

Please sign in to comment.