[RIP-70-2]Delete topic queue lock

store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java

@@ -28,7 +28,6 @@
 import org.apache.rocketmq.common.attribute.CQType;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.message.MessageConst;
-import org.apache.rocketmq.common.message.MessageExtBrokerInner;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 import org.apache.rocketmq.store.config.BrokerRole;
@@ -771,17 +770,13 @@ private void doDispatchLmqQueue(DispatchRequest request, int maxRetries, String
     }
 
     @Override
-    public void assignQueueOffset(QueueOffsetOperator queueOffsetOperator, MessageExtBrokerInner msg) {
-        String topicQueueKey = getTopic() + "-" + getQueueId();
-        long queueOffset = queueOffsetOperator.getQueueOffset(topicQueueKey);
-        msg.setQueueOffset(queueOffset);
+    public void increaseQueueOffset(QueueOffsetOperator queueOffsetOperator, short messageNum) {
+        queueOffsetOperator.increaseQueueOffset(topic + "-" + queueId, messageNum);
     }
 
     @Override
-    public void increaseQueueOffset(QueueOffsetOperator queueOffsetOperator, MessageExtBrokerInner msg,
-        short messageNum) {
-        String topicQueueKey = getTopic() + "-" + getQueueId();
-        queueOffsetOperator.increaseQueueOffset(topicQueueKey, messageNum);
+    public long getQueueOffset(QueueOffsetOperator queueOffsetOperator) {
+        return queueOffsetOperator.getQueueOffset(topic + "-" + queueId);
     }
 
     private boolean putMessagePositionInfo(final long offset, final int size, final long tagsCode,

store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java

@@ -708,6 +708,7 @@ public void setBrokerInitMaxOffset(long brokerInitMaxOffset) {
         this.brokerInitMaxOffset = brokerInitMaxOffset;
     }
 
+    @Override
     public SystemClock getSystemClock() {
         return systemClock;
     }
@@ -2016,6 +2017,7 @@ public void setStateMachineVersion(long stateMachineVersion) {
         this.stateMachineVersion = stateMachineVersion;
     }
 
+    @Override
     public BrokerStatsManager getBrokerStatsManager() {
         return brokerStatsManager;
     }
@@ -2024,6 +2026,7 @@ public BrokerConfig getBrokerConfig() {
         return brokerConfig;
     }
 
+    @Override
     public int remainTransientStoreBufferNumbs() {
         if (this.isTransientStorePoolEnable()) {
             return this.transientStorePool.availableBufferNums();
@@ -2123,23 +2126,6 @@ public boolean isSyncMaster() {
         return BrokerRole.SYNC_MASTER == this.getMessageStoreConfig().getBrokerRole();
     }
 
-    @Override
-    public void assignOffset(MessageExtBrokerInner msg) throws RocksDBException {
-        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
-
-        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
-            this.consumeQueueStore.assignQueueOffset(msg);
-        }
-    }
-
-    @Override
-    public void increaseOffset(MessageExtBrokerInner msg, short messageNum) {
-        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
-
-        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
-            this.consumeQueueStore.increaseQueueOffset(msg, messageNum);
-        }
-    }
 
     public ConcurrentMap<String, TopicConfig> getTopicConfigs() {
         return this.topicConfigTable;

store/src/main/java/org/apache/rocketmq/store/MessageStore.java

@@ -742,23 +742,6 @@ void onCommitLogDispatch(DispatchRequest dispatchRequest, boolean doDispatch, Ma
      */
     boolean isSyncMaster();
 
-    /**
-     * Assign a message to queue offset. If there is a race condition, you need to lock/unlock this method
-     * yourself.
-     *
-     * @param msg        message
-     * @throws RocksDBException
-     */
-    void assignOffset(MessageExtBrokerInner msg) throws RocksDBException;
-
-    /**
-     * Increase queue offset in memory table. If there is a race condition, you need to lock/unlock this method
-     *
-     * @param msg        message
-     * @param messageNum message num
-     */
-    void increaseOffset(MessageExtBrokerInner msg, short messageNum);
-
     /**
      * Get master broker message store in process in broker container
      *

store/src/main/java/org/apache/rocketmq/store/dledger/DLedgerCommitLog.java

@@ -570,52 +570,48 @@ public CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner
 
         encodeResult = this.messageSerializer.serialize(msg);
         if (encodeResult.status != AppendMessageStatus.PUT_OK) {
-            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, new AppendMessageResult(encodeResult.status)));
+            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL,
+                new AppendMessageResult(encodeResult.status)));
         }
-
-        String topicQueueKey = msg.getTopic() + "-" + msg.getQueueId();
-        topicQueueLock.lock(topicQueueKey);
+        putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
         try {
-            defaultMessageStore.assignOffset(msg);
-
-            putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
             long elapsedTimeInLock;
             long queueOffset;
-            try {
-                beginTimeInDledgerLock = this.defaultMessageStore.getSystemClock().now();
-                queueOffset = getQueueOffsetByKey(msg, tranType);
-                encodeResult.setQueueOffsetKey(queueOffset, false);
-                AppendEntryRequest request = new AppendEntryRequest();
-                request.setGroup(dLedgerConfig.getGroup());
-                request.setRemoteId(dLedgerServer.getMemberState().getSelfId());
-                request.setBody(encodeResult.getData());
-                dledgerFuture = (AppendFuture<AppendEntryResponse>) dLedgerServer.handleAppend(request);
-                if (dledgerFuture.getPos() == -1) {
-                    return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.OS_PAGE_CACHE_BUSY, new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR)));
-                }
-                long wroteOffset = dledgerFuture.getPos() + DLedgerEntry.BODY_OFFSET;
 
-                int msgIdLength = (msg.getSysFlag() & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8 : 16 + 4 + 8;
-                ByteBuffer buffer = ByteBuffer.allocate(msgIdLength);
-
-                String msgId = MessageDecoder.createMessageId(buffer, msg.getStoreHostBytes(), wroteOffset);
-                elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
-                appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, encodeResult.getData().length, msgId, System.currentTimeMillis(), queueOffset, elapsedTimeInLock);
-            } finally {
-                beginTimeInDledgerLock = 0;
-                putMessageLock.unlock();
+            msg.setQueueOffset(this.defaultMessageStore.getQueueStore().getQueueOffset(msg.getTopic(), msg.getQueueId()));
+            beginTimeInDledgerLock = this.defaultMessageStore.getSystemClock().now();
+            queueOffset = getQueueOffsetByKey(msg, tranType);
+            encodeResult.setQueueOffsetKey(queueOffset, false);
+            AppendEntryRequest request = new AppendEntryRequest();
+            request.setGroup(dLedgerConfig.getGroup());
+            request.setRemoteId(dLedgerServer.getMemberState().getSelfId());
+            request.setBody(encodeResult.getData());
+            dledgerFuture = (AppendFuture<AppendEntryResponse>) dLedgerServer.handleAppend(request);
+            if (dledgerFuture.getPos() == -1) {
+                return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.OS_PAGE_CACHE_BUSY,
+                        new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR)));
             }
+            long wroteOffset = dledgerFuture.getPos() + DLedgerEntry.BODY_OFFSET;
+
+            int msgIdLength = (msg.getSysFlag() & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8 : 16 + 4 + 8;
+            ByteBuffer buffer = ByteBuffer.allocate(msgIdLength);
+
+            String msgId = MessageDecoder.createMessageId(buffer, msg.getStoreHostBytes(), wroteOffset);
+            elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
+            appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, encodeResult.getData().length,
+                    msgId, System.currentTimeMillis(), queueOffset, elapsedTimeInLock);
 
+            defaultMessageStore.getQueueStore().increaseQueueOffset(msg.getTopic(), msg.getQueueId(), getMessageNum(msg));
             if (elapsedTimeInLock > 500) {
-                log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, appendResult);
+                log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}",
+                    elapsedTimeInLock, msg.getBody().length, appendResult);
             }
-
-            defaultMessageStore.increaseOffset(msg, getMessageNum(msg));
-        } catch (Exception e) {
-            log.error("Put message error", e);
+        } catch (Exception ex) {
+            log.error("Put message error", ex);
             return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR)));
         } finally {
-            topicQueueLock.unlock(topicQueueKey);
+            beginTimeInDledgerLock = 0;
+            putMessageLock.unlock();
         }
 
         return dledgerFuture.thenApply(appendEntryResponse -> {
@@ -688,78 +684,73 @@ public CompletableFuture<PutMessageResult> asyncPutMessages(MessageExtBatch mess
 
         encodeResult = this.messageSerializer.serialize(messageExtBatch);
         if (encodeResult.status != AppendMessageStatus.PUT_OK) {
-            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, new AppendMessageResult(encodeResult
-                .status)));
+            return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL,
+                new AppendMessageResult(encodeResult.status)));
         }
 
-        int batchNum = encodeResult.batchData.size();
-        topicQueueLock.lock(encodeResult.queueOffsetKey);
+        putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
+
         try {
-            defaultMessageStore.assignOffset(messageExtBatch);
+            messageExtBatch.setQueueOffset(this.defaultMessageStore.getQueueStore().getQueueOffset(messageExtBatch.getTopic(), messageExtBatch.getQueueId()));
 
-            putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
             msgIdBuilder.setLength(0);
             long elapsedTimeInLock;
             long queueOffset;
             int msgNum = 0;
-            try {
-                beginTimeInDledgerLock = this.defaultMessageStore.getSystemClock().now();
-                queueOffset = getQueueOffsetByKey(messageExtBatch, tranType);
-                encodeResult.setQueueOffsetKey(queueOffset, true);
-                BatchAppendEntryRequest request = new BatchAppendEntryRequest();
-                request.setGroup(dLedgerConfig.getGroup());
-                request.setRemoteId(dLedgerServer.getMemberState().getSelfId());
-                request.setBatchMsgs(encodeResult.batchData);
-                AppendFuture<AppendEntryResponse> appendFuture = (AppendFuture<AppendEntryResponse>) dLedgerServer.handleAppend(request);
-                if (appendFuture.getPos() == -1) {
-                    log.warn("HandleAppend return false due to error code {}", appendFuture.get().getCode());
-                    return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.OS_PAGE_CACHE_BUSY, new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR)));
-                }
-                dledgerFuture = (BatchAppendFuture<AppendEntryResponse>) appendFuture;
 
-                long wroteOffset = 0;
+            beginTimeInDledgerLock = this.defaultMessageStore.getSystemClock().now();
+            queueOffset = getQueueOffsetByKey(messageExtBatch, tranType);
+            encodeResult.setQueueOffsetKey(queueOffset, true);
+            BatchAppendEntryRequest request = new BatchAppendEntryRequest();
+            request.setGroup(dLedgerConfig.getGroup());
+            request.setRemoteId(dLedgerServer.getMemberState().getSelfId());
+            request.setBatchMsgs(encodeResult.batchData);
+            AppendFuture<AppendEntryResponse> appendFuture = (AppendFuture<AppendEntryResponse>) dLedgerServer.handleAppend(request);
+            if (appendFuture.getPos() == -1) {
+                log.warn("HandleAppend return false due to error code {}", appendFuture.get().getCode());
+                return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.OS_PAGE_CACHE_BUSY,
+                    new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR)));
+            }
+            dledgerFuture = (BatchAppendFuture<AppendEntryResponse>) appendFuture;
+
+            long wroteOffset = 0;
 
-                int msgIdLength = (messageExtBatch.getSysFlag() & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8 : 16 + 4 + 8;
-                ByteBuffer buffer = ByteBuffer.allocate(msgIdLength);
+            int msgIdLength = (messageExtBatch.getSysFlag() & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 + 8 : 16 + 4 + 8;
+            ByteBuffer buffer = ByteBuffer.allocate(msgIdLength);
 
-                boolean isFirstOffset = true;
-                long firstWroteOffset = 0;
-                for (long pos : dledgerFuture.getPositions()) {
-                    wroteOffset = pos + DLedgerEntry.BODY_OFFSET;
-                    if (isFirstOffset) {
-                        firstWroteOffset = wroteOffset;
-                        isFirstOffset = false;
-                    }
-                    String msgId = MessageDecoder.createMessageId(buffer, messageExtBatch.getStoreHostBytes(), wroteOffset);
-                    if (msgIdBuilder.length() > 0) {
-                        msgIdBuilder.append(',').append(msgId);
-                    } else {
-                        msgIdBuilder.append(msgId);
-                    }
-                    msgNum++;
+            boolean isFirstOffset = true;
+            long firstWroteOffset = 0;
+            for (long pos : dledgerFuture.getPositions()) {
+                wroteOffset = pos + DLedgerEntry.BODY_OFFSET;
+                if (isFirstOffset) {
+                    firstWroteOffset = wroteOffset;
+                    isFirstOffset = false;
+                }
+                String msgId = MessageDecoder.createMessageId(buffer, messageExtBatch.getStoreHostBytes(), wroteOffset);
+                if (msgIdBuilder.length() > 0) {
+                    msgIdBuilder.append(',').append(msgId);
+                } else {
+                    msgIdBuilder.append(msgId);
                 }
+                msgNum++;
+            }
 
-                elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
-                appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, firstWroteOffset, encodeResult.totalMsgLen,
+            elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginTimeInDledgerLock;
+            appendResult = new AppendMessageResult(AppendMessageStatus.PUT_OK, firstWroteOffset, encodeResult.totalMsgLen,
                     msgIdBuilder.toString(), System.currentTimeMillis(), queueOffset, elapsedTimeInLock);
-                appendResult.setMsgNum(msgNum);
-            } finally {
-                beginTimeInDledgerLock = 0;
-                putMessageLock.unlock();
-            }
+            appendResult.setMsgNum(msgNum);
 
+            defaultMessageStore.getQueueStore().increaseQueueOffset(messageExtBatch.getTopic(), messageExtBatch.getQueueId(), (short) encodeResult.batchData.size());
             if (elapsedTimeInLock > 500) {
                 log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}",
-                    elapsedTimeInLock, messageExtBatch.getBody().length, appendResult);
+                        elapsedTimeInLock, messageExtBatch.getBody().length, appendResult);
             }
-
-            defaultMessageStore.increaseOffset(messageExtBatch, (short) batchNum);
-
-        } catch (Exception e) {
-            log.error("Put message error", e);
+        } catch (Exception re) {
+            log.error("Put message error", re);
             return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR)));
-        }  finally {
-            topicQueueLock.unlock(encodeResult.queueOffsetKey);
+        } finally {
+            beginTimeInDledgerLock = 0;
+            putMessageLock.unlock();
         }
 
         return dledgerFuture.thenApply(appendEntryResponse -> {

store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java

@@ -588,16 +588,6 @@ public boolean isSyncMaster() {
         return next.isSyncMaster();
     }
 
-    @Override
-    public void assignOffset(MessageExtBrokerInner msg) throws RocksDBException {
-        next.assignOffset(msg);
-    }
-
-    @Override
-    public void increaseOffset(MessageExtBrokerInner msg, short messageNum) {
-        next.increaseOffset(msg, messageNum);
-    }
-
     @Override
     public List<PutMessageHook> getPutMessageHookList() {
         return next.getPutMessageHookList();