How to manual commit do not recorverd offset already sent DLT through CommonErrorHandler

A simple example is currently being made through the spring kafka.

If an exception occurs at the service layer, I want to commit the original offset after trying to retry and loading it into the dead letter queue.

However, the dead letter queue is loaded properly, but the original message remains in the kafka because the commit is not processed.

To show you my code, it is as follows.

• KafkaConfig.java

...
 @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setCommonErrorHandler(kafkaListenerErrorHandler());
        factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
        return factory;
    }

    private CommonErrorHandler kafkaListenerErrorHandler() {
        DefaultErrorHandler defaultErrorHandler = new DefaultErrorHandler(
            new DeadLetterPublishingRecoverer(template, DEAD_TOPIC_DESTINATION_RESOLVER),
            new FixedBackOff(1000, 3));
       
        defaultErrorHandler.setCommitRecovered(true);
        defaultErrorHandler.setAckAfterHandle(true);
        defaultErrorHandler.setResetStateOnRecoveryFailure(false);

        
        return defaultErrorHandler;
    }
...

• KafkaListener.java

...
@KafkaListener(topics = TOPIC_NAME, containerFactory = "kafkaListenerContainerFactory", groupId = "stock-adjustment-0")
    public void subscribe(final String message, Acknowledgment ack) throws IOException {

        log.info(String.format("Message Received : [%s]", message));

        StockAdjustment stockAdjustment = StockAdjustment.deserializeJSON(message);

        if(stockService.isAlreadyProcessedOrderId(stockAdjustment.getOrderId())) {
            log.info(String.format("AlreadyProcessedOrderId : [%s]", stockAdjustment.getOrderId()));
        } else {
            if(stockAdjustment.getAdjustmentType().equals("REDUCE")) {
                stockService.decreaseStock(stockAdjustment);
            }
        }
    ack.acknowledge(); // <<< does not work!
    }
...

• Stockservice.java

...
        if(stockAdjustment.getQty() > stock.getAvailableStockQty()) {
            throw new RuntimeException(String.format("Stock decreased Request [decreasedQty: %s][availableQty : %s]", stockAdjustment.getQty(), stock.getAvailableStockQty()));
        }
...

At this time, when RuntimeException occur in the service layer as above, the DLT is issued through an CommonErrorhandler according to the Kafka setting.

However, after issuing the DLT, the original message remains in Kafka, so there is a need for a solution.

I looked it up and found that the setting I wrote is processed through SeekUtils.seekOrRecover(), and if it is not processed even if the maximum number of attempts is not processed, an exception occurs and the original offset is rolled back without processing a commit.

According to the document, it seems that the AfterRollbackProcessor handles rollback if it fails with the default value, but I don't know how to write the code to commit even if it fails.

EDITED

The above code and settings work normally.

I thought the consumer lag would occur, but when I judged the actual logic code(SeekUtils.seekOrRecover()) and checked the offset commit and lag, I confirmed that it works normally.

I think it was caused by my mistake.