path: root/patch007-backport-fix-some-bugs.patch

From 90c5382aee07879a80309f257f04114201ccaac6 Mon Sep 17 00:00:00 2001
From: ShuangxiDing <dingshuangxi888@gmail.com>
Date: Fri, 21 Jul 2023 20:28:58 +0800
Subject: [PATCH 01/10] [ISSUE #7061] Support forward HAProxyMessage for Multi
 Protocol server. (#7062)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* Support dynamic modification of grpc tls mode to improve the scalability of ProtocolNegotiator

* Support dynamic modification of grpc tls mode to improve the scalability of ProtocolNegotiator

* [ISSUE #6866] Move the judgment logic of grpc TLS mode to improve the scalability of ProtocolNegotiator

* [ISSUE #6866] Move the judgment logic of grpc TLS mode to improve the scalability of ProtocolNegotiator

* [ISSUE #6866] Move the judgment logic of grpc TLS mode to improve the scalability of ProtocolNegotiator

* [ISSUE #6866] Move the judgment logic of grpc TLS mode to improve the scalability of ProtocolNegotiator

* Support proxy protocol for gRPC server.

* Support proxy protocol for gRPC server.

* Support proxy protocol for gRPC server.

* Support proxy protocol for gRPC server.

* Support proxy protocol for gRPC server.

* Support proxy protocol for gRPC and Remoting server.

* 回滚netty的升级

* Support proxy protocol for gRPC and Remoting server.

* Support proxy protocol for gRPC and Remoting server.

* Support proxy protocol for gRPC and Remoting server.

* add grpc-netty-codec-haproxy in bazel

* add grpc-netty-codec-haproxy in bazel

* Support proxy protocol for gRPC and Remoting server.

* Fix Test

* add grpc-netty-codec-haproxy in bazel

* add ProxyProtocolTest for Remoting

* Support HAProxyMessage forward for multi protocol server.

---------

Co-authored-by: 徒钟 <shuangxi.dsx@alibaba-inc.com>
---
 .../http2proxy/HAProxyMessageForwarder.java   | 129 ++++++++++++++++++
 .../http2proxy/Http2ProtocolProxyHandler.java |  23 +++-
 .../http2proxy/Http2ProxyBackendHandler.java  |   2 +
 .../http2proxy/Http2ProxyFrontendHandler.java |  28 ++--
 4 files changed, 164 insertions(+), 18 deletions(-)
 create mode 100644 proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/HAProxyMessageForwarder.java

diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/HAProxyMessageForwarder.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/HAProxyMessageForwarder.java
new file mode 100644
index 000000000..8f139d3d9
--- /dev/null
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/HAProxyMessageForwarder.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.apache.rocketmq.proxy.remoting.protocol.http2proxy;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.haproxy.HAProxyCommand;
+import io.netty.handler.codec.haproxy.HAProxyMessage;
+import io.netty.handler.codec.haproxy.HAProxyProtocolVersion;
+import io.netty.handler.codec.haproxy.HAProxyProxiedProtocol;
+import io.netty.handler.codec.haproxy.HAProxyTLV;
+import io.netty.util.Attribute;
+import io.netty.util.DefaultAttributeMap;
+import org.apache.commons.codec.binary.Hex;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.rocketmq.acl.common.AclUtils;
+import org.apache.rocketmq.common.constant.HAProxyConstants;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.remoting.netty.AttributeKeys;
+
+import java.lang.reflect.Field;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
+
+public class HAProxyMessageForwarder extends ChannelInboundHandlerAdapter {
+
+    private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);
+
+    private static final Field FIELD_ATTRIBUTE =
+            FieldUtils.getField(DefaultAttributeMap.class, "attributes", true);
+
+    private final Channel outboundChannel;
+
+    public HAProxyMessageForwarder(final Channel outboundChannel) {
+        this.outboundChannel = outboundChannel;
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        try {
+            forwardHAProxyMessage(ctx.channel(), outboundChannel);
+            ctx.fireChannelRead(msg);
+        } catch (Exception e) {
+            log.error("Forward HAProxyMessage from Remoting to gRPC server error.", e);
+            throw e;
+        } finally {
+            ctx.pipeline().remove(this);
+        }
+    }
+
+    private void forwardHAProxyMessage(Channel inboundChannel, Channel outboundChannel) throws Exception {
+        if (!inboundChannel.hasAttr(AttributeKeys.PROXY_PROTOCOL_ADDR)) {
+            return;
+        }
+
+        if (!(inboundChannel instanceof DefaultAttributeMap)) {
+            return;
+        }
+
+        Attribute<?>[] attributes = (Attribute<?>[]) FieldUtils.readField(FIELD_ATTRIBUTE, inboundChannel);
+        if (ArrayUtils.isEmpty(attributes)) {
+            return;
+        }
+
+        String sourceAddress = null, destinationAddress = null;
+        int sourcePort = 0, destinationPort = 0;
+        List<HAProxyTLV> haProxyTLVs = new ArrayList<>();
+
+        for (Attribute<?> attribute : attributes) {
+            String attributeKey = attribute.key().name();
+            if (!StringUtils.startsWith(attributeKey, HAProxyConstants.PROXY_PROTOCOL_PREFIX)) {
+                continue;
+            }
+            String attributeValue = (String) attribute.get();
+            if (StringUtils.isEmpty(attributeValue)) {
+                continue;
+            }
+            if (attribute.key() == AttributeKeys.PROXY_PROTOCOL_ADDR) {
+                sourceAddress = attributeValue;
+            }
+            if (attribute.key() == AttributeKeys.PROXY_PROTOCOL_PORT) {
+                sourcePort = Integer.parseInt(attributeValue);
+            }
+            if (attribute.key() == AttributeKeys.PROXY_PROTOCOL_SERVER_ADDR) {
+                destinationAddress = attributeValue;
+            }
+            if (attribute.key() == AttributeKeys.PROXY_PROTOCOL_SERVER_PORT) {
+                destinationPort = Integer.parseInt(attributeValue);
+            }
+            if (StringUtils.startsWith(attributeKey, HAProxyConstants.PROXY_PROTOCOL_TLV_PREFIX)) {
+                String typeString = StringUtils.substringAfter(attributeKey, HAProxyConstants.PROXY_PROTOCOL_TLV_PREFIX);
+                ByteBuf byteBuf = Unpooled.buffer();
+                byteBuf.writeBytes(attributeValue.getBytes(Charset.defaultCharset()));
+                HAProxyTLV haProxyTLV = new HAProxyTLV(Hex.decodeHex(typeString)[0], byteBuf);
+                haProxyTLVs.add(haProxyTLV);
+            }
+        }
+
+        HAProxyProxiedProtocol proxiedProtocol = AclUtils.isColon(sourceAddress) ? HAProxyProxiedProtocol.TCP6 :
+                HAProxyProxiedProtocol.TCP4;
+
+        HAProxyMessage message = new HAProxyMessage(HAProxyProtocolVersion.V2, HAProxyCommand.PROXY,
+                proxiedProtocol, sourceAddress, destinationAddress, sourcePort, destinationPort, haProxyTLVs);
+        outboundChannel.writeAndFlush(message).sync();
+    }
+}
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProtocolProxyHandler.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProtocolProxyHandler.java
index 913f35c93..c37db92af 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProtocolProxyHandler.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProtocolProxyHandler.java
@@ -24,13 +24,14 @@ import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
+import io.netty.handler.codec.haproxy.HAProxyMessageEncoder;
 import io.netty.handler.ssl.ApplicationProtocolConfig;
 import io.netty.handler.ssl.ApplicationProtocolNames;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.ssl.SslProvider;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
-import javax.net.ssl.SSLException;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
@@ -38,8 +39,11 @@ import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.ProxyConfig;
 import org.apache.rocketmq.proxy.remoting.protocol.ProtocolHandler;
 import org.apache.rocketmq.remoting.common.TlsMode;
+import org.apache.rocketmq.remoting.netty.AttributeKeys;
 import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
 
+import javax.net.ssl.SSLException;
+
 public class Http2ProtocolProxyHandler implements ProtocolHandler {
     private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);
     private static final String LOCAL_HOST = "127.0.0.1";
@@ -101,11 +105,8 @@ public class Http2ProtocolProxyHandler implements ProtocolHandler {
             .handler(new ChannelInitializer<Channel>() {
                 @Override
                 protected void initChannel(Channel ch) throws Exception {
-                    if (sslContext != null) {
-                        ch.pipeline()
-                            .addLast(sslContext.newHandler(ch.alloc(), LOCAL_HOST, config.getGrpcServerPort()));
-                    }
-                    ch.pipeline().addLast(new Http2ProxyBackendHandler(inboundChannel));
+                    ch.pipeline().addLast(null, Http2ProxyBackendHandler.HANDLER_NAME,
+                            new Http2ProxyBackendHandler(inboundChannel));
                 }
             })
             .option(ChannelOption.AUTO_READ, false)
@@ -120,7 +121,15 @@ public class Http2ProtocolProxyHandler implements ProtocolHandler {
         }
 
         final Channel outboundChannel = f.channel();
+        if (inboundChannel.hasAttr(AttributeKeys.PROXY_PROTOCOL_ADDR)) {
+            ctx.pipeline().addLast(new HAProxyMessageForwarder(outboundChannel));
+            outboundChannel.pipeline().addFirst(HAProxyMessageEncoder.INSTANCE);
+        }
 
-        ctx.pipeline().addLast(new Http2ProxyFrontendHandler(outboundChannel));
+        SslHandler sslHandler = null;
+        if (sslContext != null) {
+            sslHandler = sslContext.newHandler(outboundChannel.alloc(), LOCAL_HOST, config.getGrpcServerPort());
+        }
+        ctx.pipeline().addLast(new Http2ProxyFrontendHandler(outboundChannel, sslHandler));
     }
 }
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProxyBackendHandler.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProxyBackendHandler.java
index 0195b0c1c..fd5408fae 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProxyBackendHandler.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProxyBackendHandler.java
@@ -29,6 +29,8 @@ import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 public class Http2ProxyBackendHandler extends ChannelInboundHandlerAdapter {
     private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);
 
+    public static final String HANDLER_NAME = "Http2ProxyBackendHandler";
+
     private final Channel inboundChannel;
 
     public Http2ProxyBackendHandler(Channel inboundChannel) {
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProxyFrontendHandler.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProxyFrontendHandler.java
index 87147a322..9b37e85e5 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProxyFrontendHandler.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/protocol/http2proxy/Http2ProxyFrontendHandler.java
@@ -19,36 +19,42 @@ package org.apache.rocketmq.proxy.remoting.protocol.http2proxy;
 
 import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.ssl.SslHandler;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 
 public class Http2ProxyFrontendHandler extends ChannelInboundHandlerAdapter {
     private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);
+
+    public static final String HANDLER_NAME = "SslHandler";
+
     // As we use inboundChannel.eventLoop() when building the Bootstrap this does not need to be volatile as
     // the outboundChannel will use the same EventLoop (and therefore Thread) as the inboundChannel.
     private final Channel outboundChannel;
+    private final SslHandler sslHandler;
 
-    public Http2ProxyFrontendHandler(final Channel outboundChannel) {
+    public Http2ProxyFrontendHandler(final Channel outboundChannel, final SslHandler sslHandler) {
         this.outboundChannel = outboundChannel;
+        this.sslHandler = sslHandler;
     }
 
     @Override
     public void channelRead(final ChannelHandlerContext ctx, Object msg) {
         if (outboundChannel.isActive()) {
-            outboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
-                @Override
-                public void operationComplete(ChannelFuture future) {
-                    if (future.isSuccess()) {
-                        // was able to flush out data, start to read the next chunk
-                        ctx.channel().read();
-                    } else {
-                        future.channel().close();
-                    }
+            if (sslHandler != null && outboundChannel.pipeline().get(HANDLER_NAME) == null) {
+                outboundChannel.pipeline().addBefore(Http2ProxyBackendHandler.HANDLER_NAME, HANDLER_NAME, sslHandler);
+            }
+
+            outboundChannel.writeAndFlush(msg).addListener((ChannelFutureListener) future -> {
+                if (future.isSuccess()) {
+                    // was able to flush out data, start to read the next chunk
+                    ctx.channel().read();
+                } else {
+                    future.channel().close();
                 }
             });
         }
-- 
2.32.0.windows.2


From 8027cfc7cbb6c120d2fc045e0caa8debe1028a31 Mon Sep 17 00:00:00 2001
From: maclong1989 <814742806@qq.com>
Date: Sun, 23 Jul 2023 09:15:05 +0800
Subject: [PATCH 02/10] [ISSUE #7063] doc: fix typo in user_guide.md

Signed-off-by: jiangyl3 <jiangyl3@asiainfo.com>
Co-authored-by: jiangyl3 <jiangyl3@asiainfo.com>
---
 docs/cn/msg_trace/user_guide.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/cn/msg_trace/user_guide.md b/docs/cn/msg_trace/user_guide.md
index d8314052b..9cf139fd3 100644
--- a/docs/cn/msg_trace/user_guide.md
+++ b/docs/cn/msg_trace/user_guide.md
@@ -35,7 +35,7 @@ namesrvAddr=XX.XX.XX.XX:9876
 RocketMQ集群中每一个Broker节点均用于存储Client端收集并发送过来的消息轨迹数据。因此，对于RocketMQ集群中的Broker节点数量并无要求和限制。
 
 ### 2.3 物理IO隔离模式
-对于消息轨迹数据量较大的场景，可以在RocketMQ集群中选择其中一个Broker节点专用于存储消息轨迹，使得用户普通的消息数据与消息轨迹数据的物理IO完全隔离，互不影响。在该模式下，RockeMQ集群中至少有两个Broker节点，其中一个Broker节点定义为存储消息轨迹数据的服务端。
+对于消息轨迹数据量较大的场景，可以在RocketMQ集群中选择其中一个Broker节点专用于存储消息轨迹，使得用户普通的消息数据与消息轨迹数据的物理IO完全隔离，互不影响。在该模式下，RocketMQ集群中至少有两个Broker节点，其中一个Broker节点定义为存储消息轨迹数据的服务端。
 
 ### 2.4 启动开启消息轨迹的Broker
 `nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties &`
-- 
2.32.0.windows.2


From 3102758487f3e21e977424d7f1b7187eb6c069cb Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=90=B4=E6=98=9F=E7=81=BF?=
 <37405937+wuyoudexiao@users.noreply.github.com>
Date: Tue, 25 Jul 2023 13:47:53 +0800
Subject: [PATCH 03/10] fix: npe in lockBatchMQ and unlockBatchMQ (#7078)

Co-authored-by: wxc <wuxingcan666@foxmail.com>
---
 .../rocketmq/proxy/processor/ConsumerProcessor.java   | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java
index cc973813b..656a6339d 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java
@@ -19,6 +19,7 @@ package org.apache.rocketmq.proxy.processor;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -425,13 +426,15 @@ public class ConsumerProcessor extends AbstractProcessor {
     }
 
     protected Set<AddressableMessageQueue> buildAddressableSet(ProxyContext ctx, Set<MessageQueue> mqSet) {
-        return mqSet.stream().map(mq -> {
+        Set<AddressableMessageQueue> addressableMessageQueueSet = new HashSet<>(mqSet.size());
+        for (MessageQueue mq:mqSet) {
             try {
-                return serviceManager.getTopicRouteService().buildAddressableMessageQueue(ctx, mq);
+                addressableMessageQueueSet.add(serviceManager.getTopicRouteService().buildAddressableMessageQueue(ctx, mq)) ;
             } catch (Exception e) {
-                return null;
+                log.error("build addressable message queue fail, messageQueue = {}", mq, e);
             }
-        }).collect(Collectors.toSet());
+        }
+        return addressableMessageQueueSet;
     }
 
     protected HashMap<String, List<AddressableMessageQueue>> buildAddressableMapByBrokerName(
-- 
2.32.0.windows.2


From 047ef7498f2203a2234052603a99a114d8a65e17 Mon Sep 17 00:00:00 2001
From: rongtong <jinrongtong5@163.com>
Date: Tue, 25 Jul 2023 14:00:22 +0800
Subject: [PATCH 04/10] Ensuring consistency between broker and nameserver data
 when deleting a topic (#7066)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: 尘央 <xinyuzhou.zxy@alibaba-inc.com>
---
 .../rocketmq/broker/BrokerController.java     |  11 ++
 .../rocketmq/broker/out/BrokerOuterAPI.java   |  62 ++++++++++
 .../processor/AdminBrokerProcessor.java       |  26 ++--
 .../broker/topic/TopicConfigManager.java      |   6 +-
 .../apache/rocketmq/common/BrokerConfig.java  |  14 +++
 .../common/namesrv/NamesrvConfig.java         |  17 +++
 .../namesrv/routeinfo/RouteInfoManager.java   |  64 ++++++++--
 .../routeinfo/RouteInfoManagerNewTest.java    |  99 +++++++++++++++
 .../rocketmq/test/util/MQAdminTestUtils.java  |  37 ++++++
 .../dledger/DLedgerProduceAndConsumeIT.java   |   2 +-
 .../test/route/CreateAndUpdateTopicIT.java    | 114 ++++++++++++++++++
 11 files changed, 429 insertions(+), 23 deletions(-)
 rename test/src/test/java/org/apache/rocketmq/test/{base => }/dledger/DLedgerProduceAndConsumeIT.java (99%)
 create mode 100644 test/src/test/java/org/apache/rocketmq/test/route/CreateAndUpdateTopicIT.java

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
index 196401e26..972457194 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
@@ -1678,6 +1678,17 @@ public class BrokerController {
         }, 1000, brokerConfig.getBrokerHeartbeatInterval(), TimeUnit.MILLISECONDS));
     }
 
+    public synchronized void registerSingleTopicAll(final TopicConfig topicConfig) {
+        TopicConfig tmpTopic = topicConfig;
+        if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
+            || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
+            // Copy the topic config and modify the perm
+            tmpTopic = new TopicConfig(topicConfig);
+            tmpTopic.setPerm(topicConfig.getPerm() & this.brokerConfig.getBrokerPermission());
+        }
+        this.brokerOuterAPI.registerSingleTopicAll(this.brokerConfig.getBrokerName(), tmpTopic, 3000);
+    }
+
     public synchronized void registerIncrementBrokerData(TopicConfig topicConfig, DataVersion dataVersion) {
         this.registerIncrementBrokerData(Collections.singletonList(topicConfig), dataVersion);
     }
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
index b6273e9ed..1793a83c0 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java
@@ -42,6 +42,7 @@ import org.apache.rocketmq.common.LockCallback;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.UnlockCallback;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.LoggerName;
@@ -120,12 +121,14 @@ import org.apache.rocketmq.remoting.protocol.header.namesrv.QueryDataVersionRequ
 import org.apache.rocketmq.remoting.protocol.header.namesrv.QueryDataVersionResponseHeader;
 import org.apache.rocketmq.remoting.protocol.header.namesrv.RegisterBrokerRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.namesrv.RegisterBrokerResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.namesrv.RegisterTopicRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.namesrv.UnRegisterBrokerRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
 import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
 import org.apache.rocketmq.remoting.protocol.namesrv.RegisterBrokerResult;
 import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.QueueData;
 import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.remoting.rpc.ClientMetadata;
 import org.apache.rocketmq.remoting.rpc.RpcClient;
@@ -614,6 +617,65 @@ public class BrokerOuterAPI {
         throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr);
     }
 
+    /**
+     * Register the topic route info of single topic to all name server nodes.
+     * This method is used to replace incremental broker registration feature.
+     */
+    public void registerSingleTopicAll(
+        final String brokerName,
+        final TopicConfig topicConfig,
+        final int timeoutMills) {
+        String topic = topicConfig.getTopicName();
+        RegisterTopicRequestHeader requestHeader = new RegisterTopicRequestHeader();
+        requestHeader.setTopic(topic);
+
+        TopicRouteData topicRouteData = new TopicRouteData();
+        List<QueueData> queueDatas = new ArrayList<>();
+        topicRouteData.setQueueDatas(queueDatas);
+
+        final QueueData queueData = new QueueData();
+        queueData.setBrokerName(brokerName);
+        queueData.setPerm(topicConfig.getPerm());
+        queueData.setReadQueueNums(topicConfig.getReadQueueNums());
+        queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
+        queueData.setTopicSysFlag(topicConfig.getTopicSysFlag());
+        queueDatas.add(queueData);
+        final byte[] topicRouteBody = topicRouteData.encode();
+
+
+        List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
+        final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
+        for (final String namesrvAddr : nameServerAddressList) {
+            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_TOPIC_IN_NAMESRV, requestHeader);
+            request.setBody(topicRouteBody);
+
+            try {
+                brokerOuterExecutor.execute(() -> {
+                    try {
+                        RemotingCommand response = BrokerOuterAPI.this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);
+                        assert response != null;
+                        LOGGER.info("Register single topic %s to broker %s with response code %s", topic, brokerName, response.getCode());
+                    } catch (Exception e) {
+                        LOGGER.warn(String.format("Register single topic %s to broker %s exception", topic, brokerName), e);
+                    } finally {
+                        countDownLatch.countDown();
+                    }
+                });
+            } catch (Exception e) {
+                LOGGER.warn("Execute single topic registration task failed, topic {}, broker name {}", topic, brokerName);
+                countDownLatch.countDown();
+            }
+
+        }
+
+        try {
+            if (!countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS)) {
+                LOGGER.warn("Registration single topic to one or more name servers timeout. Timeout threshold: {}ms", timeoutMills);
+            }
+        } catch (InterruptedException ignore) {
+        }
+    }
+
     public List<Boolean> needRegister(
         final String clusterName,
         final String brokerAddr,
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index 892a71330..569a1c57b 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -441,13 +441,18 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
 
         try {
             this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig);
-            this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion());
+            if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) {
+                this.brokerController.registerSingleTopicAll(topicConfig);
+            } else {
+                this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion());
+            }
             response.setCode(ResponseCode.SUCCESS);
         } catch (Exception e) {
             LOGGER.error("Update / create topic failed for [{}]", request, e);
             response.setCode(ResponseCode.SYSTEM_ERROR);
             response.setRemark(e.getMessage());
         }
+
         return response;
     }
 
