Commit strategy and delaying acknowledgement with Kafka connector.

[javadevmtl]

Hi, I have implemented a retry mechanism using the reactive messaging with the Kafka connector. The system uses 3 topics, a "main" topic, a retry topic and a dlq topic.

• The main topic is setup to dlq into the retry topic, the retry topic is setup to dlq into the dlq topic and processing stops there.
• The retry topic uses the metadata/headers and looks for a retry count if the count is below the max it will emit the message back into the main topic, otherwise the message will dlq into the dlq topic and stop there.
• The retry topic will also check the message time if it's below the threshold then it will delay it otherwise emit it using the logic below:

if (delayTimeSecs == 0)
    return Uni
            .createFrom()
            .item(in.addMetadata(metadataOut)
                    .withPayload(in.getPayload()));
else
    return Uni
            .createFrom()
            .item(in.addMetadata(metadataOut)
                    .withPayload(in.getPayload()))
            .onItem().delayIt()
            .by(Duration.ofSeconds(delayTimeSecs));

Note: The delay delayTimeSecs can be up to 5 minutes.

All my topics are set to default commit strategy which is throttled. When a message is delayed as above, in the logs we see the message There is more than 1 message that has not been acknowledged for more than 60 seconds...

So the the questions are:
1- Can there be different commit strategies for each topic?
2- If so, should I increase the retry topic throttled.unprocessed-record-max-age.ms so I don't see the error any more specifically for the retry topic.
3- Or should I set the retry topic to latest

[quarkus-bot[bot]]

/cc @alesj, @cescoffier, @ozangunalp

[ozangunalp]

You can configure commit strategies independently per channel like mp.messaging.incoming.retry.commit-strategy=latest.

Normally you'd have to handle message (n)acks manually when you are consuming Message types. In your case, you're probably getting away with it because you are returning the incoming message as outgoing, which will be acked on Kafka delivery confirmation.

I think both of your options can be viable depending on your case:

• If you consider that your processing is finished after delayed message delivery, then you need to increase the throttled.unprocessed-record-max-age.ms.
• Otherwise I think you can try acknowledging the message before calling delayIt on your uni. So with the throttled commit strategy it will register the message as processed.
  If I am not mistaken it would be something like

Uni.createFrom()
    .item(in.addMetadata(metadataOut).withPayload(in.getPayload()))
    .onItem().call(() -> Uni.createFrom().completionStage(in.ack))
    .onItem().delayIt().by(Duration.ofSeconds(delayTimeSecs))

• If you don't expect a high load on that topic you can choose latest commit strategy, which will commitAsync each processed message, without waiting for the commit confirmation.

[javadevmtl]

Hi, if I delay after the ack the message will go out right away right? I need to make sure the messages goes out after the delay.

Also if I ack before and delay after, it means we will get into "at most once" processing, since at this point the offset is committed, but the message is still in memory waiting to go out and there's possibility of message loss.

I will try latest strategy with the "retry" topic. It's only supposed to happen on exceptional cases.