ni-ze commented on code in PR #6641:
URL: https://github.com/apache/rocketmq/pull/6641#discussion_r1177279061
##########
store/src/main/java/org/apache/rocketmq/store/queue/QueueOffsetOperator.java:
##########
@@ -28,47 +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);
}
public void updateQueueOffset(String topicQueueKey, long offset) {
this.topicQueueTable.put(topicQueueKey, offset);
}
Review Comment:
updateQueueOffset seems not be used.
--
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]