@@ -769,7 +774,8 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
         return response;
     }
 
-    private synchronized RemotingCommand updateColdDataFlowCtrGroupConfig(ChannelHandlerContext ctx, RemotingCommand request) {
+    private synchronized RemotingCommand updateColdDataFlowCtrGroupConfig(ChannelHandlerContext ctx,
+        RemotingCommand request) {
         final RemotingCommand response = RemotingCommand.createResponseCommand(null);
         LOGGER.info("updateColdDataFlowCtrGroupConfig called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
 
@@ -876,7 +882,7 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
             }
             MessageStore messageStore = this.brokerController.getMessageStore();
             if (messageStore instanceof DefaultMessageStore) {
-                DefaultMessageStore defaultMessageStore = (DefaultMessageStore)messageStore;
+                DefaultMessageStore defaultMessageStore = (DefaultMessageStore) messageStore;
                 if (mode == LibC.MADV_NORMAL) {
                     defaultMessageStore.getMessageStoreConfig().setDataReadAheadEnable(true);
                 } else {
@@ -1835,13 +1841,13 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
     /**
      * Reset consumer offset.
      *
-     * @param topic     Required, not null.
-     * @param group     Required, not null.
-     * @param queueId   if target queue ID is negative, all message queues will be reset;
-     *                  otherwise, only the target queue would get reset.
-     * @param timestamp if timestamp is negative, offset would be reset to broker offset at the time being;
-     *                  otherwise, binary search is performed to locate target offset.
-     * @param offset    Target offset to reset to if target queue ID is properly provided.
+     * @param topic Required, not null.
+     * @param group Required, not null.
+     * @param queueId if target queue ID is negative, all message queues will be reset; otherwise, only the target queue
+     * would get reset.
+     * @param timestamp if timestamp is negative, offset would be reset to broker offset at the time being; otherwise,
+     * binary search is performed to locate target offset.
+     * @param offset Target offset to reset to if target queue ID is properly provided.
      * @return Affected queues and their new offset
      */
     private RemotingCommand resetOffsetInner(String topic, String group, int queueId, long timestamp, Long offset) {
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
index e5fdd8675..e90530512 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/topic/TopicConfigManager.java
@@ -305,7 +305,11 @@ public class TopicConfigManager extends ConfigManager {
             log.error("createTopicIfAbsent ", e);
         }
         if (createNew && register) {
-            this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion);
+            if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) {
+                this.brokerController.registerSingleTopicAll(topicConfig);
+            } else {
+                this.brokerController.registerIncrementBrokerData(topicConfig, dataVersion);
+            }
         }
         return this.topicConfigTable.get(topicConfig.getTopicName());
     }
diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
index a4d82d1c5..02c692e2b 100644
--- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@ -386,6 +386,12 @@ public class BrokerConfig extends BrokerIdentity {
      */
     private boolean popResponseReturnActualRetryTopic = false;
 
+    /**
+     * If both the deleteTopicWithBrokerRegistration flag in the NameServer configuration and this flag are set to true,
+     * it guarantees the ultimate consistency of data between the broker and the nameserver during topic deletion.
+     */
+    private boolean enableSingleTopicRegister = false;
+
     public long getMaxPopPollingSize() {
         return maxPopPollingSize;
     }
@@ -1689,4 +1695,12 @@ public class BrokerConfig extends BrokerIdentity {
     public void setPopResponseReturnActualRetryTopic(boolean popResponseReturnActualRetryTopic) {
         this.popResponseReturnActualRetryTopic = popResponseReturnActualRetryTopic;
     }
+
+    public boolean isEnableSingleTopicRegister() {
+        return enableSingleTopicRegister;
+    }
+
+    public void setEnableSingleTopicRegister(boolean enableSingleTopicRegister) {
+        this.enableSingleTopicRegister = enableSingleTopicRegister;
+    }
 }
diff --git a/common/src/main/java/org/apache/rocketmq/common/namesrv/NamesrvConfig.java b/common/src/main/java/org/apache/rocketmq/common/namesrv/NamesrvConfig.java
index 700febfe2..5b8a6dedb 100644
--- a/common/src/main/java/org/apache/rocketmq/common/namesrv/NamesrvConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/namesrv/NamesrvConfig.java
@@ -82,6 +82,15 @@ public class NamesrvConfig {
 
     private int waitSecondsForService = 45;
 
+    /**
+     * If enable this flag, the topics that don't exist in broker registration payload will be deleted from name server.
+     *
+     * WARNING:
+     * 1. Enable this flag and "enableSingleTopicRegister" of broker config meanwhile to avoid losing topic route info unexpectedly.
+     * 2. This flag does not support static topic currently.
+     */
+    private boolean deleteTopicWithBrokerRegistration = false;
+
     public boolean isOrderMessageEnable() {
         return orderMessageEnable;
     }
@@ -241,4 +250,12 @@ public class NamesrvConfig {
     public void setWaitSecondsForService(int waitSecondsForService) {
         this.waitSecondsForService = waitSecondsForService;
     }
+
+    public boolean isDeleteTopicWithBrokerRegistration() {
+        return deleteTopicWithBrokerRegistration;
+    }
+
+    public void setDeleteTopicWithBrokerRegistration(boolean deleteTopicWithBrokerRegistration) {
+        this.deleteTopicWithBrokerRegistration = deleteTopicWithBrokerRegistration;
+    }
 }
diff --git a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
index ac27d76ce..0055a1cc8 100644
--- a/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
+++ b/namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java
@@ -121,9 +121,18 @@ public class RouteInfoManager {
         if (queueDatas == null || queueDatas.isEmpty()) {
             return;
         }
+
         try {
             this.lock.writeLock().lockInterruptibly();
             if (this.topicQueueTable.containsKey(topic)) {
+                Map<String, QueueData> queueDataMap  = this.topicQueueTable.get(topic);
+                for (QueueData queueData : queueDatas) {
+                    if (!this.brokerAddrTable.containsKey(queueData.getBrokerName())) {
+                        log.warn("Register topic contains illegal broker, {}, {}", topic, queueData);
+                        return;
+                    }
+                    queueDataMap.put(queueData.getBrokerName(), queueData);
+                }
                 log.info("Topic route already exist.{}, {}", topic, this.topicQueueTable.get(topic));
             } else {
                 // check and construct queue data map
@@ -299,7 +308,32 @@ public class RouteInfoManager {
 
                 ConcurrentMap<String, TopicConfig> tcTable =
                     topicConfigWrapper.getTopicConfigTable();
+
                 if (tcTable != null) {
+
+                    TopicConfigAndMappingSerializeWrapper mappingSerializeWrapper = TopicConfigAndMappingSerializeWrapper.from(topicConfigWrapper);
+                    Map<String, TopicQueueMappingInfo> topicQueueMappingInfoMap = mappingSerializeWrapper.getTopicQueueMappingInfoMap();
+
+                    // Delete the topics that don't exist in tcTable from the current broker
+                    // Static topic is not supported currently
+                    if (namesrvConfig.isDeleteTopicWithBrokerRegistration() && topicQueueMappingInfoMap.isEmpty()) {
+                        final Set<String> oldTopicSet = topicSetOfBrokerName(brokerName);
+                        final Set<String> newTopicSet = tcTable.keySet();
+                        final Sets.SetView<String> toDeleteTopics = Sets.difference(oldTopicSet, newTopicSet);
+                        for (final String toDeleteTopic : toDeleteTopics) {
+                            Map<String, QueueData> queueDataMap = topicQueueTable.get(toDeleteTopic);
+                            final QueueData removedQD = queueDataMap.remove(brokerName);
+                            if (removedQD != null) {
+                                log.info("deleteTopic, remove one broker's topic {} {} {}", brokerName, toDeleteTopic, removedQD);
+                            }
+
+                            if (queueDataMap.isEmpty()) {
+                                log.info("deleteTopic, remove the topic all queue {}", toDeleteTopic);
+                                topicQueueTable.remove(toDeleteTopic);
+                            }
+                        }
+                    }
+
                     for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
                         if (registerFirst || this.isTopicConfigChanged(clusterName, brokerAddr,
                             topicConfigWrapper.getDataVersion(), brokerName,
@@ -312,19 +346,17 @@ public class RouteInfoManager {
                             this.createAndUpdateQueueData(brokerName, topicConfig);
                         }
                     }
-                }
 
-                if (this.isBrokerTopicConfigChanged(clusterName, brokerAddr, topicConfigWrapper.getDataVersion()) || registerFirst) {
-                    TopicConfigAndMappingSerializeWrapper mappingSerializeWrapper = TopicConfigAndMappingSerializeWrapper.from(topicConfigWrapper);
-                    Map<String, TopicQueueMappingInfo> topicQueueMappingInfoMap = mappingSerializeWrapper.getTopicQueueMappingInfoMap();
-                    //the topicQueueMappingInfoMap should never be null, but can be empty
-                    for (Map.Entry<String, TopicQueueMappingInfo> entry : topicQueueMappingInfoMap.entrySet()) {
-                        if (!topicQueueMappingInfoTable.containsKey(entry.getKey())) {
-                            topicQueueMappingInfoTable.put(entry.getKey(), new HashMap<>());
+                    if (this.isBrokerTopicConfigChanged(clusterName, brokerAddr, topicConfigWrapper.getDataVersion()) || registerFirst) {
+                        //the topicQueueMappingInfoMap should never be null, but can be empty
+                        for (Map.Entry<String, TopicQueueMappingInfo> entry : topicQueueMappingInfoMap.entrySet()) {
+                            if (!topicQueueMappingInfoTable.containsKey(entry.getKey())) {
+                                topicQueueMappingInfoTable.put(entry.getKey(), new HashMap<>());
+                            }
+                            //Note asset brokerName equal entry.getValue().getBname()
+                            //here use the mappingDetail.bname
+                            topicQueueMappingInfoTable.get(entry.getKey()).put(entry.getValue().getBname(), entry.getValue());
                         }
-                        //Note asset brokerName equal entry.getValue().getBname()
-                        //here use the mappingDetail.bname
-                        topicQueueMappingInfoTable.get(entry.getKey()).put(entry.getValue().getBname(), entry.getValue());
                     }
                 }
             }
@@ -374,6 +406,16 @@ public class RouteInfoManager {
         return result;
     }
 
+    private Set<String> topicSetOfBrokerName(final String brokerName) {
+        Set<String> topicOfBroker = new HashSet<>();
+        for (final Entry<String, Map<String, QueueData>> entry : this.topicQueueTable.entrySet()) {
+            if (entry.getValue().containsKey(brokerName)) {
+                topicOfBroker.add(entry.getKey());
+            }
+        }
+        return topicOfBroker;
+    }
+
     public BrokerMemberGroup getBrokerMemberGroup(String clusterName, String brokerName) {
         BrokerMemberGroup groupMember = new BrokerMemberGroup(clusterName, brokerName);
         try {
diff --git a/namesrv/src/test/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManagerNewTest.java b/namesrv/src/test/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManagerNewTest.java
index b53519e5f..6002d1f5a 100644
--- a/namesrv/src/test/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManagerNewTest.java
+++ b/namesrv/src/test/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManagerNewTest.java
@@ -22,6 +22,7 @@ import io.netty.channel.Channel;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -37,6 +38,7 @@ import org.apache.rocketmq.remoting.protocol.body.TopicList;
 import org.apache.rocketmq.remoting.protocol.header.namesrv.UnRegisterBrokerRequestHeader;
 import org.apache.rocketmq.remoting.protocol.namesrv.RegisterBrokerResult;
 import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.QueueData;
 import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.junit.After;
 import org.junit.Before;
@@ -624,6 +626,92 @@ public class RouteInfoManagerNewTest {
             .containsValues(BrokerBasicInfo.defaultBroker().brokerAddr, BrokerBasicInfo.slaveBroker().brokerAddr);
     }
 
+    @Test
+    public void keepTopicWithBrokerRegistration() {
+        RegisterBrokerResult masterResult = registerBrokerWithNormalTopic(BrokerBasicInfo.defaultBroker(), "TestTopic", "TestTopic1");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNotNull();
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1")).isNotNull();
+
+        masterResult = registerBrokerWithNormalTopic(BrokerBasicInfo.defaultBroker(),  "TestTopic1");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNotNull();
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1")).isNotNull();
+    }
+
+    @Test
+    public void deleteTopicWithBrokerRegistration() {
+        config.setDeleteTopicWithBrokerRegistration(true);
+        registerBrokerWithNormalTopic(BrokerBasicInfo.defaultBroker(), "TestTopic", "TestTopic1");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNotNull();
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1")).isNotNull();
+
+        registerBrokerWithNormalTopic(BrokerBasicInfo.defaultBroker(),  "TestTopic1");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNull();
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1")).isNotNull();
+    }
+
+    @Test
+    public void deleteTopicWithBrokerRegistration2() {
+        // Register two brokers and delete a specific one by one
+        config.setDeleteTopicWithBrokerRegistration(true);
+        final BrokerBasicInfo master1 = BrokerBasicInfo.defaultBroker();
+        final BrokerBasicInfo master2 = BrokerBasicInfo.defaultBroker().name(DEFAULT_BROKER + 1).addr(DEFAULT_ADDR + 9);
+
+        registerBrokerWithNormalTopic(master1, "TestTopic", "TestTopic1");
+        registerBrokerWithNormalTopic(master2, "TestTopic", "TestTopic1");
+
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic").getBrokerDatas()).hasSize(2);
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1").getBrokerDatas()).hasSize(2);
+
+
+        registerBrokerWithNormalTopic(master1,  "TestTopic1");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic").getBrokerDatas()).hasSize(1);
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic").getBrokerDatas().get(0).getBrokerName())
+            .isEqualTo(master2.brokerName);
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1").getBrokerDatas()).hasSize(2);
+
+        registerBrokerWithNormalTopic(master2,  "TestTopic1");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNull();
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1").getBrokerDatas()).hasSize(2);
+    }
+
+    @Test
+    public void registerSingleTopicWithBrokerRegistration() {
+        config.setDeleteTopicWithBrokerRegistration(true);
+        final BrokerBasicInfo master1 = BrokerBasicInfo.defaultBroker();
+
+        registerSingleTopicWithBrokerName(master1.brokerName, "TestTopic");
+
+        // Single topic registration failed because there is no broker connection exists
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNull();
+
+        // Register broker with TestTopic first and then register single topic TestTopic1
+        registerBrokerWithNormalTopic(master1, "TestTopic");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNotNull();
+
+        registerSingleTopicWithBrokerName(master1.brokerName, "TestTopic1");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1")).isNotNull();
+
+        // Register the two topics to keep the route info
+        registerBrokerWithNormalTopic(master1, "TestTopic", "TestTopic1");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNotNull();
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1")).isNotNull();
+
+        // Cancel the TestTopic1 with broker registration
+        registerBrokerWithNormalTopic(master1, "TestTopic");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNotNull();
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1")).isNull();
+
+        // Add TestTopic1 and cancel all the topics with broker un-registration
+        registerSingleTopicWithBrokerName(master1.brokerName, "TestTopic1");
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1")).isNotNull();
+
+        routeInfoManager.unregisterBroker(master1.clusterName, master1.brokerAddr, master1.brokerName, 0);
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic")).isNull();
+        assertThat(routeInfoManager.pickupTopicRouteData("TestTopic1")).isNull();
+
+
+    }
+
     private RegisterBrokerResult registerBrokerWithNormalTopic(BrokerBasicInfo brokerInfo, String... topics) {
         ConcurrentHashMap<String, TopicConfig> topicConfigConcurrentHashMap = new ConcurrentHashMap<>();
         TopicConfig baseTopic = new TopicConfig("baseTopic");
@@ -711,6 +799,17 @@ public class RouteInfoManagerNewTest {
         return registerBrokerResult;
     }
 
+    private void registerSingleTopicWithBrokerName(String brokerName, String... topics) {
+        for (final String topic : topics) {
+            QueueData queueData = new QueueData();
+            queueData.setBrokerName(brokerName);
+            queueData.setReadQueueNums(8);
+            queueData.setWriteQueueNums(8);
+            queueData.setPerm(6);
+            routeInfoManager.registerTopic(topic, Collections.singletonList(queueData));
+        }
+    }
+
     static class BrokerBasicInfo {
         String clusterName;
         String brokerName;
diff --git a/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java b/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java
index 11b00a72c..d3d5de9e2 100644
--- a/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java
+++ b/test/src/main/java/org/apache/rocketmq/test/util/MQAdminTestUtils.java
@@ -17,6 +17,7 @@
 
 package org.apache.rocketmq.test.util;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -38,6 +39,7 @@ import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
 import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
 import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.remoting.protocol.statictopic.TopicConfigAndQueueMapping;
 import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingOne;
 import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingUtils;
@@ -319,4 +321,39 @@ public class MQAdminTestUtils {
         }
         return consumeStats;
     }
+
+    /**
+     * Delete topic from broker only without cleaning route info from name server forwardly
+     *
+     * @param nameSrvAddr the namesrv addr to connect
+     * @param brokerName the specific broker
+     * @param topic the specific topic to delete
+     */
+    public static void deleteTopicFromBrokerOnly(String nameSrvAddr, String brokerName, String topic) {
+        DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt();
+        mqAdminExt.setNamesrvAddr(nameSrvAddr);
+
+        try {
+            mqAdminExt.start();
+            String brokerAddr = CommandUtil.fetchMasterAddrByBrokerName(mqAdminExt, brokerName);
+            mqAdminExt.deleteTopicInBroker(Collections.singleton(brokerAddr), topic);
+        } catch (Exception ignored) {
+        } finally {
+            mqAdminExt.shutdown();
+        }
+    }
+
+    public static TopicRouteData examineTopicRouteInfo(String nameSrvAddr, String topicName) {
+        DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt();
+        mqAdminExt.setNamesrvAddr(nameSrvAddr);
+        TopicRouteData route = null;
+        try {
+            mqAdminExt.start();
+            route = mqAdminExt.examineTopicRouteInfo(topicName);
+        } catch (Exception ignored) {
+        } finally {
+            mqAdminExt.shutdown();
+        }
+        return route;
+    }
 }
diff --git a/test/src/test/java/org/apache/rocketmq/test/base/dledger/DLedgerProduceAndConsumeIT.java b/test/src/test/java/org/apache/rocketmq/test/dledger/DLedgerProduceAndConsumeIT.java
similarity index 99%
rename from test/src/test/java/org/apache/rocketmq/test/base/dledger/DLedgerProduceAndConsumeIT.java
rename to test/src/test/java/org/apache/rocketmq/test/dledger/DLedgerProduceAndConsumeIT.java
index 9e142eb61..43fefd616 100644
--- a/test/src/test/java/org/apache/rocketmq/test/base/dledger/DLedgerProduceAndConsumeIT.java
+++ b/test/src/test/java/org/apache/rocketmq/test/dledger/DLedgerProduceAndConsumeIT.java
@@ -14,7 +14,7 @@
  *  See the License for the specific language governing permissions and
  *  limitations under the License.
  */
-package org.apache.rocketmq.test.base.dledger;
+package org.apache.rocketmq.test.dledger;
 
 import java.util.UUID;
 import org.apache.rocketmq.broker.BrokerController;
diff --git a/test/src/test/java/org/apache/rocketmq/test/route/CreateAndUpdateTopicIT.java b/test/src/test/java/org/apache/rocketmq/test/route/CreateAndUpdateTopicIT.java
new file mode 100644
index 000000000..7e3c7b871
--- /dev/null
+++ b/test/src/test/java/org/apache/rocketmq/test/route/CreateAndUpdateTopicIT.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.test.route;
+
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
+import org.apache.rocketmq.test.base.BaseConf;
+import org.apache.rocketmq.test.util.MQAdminTestUtils;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class CreateAndUpdateTopicIT extends BaseConf {
+
+    @Test
+    public void testCreateOrUpdateTopic_EnableSingleTopicRegistration() {
+        String topic = "test-topic-without-broker-registration";
+        brokerController1.getBrokerConfig().setEnableSingleTopicRegister(true);
+        brokerController2.getBrokerConfig().setEnableSingleTopicRegister(true);
+        brokerController3.getBrokerConfig().setEnableSingleTopicRegister(true);
+
+        final boolean createResult = MQAdminTestUtils.createTopic(NAMESRV_ADDR, CLUSTER_NAME, topic, 8, null);
+        assertThat(createResult).isTrue();
+
+        TopicRouteData route = MQAdminTestUtils.examineTopicRouteInfo(NAMESRV_ADDR, topic);
+        assertThat(route.getBrokerDatas()).hasSize(3);
+        assertThat(route.getQueueDatas()).hasSize(3);
+
+        brokerController1.getBrokerConfig().setEnableSingleTopicRegister(false);
+        brokerController2.getBrokerConfig().setEnableSingleTopicRegister(false);
+        brokerController3.getBrokerConfig().setEnableSingleTopicRegister(false);
+
+    }
+
+    @Test
+    public void testDeleteTopicFromNameSrvWithBrokerRegistration() {
+        namesrvController.getNamesrvConfig().setDeleteTopicWithBrokerRegistration(true);
+        brokerController1.getBrokerConfig().setEnableSingleTopicRegister(true);
+        brokerController2.getBrokerConfig().setEnableSingleTopicRegister(true);
+        brokerController3.getBrokerConfig().setEnableSingleTopicRegister(true);
+
+        String testTopic1 = "test-topic-keep-route";
+        String testTopic2 = "test-topic-delete-route";
+
+        boolean createResult = MQAdminTestUtils.createTopic(NAMESRV_ADDR, CLUSTER_NAME, testTopic1, 8, null);
+        assertThat(createResult).isTrue();
+
+
+        createResult = MQAdminTestUtils.createTopic(NAMESRV_ADDR, CLUSTER_NAME, testTopic2, 8, null);
+        assertThat(createResult).isTrue();
+
+
+        TopicRouteData route = MQAdminTestUtils.examineTopicRouteInfo(NAMESRV_ADDR, testTopic2);
+        assertThat(route.getBrokerDatas()).hasSize(3);
+
+        MQAdminTestUtils.deleteTopicFromBrokerOnly(NAMESRV_ADDR, BROKER1_NAME, testTopic2);
+
+        // Deletion is lazy, trigger broker registration
+        brokerController1.registerBrokerAll(false, false, true);
+
+        // The route info of testTopic2 will be removed from broker1 after the registration
+        route = MQAdminTestUtils.examineTopicRouteInfo(NAMESRV_ADDR, testTopic2);
+        assertThat(route.getBrokerDatas()).hasSize(2);
+        assertThat(route.getQueueDatas().get(0).getBrokerName()).isEqualTo(BROKER2_NAME);
+        assertThat(route.getQueueDatas().get(1).getBrokerName()).isEqualTo(BROKER3_NAME);
+
+        brokerController1.getBrokerConfig().setEnableSingleTopicRegister(false);
+        brokerController2.getBrokerConfig().setEnableSingleTopicRegister(false);
+        brokerController3.getBrokerConfig().setEnableSingleTopicRegister(false);
+        namesrvController.getNamesrvConfig().setDeleteTopicWithBrokerRegistration(false);
+    }
+
+    @Test
+    public void testStaticTopicNotAffected() throws Exception {
+        namesrvController.getNamesrvConfig().setDeleteTopicWithBrokerRegistration(true);
+        brokerController1.getBrokerConfig().setEnableSingleTopicRegister(true);
+        brokerController2.getBrokerConfig().setEnableSingleTopicRegister(true);
+        brokerController3.getBrokerConfig().setEnableSingleTopicRegister(true);
+
+        String testTopic = "test-topic-not-affected";
+        String testStaticTopic = "test-static-topic";
+
+        boolean createResult = MQAdminTestUtils.createTopic(NAMESRV_ADDR, CLUSTER_NAME, testTopic, 8, null);
+        assertThat(createResult).isTrue();
+
+        TopicRouteData route = MQAdminTestUtils.examineTopicRouteInfo(NAMESRV_ADDR, testTopic);
+        assertThat(route.getBrokerDatas()).hasSize(3);
+        assertThat(route.getQueueDatas()).hasSize(3);
+
+        MQAdminTestUtils.createStaticTopicWithCommand(testStaticTopic, 10, null, CLUSTER_NAME, NAMESRV_ADDR);
+
+        assertThat(route.getBrokerDatas()).hasSize(3);
+        assertThat(route.getQueueDatas()).hasSize(3);
+
+        brokerController1.getBrokerConfig().setEnableSingleTopicRegister(false);
+        brokerController2.getBrokerConfig().setEnableSingleTopicRegister(false);
+        brokerController3.getBrokerConfig().setEnableSingleTopicRegister(false);
+        namesrvController.getNamesrvConfig().setDeleteTopicWithBrokerRegistration(false);
+    }
+}
-- 
2.32.0.windows.2


From 32eb1d55570af81641a4a40d96ff5554329b93cb Mon Sep 17 00:00:00 2001
From: gaoyf <gaoyf@users.noreply.github.com>
Date: Tue, 25 Jul 2023 15:26:20 +0800
Subject: [PATCH 05/10] [ISSUE #7068] Fix failed to create syncer topic when
 the proxy was just started (#7076)

---
 .../apache/rocketmq/client/impl/mqclient/MQClientAPIFactory.java | 1 +
 1 file changed, 1 insertion(+)

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIFactory.java b/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIFactory.java
index f7d9b11ba..c68859b28 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIFactory.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIFactory.java
@@ -104,6 +104,7 @@ public class MQClientAPIFactory implements StartAndShutdown {
             rpcHook);
 
         if (!mqClientAPIExt.updateNameServerAddressList()) {
+            mqClientAPIExt.fetchNameServerAddr();
             this.scheduledExecutorService.scheduleAtFixedRate(
                 mqClientAPIExt::fetchNameServerAddr,
                 Duration.ofSeconds(10).toMillis(),
-- 
2.32.0.windows.2


From d79737788078707168c0258c4af0d800de32c137 Mon Sep 17 00:00:00 2001
From: Vincent Lee <cool8511@gmail.com>
Date: Thu, 27 Jul 2023 10:51:51 +0800
Subject: [PATCH 06/10] [ISSUE #7056] Avoid close success channel if invokeSync
 most time cost on get connection for channel (#7057)

* fix: avoid close success channel if invokeSync most time cost on get channel

Change-Id: I29741cf55ac6333bfa30fef755357b78a22b1325

* fix: ci style

Change-Id: I8c9b86e9cb6f1463bf213e64c9b8c139afa794c8
---
 .../rocketmq/remoting/netty/NettyRemotingClient.java  | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java
index 9715b918a..8491f4354 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java
@@ -88,6 +88,7 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
     private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);
 
     private static final long LOCK_TIMEOUT_MILLIS = 3000;
+    private static final long MIN_CLOSE_TIMEOUT_MILLIS = 100;
 
     private final NettyClientConfig nettyClientConfig;
     private final Bootstrap bootstrap = new Bootstrap();
@@ -524,13 +525,15 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
         final Channel channel = this.getAndCreateChannel(addr);
         String channelRemoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
         if (channel != null && channel.isActive()) {
+            long left = timeoutMillis;
             try {
                 doBeforeRpcHooks(channelRemoteAddr, request);
                 long costTime = System.currentTimeMillis() - beginStartTime;
-                if (timeoutMillis < costTime) {
+                left -= costTime;
+                if (left <= 0) {
                     throw new RemotingTimeoutException("invokeSync call the addr[" + channelRemoteAddr + "] timeout");
                 }
-                RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
+                RemotingCommand response = this.invokeSyncImpl(channel, request, left);
                 doAfterRpcHooks(channelRemoteAddr, request, response);
                 this.updateChannelLastResponseTime(addr);
                 return response;
@@ -539,7 +542,9 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
                 this.closeChannel(addr, channel);
                 throw e;
             } catch (RemotingTimeoutException e) {
-                if (nettyClientConfig.isClientCloseSocketIfTimeout()) {
+                // avoid close the success channel if left timeout is small, since it may cost too much time in get the success channel, the left timeout for read is small
+                boolean shouldClose = left > MIN_CLOSE_TIMEOUT_MILLIS || left > timeoutMillis / 4;
+                if (nettyClientConfig.isClientCloseSocketIfTimeout() && shouldClose) {
                     this.closeChannel(addr, channel);
                     LOGGER.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, channelRemoteAddr);
                 }
-- 
2.32.0.windows.2


From d0a69be563785ca815dc31ef1aab4c1bc5588c01 Mon Sep 17 00:00:00 2001
From: zd46319 <zd46319@163.com>
Date: Thu, 27 Jul 2023 16:56:41 +0800
Subject: [PATCH 07/10] [ISSUE #6810] Fix the bug of mistakenly deleting data
 in clientChannelTable when the channel expire (#7073)

---
 .../broker/client/ProducerManager.java        |  5 ++-
 .../broker/client/ProducerManagerTest.java    | 34 +++++++++++++++++++
 2 files changed, 38 insertions(+), 1 deletion(-)

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java b/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java
index 52d67bf28..f9fe1193e 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java
@@ -112,7 +112,10 @@ public class ProducerManager {
                 long diff = System.currentTimeMillis() - info.getLastUpdateTimestamp();
                 if (diff > CHANNEL_EXPIRED_TIMEOUT) {
                     it.remove();
-                    clientChannelTable.remove(info.getClientId());
+                    Channel channelInClientTable = clientChannelTable.get(info.getClientId());
+                    if (channelInClientTable != null && channelInClientTable.equals(info.getChannel())) {
+                        clientChannelTable.remove(info.getClientId());
+                    }
                     log.warn(
                             "ProducerManager#scanNotActiveChannel: remove expired channel[{}] from ProducerManager groupChannelTable, producer group name: {}",
                             RemotingHelper.parseChannelRemoteAddr(info.getChannel()), group);
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/client/ProducerManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/client/ProducerManagerTest.java
index dac5468c8..3d6091e02 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/client/ProducerManagerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/client/ProducerManagerTest.java
@@ -27,6 +27,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -79,6 +80,39 @@ public class ProducerManagerTest {
         assertThat(producerManager.findChannel("clientId")).isNull();
     }
 
+    @Test
+    public void scanNotActiveChannelWithSameClientId() throws Exception {
+        producerManager.registerProducer(group, clientInfo);
+        Channel channel1 = Mockito.mock(Channel.class);
+        ClientChannelInfo clientInfo1 = new ClientChannelInfo(channel1, clientInfo.getClientId(), LanguageCode.JAVA, 0);
+        producerManager.registerProducer(group, clientInfo1);
+        AtomicReference<String> groupRef = new AtomicReference<>();
+        AtomicReference<ClientChannelInfo> clientChannelInfoRef = new AtomicReference<>();
+        producerManager.appendProducerChangeListener((event, group, clientChannelInfo) -> {
+            switch (event) {
+                case GROUP_UNREGISTER:
+                    groupRef.set(group);
+                    break;
+                case CLIENT_UNREGISTER:
+                    clientChannelInfoRef.set(clientChannelInfo);
+                    break;
+                default:
+                    break;
+            }
+        });
+        assertThat(producerManager.getGroupChannelTable().get(group).get(channel)).isNotNull();
+        assertThat(producerManager.getGroupChannelTable().get(group).get(channel1)).isNotNull();
+        assertThat(producerManager.findChannel("clientId")).isNotNull();
+        Field field = ProducerManager.class.getDeclaredField("CHANNEL_EXPIRED_TIMEOUT");
+        field.setAccessible(true);
+        long channelExpiredTimeout = field.getLong(producerManager);
+        clientInfo.setLastUpdateTimestamp(System.currentTimeMillis() - channelExpiredTimeout - 10);
+        when(channel.close()).thenReturn(mock(ChannelFuture.class));
+        producerManager.scanNotActiveChannel();
+        assertThat(producerManager.getGroupChannelTable().get(group).get(channel1)).isNotNull();
+        assertThat(producerManager.findChannel("clientId")).isNotNull();
+    }
+
     @Test
     public void doChannelCloseEvent() throws Exception {
         producerManager.registerProducer(group, clientInfo);
-- 
2.32.0.windows.2


From d429bd72dfae0901f4325c8e9c6ce631286e40d4 Mon Sep 17 00:00:00 2001
From: cnScarb <jjhfen00@163.com>
Date: Fri, 28 Jul 2023 09:46:39 +0800
Subject: [PATCH 08/10] [ISSUE #7039] Fix retry message filter when subtype is
 TAG (#7040)

---
 .../broker/filter/ExpressionForRetryMessageFilter.java      | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/filter/ExpressionForRetryMessageFilter.java b/broker/src/main/java/org/apache/rocketmq/broker/filter/ExpressionForRetryMessageFilter.java
index d2d1087ef..bc01b21cb 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/filter/ExpressionForRetryMessageFilter.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/filter/ExpressionForRetryMessageFilter.java
@@ -45,12 +45,12 @@ public class ExpressionForRetryMessageFilter extends ExpressionMessageFilter {
             return true;
         }
 
-        boolean isRetryTopic = subscriptionData.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX);
-
-        if (!isRetryTopic && ExpressionType.isTagType(subscriptionData.getExpressionType())) {
+        if (ExpressionType.isTagType(subscriptionData.getExpressionType())) {
             return true;
         }
 
+        boolean isRetryTopic = subscriptionData.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX);
+
         ConsumerFilterData realFilterData = this.consumerFilterData;
         Map<String, String> tempProperties = properties;
         boolean decoded = false;
-- 
2.32.0.windows.2


From 8baa51e85e569429293720b2ba7fcaee745abecc Mon Sep 17 00:00:00 2001
From: Zack_Aayush <60972989+AayushSaini101@users.noreply.github.com>
Date: Sun, 30 Jul 2023 09:02:02 +0530
Subject: [PATCH 09/10] [ISSUE #7091] Update the cd command in README (#7096)

* Update the cd command

* Removed extra space

---------

Co-authored-by: Aayush <aaayush@redhat.com>
---
 README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 393ef88e6..56d253ce1 100644
--- a/README.md
+++ b/README.md
@@ -63,7 +63,7 @@ $ unzip rocketmq-all-5.1.3-bin-release.zip
 
 Prepare a terminal and change to the extracted `bin` directory:
 ```shell
-$ cd rocketmq-all-5.1.3/bin
+$ cd rocketmq-all-5.1.3-bin-release/bin
 ```
 
 **1) Start NameServer**
-- 
2.32.0.windows.2


From 8bcc94829d2ef2597a8eeab3c6b7099432a0bea1 Mon Sep 17 00:00:00 2001
From: weihubeats <weihubeats@163.com>
Date: Tue, 1 Aug 2023 10:15:07 +0800
Subject: [PATCH 10/10] [ISSUE #7077] Schedule CQ offset invalid. offset=77,
 cqMinOffset=0, cqMaxOffset=74, queueId=1 (#7084)

* Adding null does not update

* delete slave put correctDelayOffset

* Remove duplicate delayOffset file loading

* add loadWhenSyncDelayOffset

* add method

* add method
---
 .../rocketmq/broker/schedule/ScheduleMessageService.java    | 6 ++++++
 .../org/apache/rocketmq/broker/slave/SlaveSynchronize.java  | 2 +-
 2 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java b/broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java
index 2a4ace098..26f09dcd0 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java
@@ -223,6 +223,12 @@ public class ScheduleMessageService extends ConfigManager {
         result = result && this.correctDelayOffset();
         return result;
     }
+    
+    public boolean loadWhenSyncDelayOffset() {
+        boolean result = super.load();
+        result = result && this.parseDelayLevel();
+        return result;
+    }
 
     public boolean correctDelayOffset() {
         try {
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java b/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java
index b9de5173b..53cdecdf8 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java
@@ -152,7 +152,7 @@ public class SlaveSynchronize {
                                     .getMessageStoreConfig().getStorePathRootDir());
                     try {
                         MixAll.string2File(delayOffset, fileName);
-                        this.brokerController.getScheduleMessageService().load();
+                        this.brokerController.getScheduleMessageService().loadWhenSyncDelayOffset();
                     } catch (IOException e) {
                         LOGGER.error("Persist file Exception, {}", fileName, e);
                     }
-- 
2.32.0.windows.2