[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

[RockteMQ-AI]

Summary

Improves pop consumer offset committing by deferring the commit until after the pop operation completes, using a pending commit mechanism. This ensures the committed offset reflects the actual nextBeginOffset of each pop round rather than lagging one round behind.

Review

• Correctness ✅ — The pending commit approach is sound: offsets are collected during handleGetMessageResult and committed after the pop future completes in commitPendingOffset. The cache-aware capping (getMinOffsetInCache) correctly prevents committing past unacknowledged cached records.
• Concurrency ✅ — The added consumerLockService.tryLock() in cleanupRecords prevents evicting records that a concurrent pop is still writing. The lock scope is appropriate.
• FIFO vs non-FIFO ✅ — FIFO path correctly commits immediately (offset = consume start point). Non-FIFO path defers via pending commits, which is the right trade-off for throughput.
• Edge cases — commitPendingOffset handles null pendingCommitList gracefully. The KeyBuilder.parseNormalTopic usage for the lock key is consistent with existing patterns.
• Tests ✅ — New PopConsumerServiceCommitOffsetTest covers the pending commit flow.

One minor note: the setPendingCommitOffset method in PopConsumerCache silently no-ops if consumerRecords is null — this is fine but worth a debug log if it becomes a diagnostic pain point.

LGTM — well-structured improvement to offset semantics with proper concurrency handling.

---

Automated review by github-manager-bot