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

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

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

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.