[ISSUE #10949] Improve pop consumer offset committing

[redlsz]

Which Issue(s) This PR Fixes

• Fixes [Enhancement] Converge pop consumer offset in the background so it does not depend on client polling #10949

Brief Description

• PopConsumerService: each pop now records its nextBeginOffset as a pending commit and commits it after the pop completes, capped at minOffsetInCache while records are still buffered, and skipped on offset reset or backward movement. FIFO path unchanged.
• PopConsumerCache: keeps the pending offset on the queue's cache entry and commits it in cleanupRecords once no in-flight record remains (all acked or persisted).
• PopConsumerCache#cleanupRecords: the timeout-eviction path now holds the same lock as popAsync, so a concurrent pop's records can't be evicted mid-write.

How Did You Test This Change?

Minimum verification test. Check whether the consumer offsets have been advanced.

public class SimpleProducerConsumerExample {
    private static final Logger log = LoggerFactory.getLogger(CustomConsoleAppender.class);

    public static void main(String[] args) throws Exception {
        String endpoints = "foobar.com:8080";
        String accessKey = "yourAccessKey";
        String secretKey = "yourSecretKey";
        String topic = "yourTopic";
        String consumerGroup = "yourConsumerGroup";

        SessionCredentialsProvider credentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey);
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .enableSsl(false)
            .setCredentialProvider(credentialsProvider)
            .build();

        ClientServiceProvider provider = ClientServiceProvider.loadService();

        // Send one normal message
        try (Producer producer = provider.newProducerBuilder()
            .setClientConfiguration(clientConfiguration)
            .setTopics(topic)
            .build()) {

            Message message = provider.newMessageBuilder()
                .setTopic(topic)
                .setBody("Hello, RocketMQ!".getBytes(StandardCharsets.UTF_8))
                .build();
            SendReceipt receipt = producer.send(message);
            log.info("Message sent, messageId={}", receipt.getMessageId());
        }

        // Receive the message and ack it
        try (SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            .setConsumerGroup(consumerGroup)
            .setAwaitDuration(Duration.ofSeconds(5))
            .setSubscriptionExpressions(Collections.singletonMap(topic, new FilterExpression("*", FilterExpressionType.TAG)))
            .build()) {

            List<MessageView> messages = consumer.receive(1, Duration.ofSeconds(15));
            if (messages != null && !messages.isEmpty()) {
                MessageView message = messages.get(0);
                consumer.ack(message);
                log.info("Message received and acked, messageId={}", message.getMessageId());
            }
        }
    }
}

[codecov-commenter]

Codecov Report

❌ Patch coverage is 93.24324% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.54%. Comparing base (efa1be5) to head (59a5923).
⚠️ Report is 1 commits behind head on develop.

Files with missing lines	Patch %	Lines
...g/apache/rocketmq/broker/pop/PopConsumerCache.java	91.66%	0 Missing and 3 partials ⚠️
...apache/rocketmq/broker/pop/PopConsumerContext.java	92.30%	0 Missing and 1 partial ⚠️
...apache/rocketmq/broker/pop/PopConsumerService.java	96.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10964      +/-   ##
=============================================
- Coverage      48.58%   48.54%   -0.05%     
- Complexity     13671    13672       +1     
=============================================
  Files           1381     1381              
  Lines         101475   101534      +59     
  Branches       13189    13198       +9     
=============================================
- Hits           49302    49290      -12     
- Misses         46176    46235      +59     
- Partials        5997     6009      +12     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[RockteMQ-AI]

Summary

This PR improves the pop consumer offset committing mechanism by introducing a pending commit offset pattern that ensures offsets are committed only after pop operations complete, and adds lock acquisition in cleanupRecords to prevent concurrent eviction during pop writes.

The changes are well-structured and address a real concurrency issue. The use of volatile for pendingCommitOffset and CAS-based advancePendingCommitOffset ensures thread safety. The lock acquisition in cleanupRecords (matching the popAsync lock) correctly prevents race conditions.

LGTM — no blocking issues found.

---

Automated review by github-manager-bot