JMX Metric records-lag incorrect when topic is produced to external to consuming thread

[retinaburn]

Expected Behavior

When viewing the JMX Metric records-lag for a given topic and partition the records-lag should be somewhat comparable to the lag reported in Red Panda/Kafka while producing.

If the production of messages is faster than the consumption of messages the records-lag should increase over time.

Actual Behaviour

If producing to kafka is done via consumption thread then its reported correctly, with the records-lag increasing over time as documented in #634

If producing to kafka is done via a REST Controller in the project, then the records-lag in JMX does not increase, but the consumer lag does increase when viewed in Red Panda/Kafka.

What's even stranger, is if you pre-load the topic with messages (say 10k) and start the app, then it records the records-lag correctly.

Steps To Reproduce

1. Navigate to https://micronaut.io/launch
2. Select kafka, micrometer-jmx, micrometer-prometheus, with name: consumer, java version: 21
3. Generate Project
4. Extract demo project
5. Define Consumer:

@KafkaListener(threads = 2)
public class Consumer {
    private Logger log = LoggerFactory.getLogger(Consumer.class);

    @Inject
    Producer producer;

    @Topic("${topic.chat}")
    public void listener(ConsumerRecord<String, String> record) throws InterruptedException{
        log.debug("{}-{}:{} {}={}", record.topic(), record.partition(), record.offset(), record.key(), record.value());
        //Thread.sleep(1000);
        // producer.sendChat("1", "1");
        // producer.sendChat("2", "2");
    }
    
}

6. Define Producer Interface:

@KafkaClient(id="producer")
public interface Producer {
    @Topic("${topic.chat}")
    void sendChat(@KafkaKey String key, String value);
}

7. Define Controller:

@Controller("/produce")
public class BatchController {
    
    @Inject
    Producer producer;
    
    @Post("/batch/{numOfThreads}/{numMessages}")
    public boolean batchProduce(@PathVariable Integer numOfThreads, @PathVariable Integer numMessages){
        ExecutorService executor = Executors.newFixedThreadPool(numOfThreads);

        for (int i=0; i<numOfThreads; i++){
            ThreadProducer threadProducer = new ThreadProducer(i, producer, numMessages);
            executor.submit(threadProducer);
        }
        
        return true;
    }

    class ThreadProducer implements Runnable {
        int numOfMessages;
        String threadId;
        Producer producer;
        ThreadProducer(int threadId, Producer producer, int numOfMessages){
            this.threadId = "" + threadId;
            this.producer = producer;
            this.numOfMessages = numOfMessages;
        }

        @Override
        public void run() {
                for(int j=0; j<numOfMessages; j++){
                    producer.sendChat(threadId, ""+j);
                }   
        }
        
    }
}

8. Execute Application
9. Confirm Lag for consumer is 0 in JMX and Red Panda/Kafka
10. curl -X POST http://localhost:8082/produce/batch/10/100000
11. Confirm Lag is increasing in Red Panda/Kafka for consumer
12. Confirm jmx kafka.consumer -> consumer-fetch-manager-metrics -> consumer-consumer-11 -> chat-room -> 0 -> Attributes -> records-lag is not increasing

Environment Information

• Operating System: Windows 11
• JDK: Adoptium Temurin 21

Example Application

https://github.com/retinaburn/micronaut-kafka-investigation

Version

4.2.0