[Python Client] How to subscribe to two topics - both with different schemas?

[Samreay]

Hi all, I'm trying to write up a simple proof-of-concept application, wherein all it does it subscribe to two separate topics, each with their own different AvroSchema. So far, this doesn't seem possible to do easily with the (synchronous) python client, because each Consumer only supports one type of schema, and a consumer.receive() blocks.

Does anyone know a nice solution for this?

[Samreay]

What I've come up with doesn't seem optimal, but it works by hiding away multiple blocking synchronous code blocks into threads and then putting the messages received into an janus.Queue (asyncio.Queues are not thread safe, so we swapped to janus). This all works under FastAPI to support the existing REST endpoints we serve.

class Consumer(Thread):
    def __init__(self, queue: janus.Queue, topic: str, schema: pulsar.schema.AvroSchema):
        super().__init__()
        self.queue = queue
        self.pulsar_client = pulsar.Client("pulsar://pulsar:6650")
        self.topic = topic
        self.schema = schema

    def run(self):

        reader = self.pulsar_client.create_reader(
            self.topic, start_message_id=pulsar.MessageId.latest, schema=self.schema
        )
        while True:
            msg = reader.read_next()
            topic = msg.topic_name()
            if "/" in topic:
                topic = topic.split("/")[-1]
            message = from_avro(msg.value())
           self.queue.sync_q.put_nowait(message)

Then when the FastAPI app starts you can kick off the threads, and also launch an async method to get from the asyncio.Queue passed into the Consumer threads.

app = FastAPI()
app.queue = asyncio.Queue()

@app.on_event("startup")
async def startup_event():
    consumer_preds = Consumer(app.queue, "predictions", Prediction.schema())
    consumer_preds.start()
    consumer_trades = Consumer(app.queue, "trades", Trade.schema())
    consumer_trades.start()

    asyncio.create_task(receive_message())

Prediction.schema and what not are just some wrappers we have to convert pydantic to avro schemas.

[yaalsn]

If pydantic, how about using JSON instead of Avro, you can convert the models to JSON easily

https://docs.pydantic.dev/latest/usage/exporting_models/

[Samreay]

Doesn't this have the same issue as Avro, which is if I want to enforce schema on the topics with a well defined producer and consumer, it looks like the synchronous single-schema Consumer in the API means we'd need to turn to multithreading or multiprocessing? Does the JSON schema allow us some way around this?

[yaalsn]

Not JSON schema, I mean sending JSON string, just bytes. I think this can solve your problem quickly.

[Samreay]

Oh yeah, if we just accept json and do the schema validation ourself, no issues. I was hoping to make use of Pulsars schema registry features and how it enforces things to guarantee a consumer with an incompatible version of the schema can't connect. Figured if the pulsar devs have put in the hard work to implement all of that, I should make use of it! (There's also no rush on our side, the nasty threading approach works, but its very nice.