[I] [Bug] RandomAsyncCommit example: NoSuchElementException on empty cache and lost messages under concurrent access [rocketmq]

["chennaji9 (via GitHub)" <gi...@apache.org>]

chennaji9 opened a new issue, #11281:
URL: https://github.com/apache/rocketmq/issues/11281

   ### Description
   
   `example/src/main/java/org/apache/rocketmq/example/simple/RandomAsyncCommit.java` (the lite-pull random-commit example) has two defects.
   
   **1. `commitableOffset` throws `NoSuchElementException` once every cached message has been consumed**
   
   ```java
   public long commitableOffset(final MessageQueue mq) {
       CachedQueue cachedQueue = this.mqCachedTable.get(mq);
       if (null != cachedQueue) {
           return cachedQueue.getMsgCachedTable().firstKey();
       }
       return -1;
   }
   ```
   
   `getMsgCachedTable()` is a `TreeMap<Long, MessageExt>`. After `putMessages` adds a batch and `removeMessage` removes the last entry (all messages consumed), the `CachedQueue` object still exists but its TreeMap is empty, and `firstKey()` throws `NoSuchElementException`. The same state is reached by `putMessages(mq, emptyList)`. The `-1` sentinel is only returned for an unknown queue, not for an empty cache.
   
   **2. Concurrent first access can silently lose a cached batch**
   
   ```java
   public void putMessages(final MessageQueue mq, final List<MessageExt> msgs) {
       CachedQueue cachedQueue = this.mqCachedTable.get(mq);
       if (null == cachedQueue) {
           cachedQueue = new CachedQueue();
           this.mqCachedTable.put(mq, cachedQueue);
       }
       ...
   }
   ```
   
   The map is a `ConcurrentHashMap` but initialization is get-then-put. Two threads consuming/committing the same queue for the first time can each construct a `CachedQueue`; the second `put` overwrites the first, so the batch cached by the first thread disappears from `commitableOffset`'s view. Additionally the inner `TreeMap` is read and written from `putMessages`, `removeMessage` and `commitableOffset` without any synchronization.
   
   ### Expected behavior
   
   - `commitableOffset` returns the `-1` sentinel when there is nothing committable (empty cache), instead of throwing.
   - Concurrent use of the example class must not lose cached messages: initialization should be atomic (e.g. `computeIfAbsent`) and the three public methods should be safe against concurrent TreeMap mutation.


-- 
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: commits-unsubscribe@rocketmq.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org