packages/rocketmq
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
rocketmq/patch007-backport-fix-some-bugs.patch

1426 lines
70 KiB
Diff
Raw Permalink Normal View History

backport fix some doc error
2023-09-19 16:10:33 +08:00
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
Reference in New Issue Copy Permalink