leizhiyuan opened a new issue, #1413: URL: https://github.com/apache/rocketmq-clients/issues/1413
### Before Creating the Bug Report - [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq-clients/discussions). - [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq-clients/issues) and [GitHub Discussions](https://github.com/apache/rocketmq-clients/discussions) of this repository and believe that this is not a duplicate. - [x] I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ. ### Programming Language of the Client Rust ### Runtime Platform Environment macOS 27 (arm64) ### RocketMQ Version of the Client/Server Client: Rust `rocketmq` 5.0.0, master `a3d1de8b` Server: a managed RocketMQ 5.x service (Tencent Cloud TDMQ for RocketMQ 5), gRPC proxy endpoint ### Run or Compiler Version rustc 1.98.0 ### Describe the Bug With FIFO consumption (`fifo = true`), a message whose consumption fails is retried locally by the client. During these local retries, `MessageView::delivery_attempt()` seen by the listener never changes. It stays at the value delivered by the server, typically `1`. **Scenario.** Business code often branches on the delivery attempt. It may give up and record the message after N attempts, alert on the last attempt, or switch to a degraded path. With the Rust client this logic never triggers for FIFO messages. The listener sees `1` on every retry, so the message keeps being retried until the client-side `max_attempts` (17 on our server) is exhausted and then goes to the DLQ. During all that time, every later message of the same message group (or lite topic) is blocked behind it. This affects both `PushConsumer` and `LitePushConsumer`, which share the same FIFO worker. The root cause is in `FifoConsumerWorker::consume_message_sequentially_static`. It increments a local counter but never writes it back to the `MessageView` that is passed to the listener: ```rust // rust/src/push_consumer.rs let mut delivery_attempt = message.delivery_attempt(); let max_delivery_attempts = retry_policy.get_max_attempts(); loop { let consume_result = (message_listener)(&message); // always sees the initial attempt match consume_result { ConsumeResult::SUCCESS => { ...; break; } ConsumeResult::FAILURE => { delivery_attempt += 1; // local only if delivery_attempt > max_delivery_attempts { ack_processor.forward_to_deadletter_queue(message, delivery_attempt, max_delivery_attempts).await; break; } else { tokio::time::sleep(retry_policy.get_next_attempt_delay(delivery_attempt)).await; } } } } ``` Because the counter is incremented before it is used, two more values differ from Java: - the DLQ request reports `delivery_attempt = max_attempts + 1`; - the first local retry waits for the delay of attempt 2 instead of attempt 1. **Java behavior that Rust should align with.** `ProcessQueueImpl#eraseFifoMessage`: ```java int attempt = messageView.getDeliveryAttempt(); if (ConsumeResult.FAILURE.equals(consumeResult) && attempt < maxAttempts) { final Duration nextAttemptDelay = retryPolicy.getNextAttemptDelay(attempt); attempt = messageView.incrementAndGetDeliveryAttempt(); final ListenableFuture<ConsumeResult> future = service.consume(messageView, nextAttemptDelay); ... } // otherwise forwardToDeadLetterQueue(messageView), which reports messageView.getDeliveryAttempt() ``` So, with `maxAttempts = 4`: - the Java listener sees attempts `1, 2, 3, 4`; - each retry waits for the delay of the attempt that just failed; - the DLQ request carries `delivery_attempt = 4`. ### Steps to Reproduce 1. Use a FIFO consumer group. For `LitePushConsumer`, set `PushConsumerOption::set_fifo(true)`. 2. Register a listener that records `message.delivery_attempt()` and returns `ConsumeResult::FAILURE` the first time. 3. Send one message, then a few more to the same message group / lite topic. ### What Did You Expect to See? As with the Java client: `FAIL#1`, `FAIL#2` (success), then the following messages. ### What Did You See Instead? In a listener that fails while `attempt <= 1`, the retry never succeeds. We observed `FAIL#1` five times in a row, and the following messages of the same lite topic were blocked. ### Additional Context The standard (non-FIFO) path is not affected: it redelivers through the server, which increments the attempt. --- Tracked in #1412. -- 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]
