redlsz opened a new pull request, #10964: URL: https://github.com/apache/rocketmq/pull/10964
<!-- Please make sure the target branch is right. In most case, the target branch should be `develop`. --> ### Which Issue(s) This PR Fixes <!-- Please ensure that the related issue has already been created, and [link this pull request to that issue using keywords](<https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword>) to ensure automatic closure. --> - Fixes #10949 ### Brief Description <!-- Write a brief description for your pull request to help the maintainer understand the reasons behind your changes. --> - **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? <!-- In order to ensure the code quality of Apache RocketMQ, we expect every pull request to have undergone thorough testing. --> 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()); } } } } ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
