Control polling logic based on processing ack

[prakashelango]

current implementation,

No fallback of messages to sqs and processing again by other containers, we needed an exactly once processing to downstream.
Problem:

1. Reading from sqs indefinitely and processing messages
2. Puts or Updates optimistic locking enabled record in DDB we have a high concurrency 1000 tps on same record. If conditional check fails we got to bruteforce and retry till all gets succeed.
   3.Because of retry looping at higher rate on same record in-flight messages are too high.

Question:
How can we control or paced demand polling from sqs only if 500 messages successfully processed and emitted to downstream, or Not to overwhelm in retry and cause higher failure.

Multi.createBy().repeating()
            .supplier(() ->
                sqsMessagePoller.receiveMessages()
                    .chain(this::transformSQSMessageToDomain)
                    .onFailure().invoke(failure -> Log.error("Error processing message!", failure))
                    .subscribe().with(tuple -> Log.debugf("Successfully processed messages and invoked api"),
                        throwable -> Log.error("Failed to process ", throwable))
            )
            .withDelay(Duration.ofMillis(10))
            .indefinitely().subscribe().with(x -> Log.debug("Current iteration of processing message complete"));

private Uni<RemainingCharge> transformSQSMessageToDomain(final Domain domain) {
        return chargeService.fetchRateAndCalculateCharges(domain)
            .chain(charges -> ChargeService.updateChargeOnly(charges, domain))
            .onFailure(ConditionalCheckFailedException.class).call(() -> {
                Log.infof("Conditional check failed for domain: %s. Retrying with updated values.", domain);
                return transformSqsMessageToDomain(domain);
            })
            .onFailure().invoke(failure -> Log.errorf("Failed to update charge for domain: %s after retries. Storing in fallback.", domain));
    }

Can someone help on this please.
I took a look at https://quarkus.io/blog/mutiny-back-pressure/ however this looks outdated with latest mutiny.