Skip to content

feat: resolve KafkaConsumer topics from expressions and config parameters (fixes #693) - #696

Merged
dgafka merged 5 commits into
mainfrom
feature/kafka-dynamic-topics-693
Aug 22, 2026
Merged

feat: resolve KafkaConsumer topics from expressions and config parameters (fixes #693)#696
dgafka merged 5 commits into
mainfrom
feature/kafka-dynamic-topics-693

Conversation

@dgafka

@dgafka dgafka commented Aug 18, 2026

Copy link
Copy Markdown
Member

Why is this change proposed?

The KafkaConsumer attribute's topic list was fixed at attribute-scan time, so a multi-tenant deployment needed a full redeploy to pick up a newly added tenant's topic. Ecotone's expression language, already used for tenant resolution elsewhere, was never wired into topic resolution, and there was no single path for resolving cross-framework config values (Symfony container parameters, Laravel config, Tempest config) into it either.

Closes #693 by letting a topics expression pull the list from a referenced service or a config value, evaluated once at first subscribe.

Description of Changes

Resulting flow

Before

graph LR
  A1["topics: 'a', 'b'"] --> B1["scanned once, baked in literally"] --> C1["subscribe('a', 'b')"]
Loading

After

graph LR
  A2["topics: expression string"] --> B2["baked as literal string (unchanged)"] --> C2["first subscribe: evaluate expression"] --> D2["merge + apply topic reference-name mapping"] --> E2["subscribe(resolved topics)"]
Loading

Any topics entry containing ( is now evaluated via the expression language at the first subscribe, instead of being used as a literal topic name. Expression results merge with any literal entries and pass through the existing topic reference-name mapping, same as static topics always have. Tempest's config-variable service is extended to resolve a typed config object's property (previously env-vars only), bringing parameter(...) to parity with the Symfony and Laravel integrations.

Out of scope

  • Live re-subscription after startup — topics stay fixed for the process lifetime after the first subscribe.
  • Resolving Symfony's native %parameter% placeholder syntax directly — confirmed structurally unreachable for Kafka's isolated container build; parameter(...) inside an expression is the cross-framework path instead.

Example

#[KafkaConsumer(
    endpointId: 'orders',
    topics: "reference('tenantTopics')->getOrderTopics()",
)]

#[KafkaConsumer(
    endpointId: 'orders',
    topics: "parameter('app.multi_tenancy.kafka.orders.topics')",
)]

Pull Request Contribution Terms

  • I have read and agree to the contribution terms outlined in CONTRIBUTING.

dgafka added 5 commits August 18, 2026 08:00
Topics containing '(' are now evaluated via ExpressionEvaluationService
at first consumer subscribe, allowing reference('service')->method() and
parameter('name') to supply topic lists dynamically for multi-tenant
setups (#693). Results still pass through the existing topic
reference-name mapping.
…bleService

getByName('Some\Config\Class::property') now reads a public property off
a config object registered in Tempest's container, in addition to the
existing plain env-var lookup. Brings parameter('...') expression
support in line with the Symfony and Laravel integrations.
…runners

The 300ms bound flaked on the MySQL CI job (322ms observed) since the
DBAL channel does a real DB round trip. Matches the 1s-polling-timeout
headroom already applied to the sibling in-memory-channel assertion.
Default 100ms execution time limit was too tight to reliably consume
5 messages under CI load, causing the consumer to stop early. Align
with the sibling execution-time-limit test's 5000ms timeout.
…umption

Same root cause as the AmqpStreamChannelTest fix: these tests consumed
2-5 messages from a real DbalBackedMessageChannelBuilder queue while
relying on the default 100ms execution time limit, which a real DB
round trip can exceed under CI load (see 4fe299f). Widen to 5000ms
for 2-message cases and 10000ms for 3-5 message cases, matching the
timeouts already used by sibling DBAL tests in this package.
@dgafka
dgafka merged commit 6ef645b into main Aug 22, 2026
8 checks passed
@dgafka
dgafka deleted the feature/kafka-dynamic-topics-693 branch August 22, 2026 10:33
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

More dynamic way of defining topics for KafkaConsumer

1 participant