How to pass multiple messages to publishers in router?

[yann-combarnous]

We are using delay handler registration, per FastStream documentation and try to batch publish after processing, using code below:

router = KafkaRouter(
    handlers=(
        KafkaRoute(
            handle_msg,
            topic_in_name,
            batch=True,
            publishers=(KafkaPublisher(topic_out_name, batch=True),),
        ),
    )
)
broker.include_router(router)

However, we keep getting the error "AssertionError: Please, connect() the broker first.", but as the consumer is parsing the incoming messages and is using the same broker as the producer, this does not seem to make sense, any tip?

Error comes from: 'faststream.confluent.publisher.asyncapi.AsyncAPIBatchPublisher object'

Return type of our message handler function is Tuple[MyMessage, ...]:

[kumaranvpl]

Hello @yann-combarnous,

It would be helpful for us if you give us minimal reproducible example to reproduce your issue. And how do you run your FastStream code?

[yann-combarnous]

Thx @kumaranvpl , I think I almost nailed down the root cause, and will post it if confirmed later.

Quick question to finish my investigation: if I have below KafkaRoute:

        KafkaRoute(
            handle_msg,
            topic_in_name,
            batch=True,
            publishers=(KafkaPublisher(topic_out_name, batch=True),),
        ),

what should my handler function return ?

When I return return tuple(list_of_messages), then I get an array in the outgoing topic.

When I call in the handler function

publisher = broker.publisher(topic_out_name, batch=True)
await publisher.publish(*outgoing_messages)

Then I get individual messages in the outgoing topic.