subscriber received two pieces of data: one normal data and one empty data

[willnufe]

I developed a service using the FastAPI framework in combination with a RabbitMQ message queue. After processing a RabbitMQ message, the message_consumer function pushes the result to a cloud-based Kafka message queue, which is then processed by subsequent sub-services.

During a single data test, it was observed that the sub-service received two pieces of data: one normal data and one empty data.

1. main_service

# from propan.fastapi import RabbitRouter
from faststream.rabbit.fastapi import RabbitRouter
from pydantic import BaseModel

from faststream import FastStream, Context, ContextRepo
from faststream.kafka import KafkaBroker


kafka_broker = KafkaBroker(KAFKA_BOOTSTRAP_SERVERS)

@asynccontextmanager
async def lifespan(app: FastAPI):

    async with kafka_broker:
        yield

    yield
    
    log.logger.info("worker exited")



router = RabbitRouter(rabbit_url, lifespan=lifespan, log_level=3)
app = FastAPI()




class Message(BaseModel):
    session: dict
    source: dict
    customData: dict
    response_route: dict


class Message2(BaseModel):
    session_id: str
    recog_result: str
    error_code: int
    file_id_list: list
    timestamp_list: list
    label: str
    customData: dict
    url: str
    time_start_list: list
    file_name: str



@kafka_broker.publisher(KAFKA_TOPIC)
@router.subscriber("test")
async def message_consumer(m):

	try:
		
		....
		response_dict = {}
		....

	except Exception as e:

		log.logger.error(f"[messageConsumerInner]: {e}")


	finally:

		start_time = int(time.time() * 1000)
		await kafka_broker.publish(Message2(**response_dict), KAFKA_TOPIC)
		end_time = int(time.time() * 1000)
		log.logger.info(f"[sendKafka]: {end_time - start_time}ms || {file_name}")



@router.post("/behavior/vad/")
async def generate(raw_request: Request):

    request_dict = await raw_request.json()

    await router.broker.publish(Message(**request_dict), "test")


app.include_router(router)

      
if __name__ == "__main__":


    uvicorn.run(
        "__main__:app",
        host='0.0.0.0',
        port=8080,
        workers=4,
        limit_concurrency=20,
        reload=False
    )

2. sub_service

import os
import time
import json

from faststream import FastStream
from faststream.kafka import KafkaBroker
from pydantic import BaseModel

broker = KafkaBroker(KAFKA_BOOTSTRAP_SERVERS, log_level=3)

app = FastStream(broker)

@broker.subscriber(
    KAFKA_TOPIC, group_id=KAFKA_CONSUMER_GROUP_ID,
    session_timeout_ms=180000, heartbeat_interval_ms=60000,
    auto_commit=True, auto_offset_reset='earliest'

)
async def handle_msg(message):
   ...

[kumaranvpl]

Hello @willnufe,

This is happening because you are using both await kafka_broker.publish and @kafka_broker.publisher(KAFKA_TOPIC). Please pick either one of it and stick with it.

Why are you receiving an additional empty message in sub_service?
The normal data was published in await kafka_broker.publish line itself and the empty data is published by decorator, which publishes whatever message returned by function. In you case there are no explicit return statements. So, by default, None is returned and that is published as empty data.

What is the solution?
Either explicitly publish it:

# no publisher decorator here
@router.subscriber("test")
async def message_consumer(m):
	try:
		....
		response_dict = {}
		....
	except Exception as e:
		log.logger.error(f"[messageConsumerInner]: {e}")
	finally:
		start_time = int(time.time() * 1000)
		await kafka_broker.publish(Message2(**response_dict), KAFKA_TOPIC)
		end_time = int(time.time() * 1000)
		log.logger.info(f"[sendKafka]: {end_time - start_time}ms || {file_name}")

Or just return the message object and let the decorator take care of publishing

@kafka_broker.publisher(KAFKA_TOPIC)
@router.subscriber("test")
async def message_consumer(m):
	try:
		....
		response_dict = {}
		....
	except Exception as e:
		log.logger.error(f"[messageConsumerInner]: {e}")
	finally:
	        # no kafka_broker.publish here
		return Message2(**response_dict)

Please pick one and use it. You shouldn't have the problem anymore.

Sorry for the late reply.