[GitHub] [rocketmq] caigy commented on a diff in pull request #6641: [ISSUE #6570] Fix the issue that expectLogicOffset is greater than currentLogicOffset in consumeQueue build when the message is illegal

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

caigy commented on code in PR #6641:
URL: https://github.com/apache/rocketmq/pull/6641#discussion_r1175944960


##########
store/src/main/java/org/apache/rocketmq/store/queue/QueueOffsetOperator.java:
##########
@@ -28,39 +28,49 @@
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 
 /**
- * QueueOffsetAssigner is a component for assigning offsets for queues.
+ * QueueOffsetOperator is a component for operating offsets for queues.
  */
-public class QueueOffsetAssigner {
+public class QueueOffsetOperator {
     private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
 
     private ConcurrentMap<String, Long> topicQueueTable = new ConcurrentHashMap<>(1024);
     private ConcurrentMap<String, Long> batchTopicQueueTable = new ConcurrentHashMap<>(1024);
     private ConcurrentMap<String/* topic-queueid */, Long/* offset */> lmqTopicQueueTable = new ConcurrentHashMap<>(1024);
 
-    public long assignQueueOffset(String topicQueueKey, short messageNum) {
+    public long getQueueOffset(String topicQueueKey) {
+        return ConcurrentHashMapUtils.computeIfAbsent(this.topicQueueTable, topicQueueKey, k -> 0L);
+    }
+
+    public void increaseQueueOffset(String topicQueueKey, short messageNum) {
         Long queueOffset = ConcurrentHashMapUtils.computeIfAbsent(this.topicQueueTable, topicQueueKey, k -> 0L);
-        this.topicQueueTable.put(topicQueueKey, queueOffset + messageNum);
-        return queueOffset;
+        topicQueueTable.put(topicQueueKey, queueOffset + messageNum);

Review Comment:
   Should we use `replace(key, oldVal, newVal)` for atomicity?



-- 
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

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