Repository navigation
Conversation
…ilities Topic support (`from`/`to`) is now provided by the capability classes `ConsumesTopics` and `ProducesTopics` instead of `PipelineComponent`. Components without a capability reject the respective section. BREAKING CHANGES: - `helm-app` and `kubernetes-app` (and custom subclasses) no longer accept `from`/`to`. - `producer-app`/`kafka-source-connector` reject `from`, `consumer-app` rejects `to` (now an error even when supplied via defaults of a shared parent key). - Kafka connectors are now `KafkaApp`s: `kafka-app` defaults apply to them. - `PipelineComponent.supports_from_section()` removed; use `isinstance(c, ConsumesTopics)`. - Topic hook methods moved from `PipelineComponent` to `ConsumesTopics`/`ProducesTopics`.
disrupted
changed the title
There was a problem hiding this comment.
🟡 Changes recommended
The generated schemas accept version: null, but the runtime validator rejects it, allowing pipelines to pass schema validation and still fail to load.
1 open issue
What changed in this PR
This refactors KPOps components so their types express supported topic capabilities and values models, while preserving generated pipeline output.
Changes:
- Introduces explicit topic input/output capabilities and rejects unsupported sections.
- Adds generic values typing and updates component discovery and defaults ancestry.
- Updates Helm repository handling, generated schemas, documentation generation, and regression tests.
| File | Description |
|---|---|
| tests/components/test_kafka_source_connector.py | Tests source input rejection and capability membership. |
| tests/components/test_kafka_sink_connector.py | Checks sink input capability. |
| tests/components/test_helm_app.py | Removes an unnecessary repository field override. |
| tests/components/test_component_hierarchy.py | Tests ancestry, discovery, values typing, and section rejection. |
| tests/components/streams_bootstrap/test_streams_bootstrap.py | Tests default and custom Helm repositories. |
| tests/components/streams_bootstrap/test_streams_app.py | Tests rejection of a null chart version. |
| tests/components/streams_bootstrap/test_producer_app.py | Updates producer capability assertions. |
| tests/components/streams_bootstrap/test_consumer_app.py | Checks that consumers have no output field. |
| kpops/pipeline.py | Makes topic wiring capability-aware. |
| kpops/core/registry.py | Excludes capabilities and parameterized generics from discovery. |
| kpops/components/streams_bootstrap/streams/streams_app.py | Declares capabilities and specializes app and cleaner values. |
| kpops/components/streams_bootstrap/streams/model.py | Specializes values with streams configuration. |
| kpops/components/streams_bootstrap/producer/producer_app.py | Declares output capability and typed cleaner values. |
| kpops/components/streams_bootstrap/producer/model.py | Specializes values with producer configuration. |
| kpops/components/streams_bootstrap/model.py | Makes the Kafka configuration type generic. |
| kpops/components/streams_bootstrap/consumer/model.py | Introduces a generic consumer values base. |
| kpops/components/streams_bootstrap/consumer/consumer_app.py | Declares input capability and typed cleaner values. |
| kpops/components/streams_bootstrap/consumer_producer/model.py | Specializes the shared consumer values base. |
| kpops/components/streams_bootstrap/consumer_producer/consumer_producer_app.py | Declares both topic capabilities and typed values. |
| kpops/components/streams_bootstrap/base.py | Adds generic values, repository fallback, and version validation. |
| kpops/components/base_components/pipeline_component.py | Rejects unsupported sections and preserves serialization order. |
| kpops/components/base_components/kubernetes_app.py | Makes Kubernetes values typing generic. |
| kpops/components/base_components/kafka_connector.py | Adopts topic capabilities and shared Kafka deployment behavior. |
| kpops/components/base_components/kafka_app.py | Defines topic consumption and production capabilities. |
| kpops/components/base_components/helm_app.py | Adds generic values and effective repository access. |
| kpops/components/base_components/cleaner.py | Makes cleaner values typing generic. |
| kpops/components/base_components/base_defaults_component.py | Filters capabilities and generic specializations from ancestry. |
| kpops/components/base_components/__init__.py | Exports the new topic capabilities. |
| hooks/gen_docs/gen_docs_components.py | Resolves generic origins for inheritance documentation. |
| docs/docs/schema/pipeline.json | Updates pipeline component fields and nullable types. |
| docs/docs/schema/defaults.json | Updates defaults definitions for the revised hierarchy. |
Review effort: Balanced
…alidation error there's very little benefit in keeping it
tobbber
reviewed
Oct 28, 2026
|
|
||
| The attributes correspond to keys and values that are used as values for the streams-bootstrap Helm chart. | ||
|
|
||
| :param kafka: consumer-app kafka section |
Contributor
There was a problem hiding this comment.
this should only be in ConsumerAppValues right?
also maybe remove the :param: lines in ConsumerAppValues for attributes defined in BaseConsumerAppValues
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This makes the component hierarchy match what each component actually supports. Components carry their own values type, and topic input and output support is defined explicitly instead of being inherited by everything.
The change removes incompatible field overrides and adds tests to protect defaults precedence and component discovery. Existing generated pipelines stay unchanged.
Breaking changes: