Handling Partial Success in Batch Processing with PullSub

[Atanusaha143]

I'm currently working on a FastStream subscriber implementation with a PullSub configuration and have a question about handling partial successes when processing messages in batches.

Here’s an outline of my setup:

class Subscriber:
    @nats_broker.subscriber(
        subject="some_subject_name",
        stream="some_stream_name",
        durable="some_consumer_name",
        pull_sub=PullSub(batch=True, batch_size=5),
        no_ack=True, 
        retry=True,
    )
    async def handle_subscriber(self, msg: NatsMessage, logger: Logger):
        try:
            logger.info("Started processing requests for 'some_subject_name'.")

            decoded_messages = Subscriber._decode_messages(msg=msg)
            logger.debug(f"Decoded messages: {decoded_messages}")

            message_dtos = Subscriber._map_message_dtos(decoded_messages=decoded_messages)
            logger.debug(f"Mapped message DTOs: {message_dtos}")

            await msg.ack()
            logger.info("Successfully processed requests.")
        except Exception as e:
            logger.error("An exception occurred during processing requests.", exc_info=e)
            await msg.nack()

    @staticmethod
    def _decode_messages(msg: NatsMessage):
        decoded_messages: list = []
        for message in msg.body:
            decoded_message = json.loads(message.decode("utf-8"))
            decoded_messages.append(decoded_message)
        return decoded_messages

    @staticmethod
    def _map_message_dtos(decoded_messages):
        pass

Question

In this setup, when a batch of 5 messages is pulled, what happens if:

1. 3 messages are processed successfully
2. 2 messages fail?

Specifically:

1. Does await msg.ack() acknowledge the entire batch, marking all 5 messages as processed?
2. How can I implement granular ack and nack handling for individual messages in the batch to ensure only successfully processed messages are acknowledged while the failed ones are retried?

Goal

I want to:

1. Acknowledge (ack) each successfully processed message individually.
2. Negative acknowledge (nack) failed messages to trigger retries.

🙏🏻 Any guidance or best practices for implementing this would be greatly appreciated. Thank you for your help!