KafkaDelayedRetryTopic consumer per delayed retry topic

[joystuffs]

Hi,

As far as I understood, when manually nacking an incoming message, this triggers the delayed-retry-topic (DRT) mechanism.

This works fine and I can see the retries being handled as expected.

However, I noticed that there is only one instance of the consumer handling all DRTs.

I was wondering if there is a way to configure a dedicated consumer per specific DRT, rather than a shared one.

I know that I could manually create dedicated consumers, but doing so would cause each message to be processed twice , once by the generic DRT consumer and once by the dedicated one.

Question / Expected Behavior

Is there a configuration option or extension point that would allow each DRT to have its own independent consumer instance?

below an example of my consumer

    @WithSpan
    @Incoming("consumer")
    @Acknowledgment(Acknowledgment.Strategy.MANUAL)
    @Blocking
    @RunOnVirtualThread
    public CompletionStage<Void> consume(IncomingKafkaRecord<String, String> message) {
        // do some logic
        try {
            // fire an http request (Blocking operation)
            client.send(request);
            // ack the message
            return message.ack();
        } catch (Exception e) {
            // nack the message
            return message.nack();
        }
    }

mp.messaging.incoming.consumer.connector=smallrye-kafka
mp.messaging.incoming.consumer.topic=foo
mp.messaging.incoming.consumer.client.id=foo_client_id_${host}
mp.messaging.incoming.consumer.group.id=foo_consumer_group_id_${host}

mp.messaging.incoming.consumer.commit-strategy=ignore
mp.messaging.incoming.consumer.auto.offset.reset=latest
mp.messaging.incoming.consumer.failure-strategy=delayed-retry-topic
mp.messaging.incoming.consumer.partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
mp.messaging.incoming.consumer.tracing-enabled=false
mp.messaging.incoming.consumer.graceful-shutdown=true

# DELAYED RETRY TOPIC (DRT) CONFIGURATION

# DRT topics (in milliseconds)
kafka.topic.delayed-retry-topics=\
retry_1000,\
retry_3000,\
retry_30000


# DRT settings
mp.messaging.incoming.consumer.delayed-retry-topic.topics=${kafka.topic.delayed-retry-topics}
mp.messaging.incoming.consumer.delayed-retry-topic.max-retries=3
mp.messaging.incoming.consumer.delayed-retry-topic.timeout=60000

# Internal DRT consumer configuration
mp.messaging.incoming.consumer.delayed-retry-topic.consumer.client.id=drt_${host}
mp.messaging.incoming.consumer.delayed-retry-topic.consumer.group.id=drt_consumer_${host}
mp.messaging.incoming.consumer.delayed-retry-topic.consumer.failure-strategy=ignore
mp.messaging.incoming.consumer.delayed-retry-topic.consumer.commit-strategy=ignore
mp.messaging.incoming.consumer.delayed-retry-topic.consumer.partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
mp.messaging.incoming.consumer.delayed-retry-topic.consumer.graceful-shutdown=true
mp.messaging.incoming.consumer.delayed-retry-topic.consumer.tracing-enabled=false

many thanks in advance

[ozangunalp]

when manually nacking an incoming message

You do not need to manually nack a message. If your processing throws an error the message is nacked automatically and the failure-strategy is applied. For that your method parameter should not be a message type.

However, I noticed that there is only one instance of the consumer handling all DRTs.

That is one consumer instance per channel. So if you've multiple channels, there would be one for each channel.

The DRT consumer subscribes to all retry topics for that channel and streams the retry message back to the incoming method.
In this usage, retry topics are designed to be only for the original topic.
If your original processed topic is "foo", the retry topics are, by default, "foo_retry_3000", "foo_retry_5000", etc.
Those are configurable, and you can use the same retry topics for different channels, but you'd discard many more messages that are retries from other channels/topics.

Why do you want to create a different consumer per retry-topic ?

[joystuffs]

You do not need to manually nack a message. If your processing throws an error the message is nacked automatically and the failure-strategy is applied. For that your method parameter should not be a message type.

I know and in my case, I need to do some logic on the record headers so that's why I manually nack the record.

That is one consumer instance per channel. So if you've multiple channels, there would be one for each channel.

ok, so then it's enough to create another channel which consumes from the same topic and I guess, given that each consumer instance will be assigned a specific partition, there wouldn't be no duplicates.

Why do you want to create a different consumer per retry-topic ?

for topic consumer isolation

let's say there are 3 dtr :

• "foo_retry_1000"
• "foo_retry_3000"
• "foo_retry_5000"

if one of the retry-topic consumers fails (e.g "foo_retry_3000" fails due a runtime exception) , unfortunately all the other dtr are revoked, I think this happens because the whole consumer-group is revoked.

  @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            log.executingConsumerRevokedRebalanceListener(consumerGroup);
            try {
                reactiveKafkaConsumer.setCachedConsumerGroupMetadata();
                reactiveKafkaConsumer.removeFromQueueRecordsFromTopicPartitions(partitions);
                commitHandler.partitionsRevoked(partitions);
                if (listener != null) {
                    listener.onPartitionsRevoked(reactiveKafkaConsumer.unwrap(), partitions);
                    log.executedConsumerRevokedRebalanceListener(consumerGroup);
                }
            } catch (RuntimeException e) {
                log.unableToExecuteConsumerRevokedRebalanceListener(consumerGroup, e);
                throw e;
            }
        }

of course, I think in this specific case, it makes sense to revoke all the consumers given that a part of the 'chain' is broken.

I noticed the same behaviour when I use a single channel subscribed to multiple topics

mp.messaging.incoming.<a-channel>.topic=topic1,topic2,topic3,topic4

[ozangunalp]

I don't know what can be done here without much complexity. I am converting this to a discussion