Single handler for multiple topics

Author: remy-slCreated Sep 3, 2020Updated Oct 4, 2023

I am trying to apply exact same logic to multiple topics. Please advise how to do it.

Steps to reproduce

I am running this simple application:

python

import faust

app = faust.App(
    "app",
    broker="kafka:9092",
    topic_allow_declare=False,
    topic_disable_leader=True,
    consumer_auto_offset_reset="latest",
)

topic_a = app.topic("a")
topic_b = app.topic("b")


async def func(stream):
    async for event in stream:
        print(event)

agent_a = app.agent(topic_a)(func)
agent_b = app.agent(topic_b)(func)

Expected behavior

I'd expect to subscribe to both topics and see output similar to this

bash

[2020-09-03 14:13:21,970] [97] [INFO] Updating subscribed topics to:
┌────────────┐
│ topic name │
├────────────┤
│ a          │
│ b          │
└────────────┘
[2020-09-03 14:13:21,971] [97] [INFO] Subscribed to topic(s):
┌────────────┐
│ topic name │
├────────────┤
│ a          │
│ b          │
└────────────┘

Actual behavior

Subscribed to a single topic, whatever comes last.

bash

[2020-09-03 14:13:21,970] [97] [INFO] Updating subscribed topics to:
┌────────────┐
│ topic name │
├────────────┤
│ b          │
└────────────┘
[2020-09-03 14:13:21,971] [97] [INFO] Subscribed to topic(s):
┌────────────┐
│ topic name │
├────────────┤
│ b          │
└────────────┘

If I switch the agent declaration order...

python

agent_b = app.agent(topic_b)(func)
agent_a = app.agent(topic_a)(func)

then respectively only topic "a" is subscribed to.

bash

[2020-09-03 14:13:21,970] [97] [INFO] Updating subscribed topics to:
┌────────────┐
│ topic name │
├────────────┤
│ a          │
└────────────┘
[2020-09-03 14:13:21,971] [97] [INFO] Subscribed to topic(s):
┌────────────┐
│ topic name │
├────────────┤
│ a          │
└────────────┘

Versions

Python version 3.8.3
Faust versions 1.10.4 and 1.11.0a1
Operating system Debian
Kafka version - running my POC on confluentinc/cp-kafka:5.5.1 image - not sure what kafka version is this

Source: robinhood/faust

View original on GitHub View discussion on GitHub