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

From 15c6889bb0abd014c06ef1452f791db9daa1ea08 Mon Sep 17 00:00:00 2001
From: lizhimins <707364882@qq.com>
Date: Tue, 11 Jul 2023 17:04:00 +0800
Subject: [PATCH 1/8] fix receive message activity attempt id not correct
 (#7012)

fix receive message activity attempt id not correct
---
 .../proxy/grpc/v2/consumer/ReceiveMessageActivity.java      | 2 +-
 .../proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java  | 6 +++---
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java
index a504179a9..cf58bb87a 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java
@@ -130,7 +130,7 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
                     subscriptionData,
                     fifo,
                     new PopMessageResultFilterImpl(maxAttempts),
-                    request.getAttemptId(),
+                    request.hasAttemptId() ? request.getAttemptId() : null,
                     timeRemaining
                 ).thenAccept(popResult -> {
                     if (proxyConfig.isEnableProxyAutoRenew() && request.getAutoRenew()) {
diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java
index 2e562504a..7fd9a9ffd 100644
--- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java
+++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivityTest.java
@@ -57,6 +57,7 @@ import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -89,7 +90,7 @@ public class ReceiveMessageActivityTest extends BaseActivityTest {
             .setRequestTimeout(Durations.fromSeconds(3))
             .build());
         when(this.messagingProcessor.popMessage(any(), any(), anyString(), anyString(), anyInt(), anyLong(),
-            pollTimeCaptor.capture(), anyInt(), any(), anyBoolean(), any(), anyString(), anyLong()))
+            pollTimeCaptor.capture(), anyInt(), any(), anyBoolean(), any(), isNull(), anyLong()))
             .thenReturn(CompletableFuture.completedFuture(new PopResult(PopStatus.NO_NEW_MSG, Collections.emptyList())));
 
 
@@ -223,7 +224,6 @@ public class ReceiveMessageActivityTest extends BaseActivityTest {
         assertEquals(Code.ILLEGAL_INVISIBLE_TIME, getResponseCodeFromReceiveMessageResponseList(responseArgumentCaptor.getAllValues()));
     }
 
-
     @Test
     public void testReceiveMessage() {
         StreamObserver<ReceiveMessageResponse> receiveStreamObserver = mock(ServerCallStreamObserver.class);
@@ -245,7 +245,7 @@ public class ReceiveMessageActivityTest extends BaseActivityTest {
             any(),
             anyBoolean(),
             any(),
-            anyString(),
+            isNull(),
             anyLong())).thenReturn(CompletableFuture.completedFuture(popResult));
 
         this.receiveMessageActivity.receiveMessage(
-- 
2.32.0.windows.2


From b4496be68705c1c0b282a07a1adeab4fffd670fe Mon Sep 17 00:00:00 2001
From: ShuangxiDing <dingshuangxi888@gmail.com>
Date: Tue, 11 Jul 2023 19:09:09 +0800
Subject: [PATCH 2/8] [ISSUE #7010] Fix the HandshakeHandler returns when
 detect haproxy version need more data (#7011)
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

* Move AttributeKey from RemotingHelper to AttributeKey.

* Fix the needs more data for HandshakeHandler.

* Fix the needs more data for HandshakeHandler.

* Fix the needs more data for HandshakeHandler.

* Fix the needs more data for HandshakeHandler.

---------

Co-authored-by: 徒钟 <shuangxi.dsx@alibaba-inc.com>
---
 .../remoting/MultiProtocolRemotingServer.java |  2 +-
 .../activity/AbstractRemotingActivity.java    | 16 +++++++------
 .../activity/ClientManagerActivity.java       | 24 ++++++++++---------
 .../AbstractRemotingActivityTest.java         | 10 ++++----
 .../remoting/common/RemotingHelper.java       | 21 ++++------------
 .../remoting/netty/AttributeKeys.java         | 11 ++++++++-
 .../remoting/netty/NettyRemotingServer.java   | 23 ++++++------------
 7 files changed, 51 insertions(+), 56 deletions(-)

diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/MultiProtocolRemotingServer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/MultiProtocolRemotingServer.java
index 858b1f022..12d728fff 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/MultiProtocolRemotingServer.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/MultiProtocolRemotingServer.java
@@ -78,7 +78,7 @@ public class MultiProtocolRemotingServer extends NettyRemotingServer {
     @Override
     protected ChannelPipeline configChannel(SocketChannel ch) {
         return ch.pipeline()
-            .addLast(this.getDefaultEventExecutorGroup(), HANDSHAKE_HANDLER_NAME, this.getHandshakeHandler())
+            .addLast(this.getDefaultEventExecutorGroup(), HANDSHAKE_HANDLER_NAME, new HandshakeHandler())
             .addLast(this.getDefaultEventExecutorGroup(),
                 new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                 new ProtocolNegotiationHandler(this.remotingProtocolHandler)
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java
index 78cd203ec..ce4a63397 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java
@@ -19,9 +19,6 @@ package org.apache.rocketmq.proxy.remoting.activity;
 
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
 import org.apache.rocketmq.acl.common.AclException;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
@@ -41,11 +38,16 @@ import org.apache.rocketmq.proxy.processor.MessagingProcessor;
 import org.apache.rocketmq.proxy.processor.channel.ChannelProtocolType;
 import org.apache.rocketmq.proxy.remoting.pipeline.RequestPipeline;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
+import org.apache.rocketmq.remoting.netty.AttributeKeys;
 import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.protocol.RequestCode;
 import org.apache.rocketmq.remoting.protocol.ResponseCode;
 
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
 public abstract class AbstractRemotingActivity implements NettyRequestProcessor {
     protected final static Logger log = LoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
     protected final MessagingProcessor messagingProcessor;
@@ -126,13 +128,13 @@ public abstract class AbstractRemotingActivity implements NettyRequestProcessor
             .setProtocolType(ChannelProtocolType.REMOTING.getName())
             .setChannel(channel)
             .setLocalAddress(NetworkUtil.socketAddress2String(ctx.channel().localAddress()))
-            .setRemoteAddress(NetworkUtil.socketAddress2String(ctx.channel().remoteAddress()));
+            .setRemoteAddress(RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
 
-        Optional.ofNullable(RemotingHelper.getAttributeValue(RemotingHelper.LANGUAGE_CODE_KEY, channel))
+        Optional.ofNullable(RemotingHelper.getAttributeValue(AttributeKeys.LANGUAGE_CODE_KEY, channel))
             .ifPresent(language -> context.setLanguage(language.name()));
-        Optional.ofNullable(RemotingHelper.getAttributeValue(RemotingHelper.CLIENT_ID_KEY, channel))
+        Optional.ofNullable(RemotingHelper.getAttributeValue(AttributeKeys.CLIENT_ID_KEY, channel))
             .ifPresent(context::setClientID);
-        Optional.ofNullable(RemotingHelper.getAttributeValue(RemotingHelper.VERSION_KEY, channel))
+        Optional.ofNullable(RemotingHelper.getAttributeValue(AttributeKeys.VERSION_KEY, channel))
             .ifPresent(version -> context.setClientVersion(MQVersion.getVersionDesc(version)));
 
         return context;
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java
index 1eb81ce92..c671593a3 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ClientManagerActivity.java
@@ -19,13 +19,20 @@ package org.apache.rocketmq.proxy.remoting.activity;
 
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
-import java.util.Set;
 import org.apache.rocketmq.broker.client.ClientChannelInfo;
 import org.apache.rocketmq.broker.client.ConsumerGroupEvent;
 import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
 import org.apache.rocketmq.broker.client.ProducerChangeListener;
 import org.apache.rocketmq.broker.client.ProducerGroupEvent;
+import org.apache.rocketmq.proxy.common.ProxyContext;
+import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.proxy.remoting.channel.RemotingChannel;
+import org.apache.rocketmq.proxy.remoting.channel.RemotingChannelManager;
+import org.apache.rocketmq.proxy.remoting.pipeline.RequestPipeline;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.netty.AttributeKeys;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.protocol.RequestCode;
 import org.apache.rocketmq.remoting.protocol.ResponseCode;
 import org.apache.rocketmq.remoting.protocol.header.UnregisterClientRequestHeader;
@@ -33,13 +40,8 @@ import org.apache.rocketmq.remoting.protocol.header.UnregisterClientResponseHead
 import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData;
 import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
 import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
-import org.apache.rocketmq.proxy.common.ProxyContext;
-import org.apache.rocketmq.proxy.processor.MessagingProcessor;
-import org.apache.rocketmq.proxy.remoting.channel.RemotingChannel;
-import org.apache.rocketmq.proxy.remoting.channel.RemotingChannelManager;
-import org.apache.rocketmq.proxy.remoting.pipeline.RequestPipeline;
-import org.apache.rocketmq.remoting.exception.RemotingCommandException;
-import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+import java.util.Set;
 
 public class ClientManagerActivity extends AbstractRemotingActivity {
 
@@ -108,9 +110,9 @@ public class ClientManagerActivity extends AbstractRemotingActivity {
         if (channel instanceof RemotingChannel) {
             RemotingChannel remotingChannel = (RemotingChannel) channel;
             Channel parent = remotingChannel.parent();
-            RemotingHelper.setPropertyToAttr(parent, RemotingHelper.CLIENT_ID_KEY, clientChannelInfo.getClientId());
-            RemotingHelper.setPropertyToAttr(parent, RemotingHelper.LANGUAGE_CODE_KEY, clientChannelInfo.getLanguage());
-            RemotingHelper.setPropertyToAttr(parent, RemotingHelper.VERSION_KEY, clientChannelInfo.getVersion());
+            RemotingHelper.setPropertyToAttr(parent, AttributeKeys.CLIENT_ID_KEY, clientChannelInfo.getClientId());
+            RemotingHelper.setPropertyToAttr(parent, AttributeKeys.LANGUAGE_CODE_KEY, clientChannelInfo.getLanguage());
+            RemotingHelper.setPropertyToAttr(parent, AttributeKeys.VERSION_KEY, clientChannelInfo.getVersion());
         }
 
     }
diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivityTest.java
index 663a83e3c..b2bd3a35f 100644
--- a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivityTest.java
+++ b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivityTest.java
@@ -21,7 +21,6 @@ import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
-import java.util.concurrent.CompletableFuture;
 import org.apache.rocketmq.acl.common.AclException;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
@@ -35,6 +34,7 @@ import org.apache.rocketmq.proxy.processor.channel.ChannelProtocolType;
 import org.apache.rocketmq.proxy.service.channel.SimpleChannel;
 import org.apache.rocketmq.proxy.service.channel.SimpleChannelHandlerContext;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
+import org.apache.rocketmq.remoting.netty.AttributeKeys;
 import org.apache.rocketmq.remoting.protocol.LanguageCode;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.protocol.RequestCode;
@@ -48,6 +48,8 @@ import org.mockito.Mock;
 import org.mockito.Spy;
 import org.mockito.junit.MockitoJUnitRunner;
 
+import java.util.concurrent.CompletableFuture;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
@@ -82,9 +84,9 @@ public class AbstractRemotingActivityTest extends InitConfigTest {
             }
         };
         Channel channel = ctx.channel();
-        RemotingHelper.setPropertyToAttr(channel, RemotingHelper.CLIENT_ID_KEY, CLIENT_ID);
-        RemotingHelper.setPropertyToAttr(channel, RemotingHelper.LANGUAGE_CODE_KEY, LanguageCode.JAVA);
-        RemotingHelper.setPropertyToAttr(channel, RemotingHelper.VERSION_KEY, MQVersion.CURRENT_VERSION);
+        RemotingHelper.setPropertyToAttr(channel, AttributeKeys.CLIENT_ID_KEY, CLIENT_ID);
+        RemotingHelper.setPropertyToAttr(channel, AttributeKeys.LANGUAGE_CODE_KEY, LanguageCode.JAVA);
+        RemotingHelper.setPropertyToAttr(channel, AttributeKeys.VERSION_KEY, MQVersion.CURRENT_VERSION);
     }
 
     @Test
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingHelper.java b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingHelper.java
index d0750b678..363b22eac 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingHelper.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingHelper.java
@@ -22,7 +22,6 @@ import io.netty.channel.ChannelFutureListener;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.rocketmq.common.constant.HAProxyConstants;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.utils.NetworkUtil;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -31,8 +30,8 @@ import org.apache.rocketmq.remoting.exception.RemotingCommandException;
 import org.apache.rocketmq.remoting.exception.RemotingConnectException;
 import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
 import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.netty.AttributeKeys;
 import org.apache.rocketmq.remoting.netty.NettySystemConfig;
-import org.apache.rocketmq.remoting.protocol.LanguageCode;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.protocol.RequestCode;
 import org.apache.rocketmq.remoting.protocol.ResponseCode;
@@ -51,16 +50,6 @@ public class RemotingHelper {
     public static final String DEFAULT_CIDR_ALL = "0.0.0.0/0";
 
     private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);
-    private static final AttributeKey<String> REMOTE_ADDR_KEY = AttributeKey.valueOf("RemoteAddr");
-
-    private static final AttributeKey<String> PROXY_PROTOCOL_ADDR = AttributeKey.valueOf(HAProxyConstants.PROXY_PROTOCOL_ADDR);
-    private static final AttributeKey<String> PROXY_PROTOCOL_PORT = AttributeKey.valueOf(HAProxyConstants.PROXY_PROTOCOL_PORT);
-
-    public static final AttributeKey<String> CLIENT_ID_KEY = AttributeKey.valueOf("ClientId");
-
-    public static final AttributeKey<Integer> VERSION_KEY = AttributeKey.valueOf("Version");
-
-    public static final AttributeKey<LanguageCode> LANGUAGE_CODE_KEY = AttributeKey.valueOf("LanguageCode");
 
     public static final Map<Integer, String> REQUEST_CODE_MAP = new HashMap<Integer, String>() {
         {
@@ -213,7 +202,7 @@ public class RemotingHelper {
         if (StringUtils.isNotBlank(addr)) {
             return addr;
         }
-        Attribute<String> att = channel.attr(REMOTE_ADDR_KEY);
+        Attribute<String> att = channel.attr(AttributeKeys.REMOTE_ADDR_KEY);
         if (att == null) {
             // mocked in unit test
             return parseChannelRemoteAddr0(channel);
@@ -227,11 +216,11 @@ public class RemotingHelper {
     }
 
     private static String getProxyProtocolAddress(Channel channel) {
-        if (!channel.hasAttr(PROXY_PROTOCOL_ADDR)) {
+        if (!channel.hasAttr(AttributeKeys.PROXY_PROTOCOL_ADDR)) {
             return null;
         }
-        String proxyProtocolAddr = getAttributeValue(PROXY_PROTOCOL_ADDR, channel);
-        String proxyProtocolPort = getAttributeValue(PROXY_PROTOCOL_PORT, channel);
+        String proxyProtocolAddr = getAttributeValue(AttributeKeys.PROXY_PROTOCOL_ADDR, channel);
+        String proxyProtocolPort = getAttributeValue(AttributeKeys.PROXY_PROTOCOL_PORT, channel);
         if (StringUtils.isBlank(proxyProtocolAddr) || proxyProtocolPort == null) {
             return null;
         }
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/AttributeKeys.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/AttributeKeys.java
index 4e69ab82d..ebdde31f4 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/AttributeKeys.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/AttributeKeys.java
@@ -19,12 +19,21 @@ package org.apache.rocketmq.remoting.netty;
 
 import io.netty.util.AttributeKey;
 import org.apache.rocketmq.common.constant.HAProxyConstants;
+import org.apache.rocketmq.remoting.protocol.LanguageCode;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class AttributeKeys {
 
+    public static final AttributeKey<String> REMOTE_ADDR_KEY = AttributeKey.valueOf("RemoteAddr");
+
+    public static final AttributeKey<String> CLIENT_ID_KEY = AttributeKey.valueOf("ClientId");
+
+    public static final AttributeKey<Integer> VERSION_KEY = AttributeKey.valueOf("Version");
+
+    public static final AttributeKey<LanguageCode> LANGUAGE_CODE_KEY = AttributeKey.valueOf("LanguageCode");
+
     public static final AttributeKey<String> PROXY_PROTOCOL_ADDR =
             AttributeKey.valueOf(HAProxyConstants.PROXY_PROTOCOL_ADDR);
 
@@ -40,6 +49,6 @@ public class AttributeKeys {
     private static final Map<String, AttributeKey<String>> ATTRIBUTE_KEY_MAP = new ConcurrentHashMap<>();
 
     public static AttributeKey<String> valueOf(String name) {
-        return ATTRIBUTE_KEY_MAP.computeIfAbsent(name, AttributeKeys::valueOf);
+        return ATTRIBUTE_KEY_MAP.computeIfAbsent(name, AttributeKey::valueOf);
     }
 }
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
index 445f06cc6..8ae87a6fa 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
@@ -37,6 +37,7 @@ import io.netty.channel.epoll.EpollServerSocketChannel;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.codec.ProtocolDetectionResult;
 import io.netty.handler.codec.ProtocolDetectionState;
 import io.netty.handler.codec.haproxy.HAProxyMessage;
@@ -73,6 +74,7 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.security.cert.CertificateException;
+import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -115,7 +117,7 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
     public static final String FILE_REGION_ENCODER_NAME = "fileRegionEncoder";
 
     // sharable handlers
-    private HandshakeHandler handshakeHandler;
+    private TlsModeHandler tlsModeHandler;
     private NettyEncoder encoder;
     private NettyConnectManageHandler connectionManageHandler;
     private NettyServerHandler serverHandler;
@@ -265,7 +267,7 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
      */
     protected ChannelPipeline configChannel(SocketChannel ch) {
         return ch.pipeline()
-            .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler)
+            .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, new HandshakeHandler())
             .addLast(defaultEventExecutorGroup,
                 encoder,
                 new NettyDecoder(),
@@ -402,7 +404,7 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
     }
 
     private void prepareSharableHandlers() {
-        handshakeHandler = new HandshakeHandler();
+        tlsModeHandler = new TlsModeHandler(TlsSystemConfig.tlsMode);
         encoder = new NettyEncoder();
         connectionManageHandler = new NettyConnectManageHandler();
         serverHandler = new NettyServerHandler();
@@ -429,10 +431,6 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
         return defaultEventExecutorGroup;
     }
 
-    public HandshakeHandler getHandshakeHandler() {
-        return handshakeHandler;
-    }
-
     public NettyEncoder getEncoder() {
         return encoder;
     }
@@ -449,17 +447,13 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
         return distributionHandler;
     }
 
-    @ChannelHandler.Sharable
-    public class HandshakeHandler extends SimpleChannelInboundHandler<ByteBuf> {
-
-        private final TlsModeHandler tlsModeHandler;
+    public class HandshakeHandler extends ByteToMessageDecoder {
 
         public HandshakeHandler() {
-            tlsModeHandler = new TlsModeHandler(TlsSystemConfig.tlsMode);
         }
 
         @Override
-        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf byteBuf) {
+        protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> out) throws Exception {
             try {
                 ProtocolDetectionResult<HAProxyProtocolVersion> detectionResult = HAProxyMessageDecoder.detectProtocol(byteBuf);
                 if (detectionResult.state() == ProtocolDetectionState.NEEDS_MORE_DATA) {
@@ -479,9 +473,6 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
                 } catch (NoSuchElementException e) {
                     log.error("Error while removing HandshakeHandler", e);
                 }
-
-                // Hand over this message to the next .
-                ctx.fireChannelRead(byteBuf.retain());
             } catch (Exception e) {
                 log.error("process proxy protocol negotiator failed.", e);
                 throw e;
-- 
2.32.0.windows.2


From 1f0f3b2d6d16de7b315c702d33f7d3557c0fc25c Mon Sep 17 00:00:00 2001
From: Ji Juntao <juntao.jjt@alibaba-inc.com>
Date: Tue, 11 Jul 2023 21:13:06 +0800
Subject: [PATCH 3/8] [ISSUE #7013] Polish ColdDataCheckService's logic (#7014)

* polish coldCtrl

* remove the catch.
---
 .../src/main/java/org/apache/rocketmq/store/CommitLog.java | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
index 5a5c90c5a..e6ee3bacc 100644
--- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -2089,6 +2089,11 @@ public class CommitLog implements Swappable {
                     } else {
                         this.waitForRunning(defaultMessageStore.getMessageStoreConfig().getTimerColdDataCheckIntervalMs());
                     }
+
+                    if (pageSize < 0) {
+                        initPageSize();
+                    }
+
                     long beginClockTimestamp = this.systemClock.now();
                     scanFilesInPageCache();
                     long costTime = this.systemClock.now() - beginClockTimestamp;
@@ -2182,7 +2187,7 @@ public class CommitLog implements Swappable {
         }
 
         private void initPageSize() {
-            if (pageSize < 0) {
+            if (pageSize < 0 && defaultMessageStore.getMessageStoreConfig().isColdDataFlowControlEnable()) {
                 try {
                     if (!MixAll.isWindows()) {
                         pageSize = LibC.INSTANCE.getpagesize();
-- 
2.32.0.windows.2


From d206590692bfdffca6bc58327e9533bc4bb68122 Mon Sep 17 00:00:00 2001
From: Lei Zhiyuan <leizhiyuan@gmail.com>
Date: Thu, 13 Jul 2023 11:17:38 +0800
Subject: [PATCH 4/8] [ISSUE #6979] Fix opaque will be duplicate in multi
 client scene (#6985)

---
 .../proxy/processor/DefaultMessagingProcessor.java | 14 ++++++++++++--
 1 file changed, 12 insertions(+), 2 deletions(-)

diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java
index 1b3f0af4e..188cb7b9b 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java
@@ -235,13 +235,23 @@ public class DefaultMessagingProcessor extends AbstractStartAndShutdown implemen
     @Override
     public CompletableFuture<RemotingCommand> request(ProxyContext ctx, String brokerName, RemotingCommand request,
         long timeoutMillis) {
-        return this.requestBrokerProcessor.request(ctx, brokerName, request, timeoutMillis);
+        int originalRequestOpaque = request.getOpaque();
+        request.setOpaque(RemotingCommand.createNewRequestId());
+        return this.requestBrokerProcessor.request(ctx, brokerName, request, timeoutMillis).thenApply(r -> {
+            request.setOpaque(originalRequestOpaque);
+            return r;
+        });
     }
 
     @Override
     public CompletableFuture<Void> requestOneway(ProxyContext ctx, String brokerName, RemotingCommand request,
         long timeoutMillis) {
-        return this.requestBrokerProcessor.requestOneway(ctx, brokerName, request, timeoutMillis);
+        int originalRequestOpaque = request.getOpaque();
+        request.setOpaque(RemotingCommand.createNewRequestId());
+        return this.requestBrokerProcessor.requestOneway(ctx, brokerName, request, timeoutMillis).thenApply(r -> {
+            request.setOpaque(originalRequestOpaque);
+            return r;
+        });
     }
 
     @Override
-- 
2.32.0.windows.2


From 33cb22e1c0fa7ba980567117230fe443ff5dbd62 Mon Sep 17 00:00:00 2001
From: lizhimins <707364882@qq.com>
Date: Thu, 13 Jul 2023 15:37:24 +0800
Subject: [PATCH 5/8] [ISSUE #7018] fix append in tiered storage when message
 offset incorrect (#7019)

* fix append in tiered storage when message offset incorrect
---
 .../tieredstore/TieredDispatcher.java         | 25 ++++++++++------
 .../tieredstore/file/CompositeFlatFile.java   | 30 +++++++++----------
 .../file/CompositeQueueFlatFile.java          |  2 +-
 .../file/CompositeQueueFlatFileTest.java      |  4 +--
 4 files changed, 34 insertions(+), 27 deletions(-)

diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredDispatcher.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredDispatcher.java
index 2a8e2ed71..6584b0e89 100644
--- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredDispatcher.java
+++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredDispatcher.java
@@ -308,9 +308,18 @@ public class TieredDispatcher extends ServiceThread implements CommitLogDispatch
                 doRedispatchRequestToWriteMap(
                     result, flatFile, dispatchOffset, newCommitLogOffset, size, tagCode, message.getByteBuffer());
                 message.release();
-                if (result != AppendResult.SUCCESS) {
-                    dispatchOffset--;
-                    break;
+
+                switch (result) {
+                    case SUCCESS:
+                        continue;
+                    case FILE_CLOSED:
+                        tieredFlatFileManager.destroyCompositeFile(flatFile.getMessageQueue());
+                        logger.info("TieredDispatcher#dispatchFlatFile: file has been close and destroy, " +
+                            "topic: {}, queueId: {}", topic, queueId);
+                        return;
+                    default:
+                        dispatchOffset--;
+                        break;
                 }
             }
 
@@ -341,15 +350,13 @@ public class TieredDispatcher extends ServiceThread implements CommitLogDispatch
 
         switch (result) {
             case SUCCESS:
-                break;
-            case OFFSET_INCORRECT:
                 long offset = MessageBufferUtil.getQueueOffset(message);
                 if (queueOffset != offset) {
-                    logger.error("[Bug] Commitlog offset incorrect, " +
-                            "result={}, topic={}, queueId={}, offset={}, msg offset={}",
-                        result, topic, queueId, queueOffset, offset);
+                    logger.error("Message cq offset in commitlog does not meet expectations, " +
+                            "result={}, topic={}, queueId={}, cq offset={}, msg offset={}",
+                        AppendResult.OFFSET_INCORRECT, topic, queueId, queueOffset, offset);
                 }
-                return;
+                break;
             case BUFFER_FULL:
                 logger.debug("Commitlog buffer full, result={}, topic={}, queueId={}, offset={}",
                     result, topic, queueId, queueOffset);
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/CompositeFlatFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/CompositeFlatFile.java
index 1243f7721..8f8ba98b1 100644
--- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/CompositeFlatFile.java
+++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/CompositeFlatFile.java
@@ -29,6 +29,7 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
@@ -58,7 +59,7 @@ public class CompositeFlatFile implements CompositeAccess {
      * dispatched to the current chunk, indicating the progress of the message distribution.
      * It's consume queue current offset.
      */
-    protected volatile long dispatchOffset;
+    protected final AtomicLong dispatchOffset;
 
     protected final ReentrantLock compositeFlatFileLock;
     protected final TieredMessageStoreConfig storeConfig;
@@ -75,6 +76,7 @@ public class CompositeFlatFile implements CompositeAccess {
         this.storeConfig = fileQueueFactory.getStoreConfig();
         this.readAheadFactor = this.storeConfig.getReadAheadMinFactor();
         this.metadataStore = TieredStoreUtil.getMetadataStore(this.storeConfig);
+        this.dispatchOffset = new AtomicLong();
         this.compositeFlatFileLock = new ReentrantLock();
         this.inFlightRequestMap = new ConcurrentHashMap<>();
         this.commitLog = new TieredCommitLog(fileQueueFactory, filePath);
@@ -83,8 +85,8 @@ public class CompositeFlatFile implements CompositeAccess {
     }
 
     protected void recoverMetadata() {
-        if (!consumeQueue.isInitialized() && this.dispatchOffset != -1) {
-            consumeQueue.setBaseOffset(this.dispatchOffset * TieredConsumeQueue.CONSUME_QUEUE_STORE_UNIT_SIZE);
+        if (!consumeQueue.isInitialized() && this.dispatchOffset.get() != -1) {
+            consumeQueue.setBaseOffset(this.dispatchOffset.get() * TieredConsumeQueue.CONSUME_QUEUE_STORE_UNIT_SIZE);
         }
     }
 
@@ -144,7 +146,7 @@ public class CompositeFlatFile implements CompositeAccess {
     }
 
     public long getDispatchOffset() {
-        return dispatchOffset;
+        return dispatchOffset.get();
     }
 
     @Override
@@ -309,7 +311,7 @@ public class CompositeFlatFile implements CompositeAccess {
         if (!consumeQueue.isInitialized()) {
             consumeQueue.setBaseOffset(offset * TieredConsumeQueue.CONSUME_QUEUE_STORE_UNIT_SIZE);
         }
-        dispatchOffset = offset;
+        dispatchOffset.set(offset);
     }
 
     @Override
@@ -323,14 +325,9 @@ public class CompositeFlatFile implements CompositeAccess {
             return AppendResult.FILE_CLOSED;
         }
 
-        long queueOffset = MessageBufferUtil.getQueueOffset(message);
-        if (dispatchOffset != queueOffset) {
-            return AppendResult.OFFSET_INCORRECT;
-        }
-
         AppendResult result = commitLog.append(message, commit);
         if (result == AppendResult.SUCCESS) {
-            dispatchOffset = queueOffset + 1;
+            dispatchOffset.incrementAndGet();
         }
         return result;
     }
@@ -483,14 +480,17 @@ public class CompositeFlatFile implements CompositeAccess {
     }
 
     public void destroy() {
-        closed = true;
-        commitLog.destroy();
-        consumeQueue.destroy();
         try {
+            closed = true;
+            compositeFlatFileLock.lock();
+            commitLog.destroy();
+            consumeQueue.destroy();
             metadataStore.deleteFileSegment(filePath, FileSegmentType.COMMIT_LOG);
             metadataStore.deleteFileSegment(filePath, FileSegmentType.CONSUME_QUEUE);
         } catch (Exception e) {
-            LOGGER.error("CompositeFlatFile#destroy: clean metadata failed: ", e);
+            LOGGER.error("CompositeFlatFile#destroy: delete file failed", e);
+        } finally {
+            compositeFlatFileLock.unlock();
         }
     }
 }
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/CompositeQueueFlatFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/CompositeQueueFlatFile.java
index c0cf79069..f6c0afed0 100644
--- a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/CompositeQueueFlatFile.java
+++ b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/CompositeQueueFlatFile.java
@@ -64,7 +64,7 @@ public class CompositeQueueFlatFile extends CompositeFlatFile {
         if (queueMetadata.getMaxOffset() < queueMetadata.getMinOffset()) {
             queueMetadata.setMaxOffset(queueMetadata.getMinOffset());
         }
-        this.dispatchOffset = queueMetadata.getMaxOffset();
+        this.dispatchOffset.set(queueMetadata.getMaxOffset());
     }
 
     public void persistMetadata() {
diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/CompositeQueueFlatFileTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/CompositeQueueFlatFileTest.java
index 9735a535e..8322c72ed 100644
--- a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/CompositeQueueFlatFileTest.java
+++ b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/CompositeQueueFlatFileTest.java
@@ -73,8 +73,8 @@ public class CompositeQueueFlatFileTest {
         CompositeQueueFlatFile flatFile = new CompositeQueueFlatFile(tieredFileAllocator, mq);
         ByteBuffer message = MessageBufferUtilTest.buildMockedMessageBuffer();
         AppendResult result = flatFile.appendCommitLog(message);
-        Assert.assertEquals(AppendResult.OFFSET_INCORRECT, result);
-        Assert.assertEquals(0L, flatFile.commitLog.getFlatFile().getFileToWrite().getAppendPosition());
+        Assert.assertEquals(AppendResult.SUCCESS, result);
+        Assert.assertEquals(122L, flatFile.commitLog.getFlatFile().getFileToWrite().getAppendPosition());
         Assert.assertEquals(0L, flatFile.commitLog.getFlatFile().getFileToWrite().getCommitPosition());
 
         flatFile = new CompositeQueueFlatFile(tieredFileAllocator, mq);
-- 
2.32.0.windows.2


From 70a66eda2c08eb5fca38356659cb6de1ac75e25e Mon Sep 17 00:00:00 2001
From: ShuangxiDing <dingshuangxi888@gmail.com>
Date: Fri, 14 Jul 2023 16:46:40 +0800
Subject: [PATCH 6/8] Fix LEAK: HAProxyMessage.release() was not called before
 it's garbage-collected (#7025)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Call HAProxyMessage.release() after reading it.

---------

Co-authored-by: 徒钟 <shuangxi.dsx@alibaba-inc.com>
---
 .../grpc/ProxyAndTlsProtocolNegotiator.java   | 52 ++++++++++---------
 .../remoting/netty/NettyRemotingServer.java   | 46 ++++++++--------
 2 files changed, 53 insertions(+), 45 deletions(-)

diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/ProxyAndTlsProtocolNegotiator.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/ProxyAndTlsProtocolNegotiator.java
index ceb9becc0..ee167bd7b 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/ProxyAndTlsProtocolNegotiator.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/ProxyAndTlsProtocolNegotiator.java
@@ -160,7 +160,7 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator
         @Override
         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
             if (msg instanceof HAProxyMessage) {
-                replaceEventWithMessage((HAProxyMessage) msg);
+                handleWithMessage((HAProxyMessage) msg);
                 ctx.fireUserEventTriggered(pne);
             } else {
                 super.channelRead(ctx, msg);
@@ -174,30 +174,34 @@ public class ProxyAndTlsProtocolNegotiator implements InternalProtocolNegotiator
          *
          * @param msg
          */
-        private void replaceEventWithMessage(HAProxyMessage msg) {
-            Attributes.Builder builder = InternalProtocolNegotiationEvent.getAttributes(pne).toBuilder();
-            if (StringUtils.isNotBlank(msg.sourceAddress())) {
-                builder.set(AttributeKeys.PROXY_PROTOCOL_ADDR, msg.sourceAddress());
-            }
-            if (msg.sourcePort() > 0) {
-                builder.set(AttributeKeys.PROXY_PROTOCOL_PORT, String.valueOf(msg.sourcePort()));
-            }
-            if (StringUtils.isNotBlank(msg.destinationAddress())) {
-                builder.set(AttributeKeys.PROXY_PROTOCOL_SERVER_ADDR, msg.destinationAddress());
-            }
-            if (msg.destinationPort() > 0) {
-                builder.set(AttributeKeys.PROXY_PROTOCOL_SERVER_PORT, String.valueOf(msg.destinationPort()));
-            }
-            if (CollectionUtils.isNotEmpty(msg.tlvs())) {
-                msg.tlvs().forEach(tlv -> {
-                    Attributes.Key<String> key = AttributeKeys.valueOf(
-                            HAProxyConstants.PROXY_PROTOCOL_TLV_PREFIX + String.format("%02x", tlv.typeByteValue()));
-                    String value = StringUtils.trim(tlv.content().toString(CharsetUtil.UTF_8));
-                    builder.set(key, value);
-                });
+        private void handleWithMessage(HAProxyMessage msg) {
+            try {
+                Attributes.Builder builder = InternalProtocolNegotiationEvent.getAttributes(pne).toBuilder();
+                if (StringUtils.isNotBlank(msg.sourceAddress())) {
+                    builder.set(AttributeKeys.PROXY_PROTOCOL_ADDR, msg.sourceAddress());
+                }
+                if (msg.sourcePort() > 0) {
+                    builder.set(AttributeKeys.PROXY_PROTOCOL_PORT, String.valueOf(msg.sourcePort()));
+                }
+                if (StringUtils.isNotBlank(msg.destinationAddress())) {
+                    builder.set(AttributeKeys.PROXY_PROTOCOL_SERVER_ADDR, msg.destinationAddress());
+                }
+                if (msg.destinationPort() > 0) {
+                    builder.set(AttributeKeys.PROXY_PROTOCOL_SERVER_PORT, String.valueOf(msg.destinationPort()));
+                }
+                if (CollectionUtils.isNotEmpty(msg.tlvs())) {
+                    msg.tlvs().forEach(tlv -> {
+                        Attributes.Key<String> key = AttributeKeys.valueOf(
+                                HAProxyConstants.PROXY_PROTOCOL_TLV_PREFIX + String.format("%02x", tlv.typeByteValue()));
+                        String value = StringUtils.trim(tlv.content().toString(CharsetUtil.UTF_8));
+                        builder.set(key, value);
+                    });
+                }
+                pne = InternalProtocolNegotiationEvent
+                        .withAttributes(InternalProtocolNegotiationEvent.getDefault(), builder.build());
+            } finally {
+                msg.release();
             }
-            pne = InternalProtocolNegotiationEvent
-                    .withAttributes(InternalProtocolNegotiationEvent.getDefault(), builder.build());
         }
     }
 
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
index 8ae87a6fa..90e358ce3 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
@@ -758,7 +758,7 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
         @Override
         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
             if (msg instanceof HAProxyMessage) {
-                fillChannelWithMessage((HAProxyMessage) msg, ctx.channel());
+                handleWithMessage((HAProxyMessage) msg, ctx.channel());
             } else {
                 super.channelRead(ctx, msg);
             }
@@ -771,26 +771,30 @@ public class NettyRemotingServer extends NettyRemotingAbstract implements Remoti
          * @param msg
          * @param channel
          */
-        private void fillChannelWithMessage(HAProxyMessage msg, Channel channel) {
-            if (StringUtils.isNotBlank(msg.sourceAddress())) {
-                channel.attr(AttributeKeys.PROXY_PROTOCOL_ADDR).set(msg.sourceAddress());
-            }
-            if (msg.sourcePort() > 0) {
-                channel.attr(AttributeKeys.PROXY_PROTOCOL_PORT).set(String.valueOf(msg.sourcePort()));
-            }
-            if (StringUtils.isNotBlank(msg.destinationAddress())) {
-                channel.attr(AttributeKeys.PROXY_PROTOCOL_SERVER_ADDR).set(msg.destinationAddress());
-            }
-            if (msg.destinationPort() > 0) {
-                channel.attr(AttributeKeys.PROXY_PROTOCOL_SERVER_PORT).set(String.valueOf(msg.destinationPort()));
-            }
-            if (CollectionUtils.isNotEmpty(msg.tlvs())) {
-                msg.tlvs().forEach(tlv -> {
-                    AttributeKey<String> key = AttributeKeys.valueOf(
-                            HAProxyConstants.PROXY_PROTOCOL_TLV_PREFIX + String.format("%02x", tlv.typeByteValue()));
-                    String value = StringUtils.trim(tlv.content().toString(CharsetUtil.UTF_8));
-                    channel.attr(key).set(value);
-                });
+        private void handleWithMessage(HAProxyMessage msg, Channel channel) {
+            try {
+                if (StringUtils.isNotBlank(msg.sourceAddress())) {
+                    channel.attr(AttributeKeys.PROXY_PROTOCOL_ADDR).set(msg.sourceAddress());
+                }
+                if (msg.sourcePort() > 0) {
+                    channel.attr(AttributeKeys.PROXY_PROTOCOL_PORT).set(String.valueOf(msg.sourcePort()));
+                }
+                if (StringUtils.isNotBlank(msg.destinationAddress())) {
+                    channel.attr(AttributeKeys.PROXY_PROTOCOL_SERVER_ADDR).set(msg.destinationAddress());
+                }
+                if (msg.destinationPort() > 0) {
+                    channel.attr(AttributeKeys.PROXY_PROTOCOL_SERVER_PORT).set(String.valueOf(msg.destinationPort()));
+                }
+                if (CollectionUtils.isNotEmpty(msg.tlvs())) {
+                    msg.tlvs().forEach(tlv -> {
+                        AttributeKey<String> key = AttributeKeys.valueOf(
+                                HAProxyConstants.PROXY_PROTOCOL_TLV_PREFIX + String.format("%02x", tlv.typeByteValue()));
+                        String value = StringUtils.trim(tlv.content().toString(CharsetUtil.UTF_8));
+                        channel.attr(key).set(value);
+                    });
+                }
+            } finally {
+                msg.release();
             }
         }
     }
-- 
2.32.0.windows.2


From 5914ff8dbb9d37e2cb48ef9c4f0256c6185b4659 Mon Sep 17 00:00:00 2001
From: lyx <56945247+lyx2000@users.noreply.github.com>
Date: Sat, 15 Jul 2023 18:15:57 +0800
Subject: [PATCH 7/8] [ISSUE #6968] fix grpc acl bug (#6969)

* feat(acl): fix acl bug

Signed-off-by: lyx <1419360299@qq.com>

# Conflicts:
#	proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java

* add access test for two client

Signed-off-by: lyx <1419360299@qq.com>

* use specific acl config

Signed-off-by: lyx <1419360299@qq.com>

* Recovering unchange file

Signed-off-by: lyx <1419360299@qq.com>

* let test pass

Signed-off-by: lyx <1419360299@qq.com>

---------

Signed-off-by: lyx <1419360299@qq.com>
---
 .../acl/plain/PlainAccessResource.java        |  21 +-
 .../acl/RemotingClientAccessTest.java         | 189 ++++++++++++++++++
 .../access_acl_conf/acl/plain_acl.yml         |  31 +++
 acl/src/test/resources/conf/acl/plain_acl.yml |   1 -
 pom.xml                                       |   1 +
 .../apache/rocketmq/proxy/ProxyStartup.java   |  17 +-
 .../proxy/grpc/GrpcServerBuilder.java         |  10 +-
 .../remoting/RemotingProtocolServer.java      |  15 +-
 8 files changed, 255 insertions(+), 30 deletions(-)
 create mode 100644 acl/src/test/java/org/apache/rocketmq/acl/RemotingClientAccessTest.java
 create mode 100644 acl/src/test/resources/access_acl_conf/acl/plain_acl.yml

diff --git a/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java b/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java
index cdbd9ea9b..72aa8ca71 100644
--- a/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java
+++ b/acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java
@@ -223,7 +223,7 @@ public class PlainAccessResource implements AccessResource {
                     if (!request.hasGroup()) {
                         throw new AclException("Consumer heartbeat doesn't have group");
                     } else {
-                        accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
+                        accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
                     }
                 }
             } else if (SendMessageRequest.getDescriptor().getFullName().equals(rpcFullName)) {
@@ -240,15 +240,15 @@ public class PlainAccessResource implements AccessResource {
                 accessResource.addResourceAndPerm(topic, Permission.PUB);
             } else if (ReceiveMessageRequest.getDescriptor().getFullName().equals(rpcFullName)) {
                 ReceiveMessageRequest request = (ReceiveMessageRequest) messageV3;
-                accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
+                accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
                 accessResource.addResourceAndPerm(request.getMessageQueue().getTopic(), Permission.SUB);
             } else if (AckMessageRequest.getDescriptor().getFullName().equals(rpcFullName)) {
                 AckMessageRequest request = (AckMessageRequest) messageV3;
-                accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
+                accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
                 accessResource.addResourceAndPerm(request.getTopic(), Permission.SUB);
             } else if (ForwardMessageToDeadLetterQueueRequest.getDescriptor().getFullName().equals(rpcFullName)) {
                 ForwardMessageToDeadLetterQueueRequest request = (ForwardMessageToDeadLetterQueueRequest) messageV3;
-                accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
+                accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
                 accessResource.addResourceAndPerm(request.getTopic(), Permission.SUB);
             } else if (EndTransactionRequest.getDescriptor().getFullName().equals(rpcFullName)) {
                 EndTransactionRequest request = (EndTransactionRequest) messageV3;
@@ -264,7 +264,7 @@ public class PlainAccessResource implements AccessResource {
                     }
                     if (command.getSettings().hasSubscription()) {
                         Subscription subscription = command.getSettings().getSubscription();
-                        accessResource.addResourceAndPerm(subscription.getGroup(), Permission.SUB);
+                        accessResource.addGroupResourceAndPerm(subscription.getGroup(), Permission.SUB);
                         for (SubscriptionEntry entry : subscription.getSubscriptionsList()) {
                             accessResource.addResourceAndPerm(entry.getTopic(), Permission.SUB);
                         }
@@ -275,17 +275,17 @@ public class PlainAccessResource implements AccessResource {
                 }
             } else if (NotifyClientTerminationRequest.getDescriptor().getFullName().equals(rpcFullName)) {
                 NotifyClientTerminationRequest request = (NotifyClientTerminationRequest) messageV3;
-                accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
+                accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
             } else if (QueryRouteRequest.getDescriptor().getFullName().equals(rpcFullName)) {
                 QueryRouteRequest request = (QueryRouteRequest) messageV3;
                 accessResource.addResourceAndPerm(request.getTopic(), Permission.ANY);
             } else if (QueryAssignmentRequest.getDescriptor().getFullName().equals(rpcFullName)) {
                 QueryAssignmentRequest request = (QueryAssignmentRequest) messageV3;
-                accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
+                accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
                 accessResource.addResourceAndPerm(request.getTopic(), Permission.SUB);
             } else if (ChangeInvisibleDurationRequest.getDescriptor().getFullName().equals(rpcFullName)) {
                 ChangeInvisibleDurationRequest request = (ChangeInvisibleDurationRequest) messageV3;
-                accessResource.addResourceAndPerm(request.getGroup(), Permission.SUB);
+                accessResource.addGroupResourceAndPerm(request.getGroup(), Permission.SUB);
                 accessResource.addResourceAndPerm(request.getTopic(), Permission.SUB);
             }
         } catch (Throwable t) {
@@ -299,6 +299,11 @@ public class PlainAccessResource implements AccessResource {
         addResourceAndPerm(resourceName, permission);
     }
 
+    private void addGroupResourceAndPerm(Resource resource, byte permission) {
+        String resourceName = NamespaceUtil.wrapNamespace(resource.getResourceNamespace(), resource.getName());
+        addResourceAndPerm(getRetryTopic(resourceName), permission);
+    }
+
     public static PlainAccessResource build(PlainAccessConfig plainAccessConfig, RemoteAddressStrategy remoteAddressStrategy) {
         PlainAccessResource plainAccessResource = new PlainAccessResource();
         plainAccessResource.setAccessKey(plainAccessConfig.getAccessKey());
diff --git a/acl/src/test/java/org/apache/rocketmq/acl/RemotingClientAccessTest.java b/acl/src/test/java/org/apache/rocketmq/acl/RemotingClientAccessTest.java
new file mode 100644
index 000000000..88c5e09a9
--- /dev/null
+++ b/acl/src/test/java/org/apache/rocketmq/acl/RemotingClientAccessTest.java
@@ -0,0 +1,189 @@
+/*
+ * 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.acl;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.AclException;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.acl.plain.AclTestHelper;
+import org.apache.rocketmq.acl.plain.PlainAccessResource;
+import org.apache.rocketmq.acl.plain.PlainAccessValidator;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class RemotingClientAccessTest {
+
+    private PlainAccessValidator plainAccessValidator;
+    private AclClientRPCHook aclClient;
+    private SessionCredentials sessionCredentials;
+
+    private File confHome;
+
+    private String clientAddress = "10.7.1.3";
+
+    @Before
+    public void init() throws IOException {
+        String folder = "access_acl_conf";
+        confHome = AclTestHelper.copyResources(folder, true);
+        System.setProperty("rocketmq.home.dir", confHome.getAbsolutePath());
+        System.setProperty("rocketmq.acl.plain.file", "/access_acl_conf/acl/plain_acl.yml".replace("/", File.separator));
+
+        plainAccessValidator = new PlainAccessValidator();
+        sessionCredentials = new SessionCredentials();
+        sessionCredentials.setAccessKey("rocketmq3");
+        sessionCredentials.setSecretKey("12345678");
+        aclClient = new AclClientRPCHook(sessionCredentials);
+    }
+
+    @After
+    public void cleanUp() {
+        AclTestHelper.recursiveDelete(confHome);
+    }
+
+    @Test(expected = AclException.class)
+    public void testProduceDenyTopic() {
+        SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+        messageRequestHeader.setTopic("topicD");
+        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
+        aclClient.doBeforeRequest(clientAddress, remotingCommand);
+
+        ByteBuffer buf = remotingCommand.encodeHeader();
+        buf.getInt();
+        buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+        buf.position(0);
+        try {
+            PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), clientAddress);
+            plainAccessValidator.validate(accessResource);
+        } catch (RemotingCommandException e) {
+            e.printStackTrace();
+            Assert.fail("Should not throw IOException");
+        }
+    }
+
+    @Test
+    public void testProduceAuthorizedTopic() {
+        SendMessageRequestHeader messageRequestHeader = new SendMessageRequestHeader();
+        messageRequestHeader.setTopic("topicA");
+        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, messageRequestHeader);
+        aclClient.doBeforeRequest(clientAddress, remotingCommand);
+
+        ByteBuffer buf = remotingCommand.encodeHeader();
+        buf.getInt();
+        buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+        buf.position(0);
+        try {
+            PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), clientAddress);
+            plainAccessValidator.validate(accessResource);
+        } catch (RemotingCommandException e) {
+            e.printStackTrace();
+            Assert.fail("Should not throw IOException");
+        }
+    }
+
+
+    @Test(expected = AclException.class)
+    public void testConsumeDenyTopic() {
+        PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
+        pullMessageRequestHeader.setTopic("topicD");
+        pullMessageRequestHeader.setConsumerGroup("groupB");
+        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
+        aclClient.doBeforeRequest("", remotingCommand);
+        ByteBuffer buf = remotingCommand.encodeHeader();
+        buf.getInt();
+        buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+        buf.position(0);
+        try {
+            PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
+            plainAccessValidator.validate(accessResource);
+        } catch (RemotingCommandException e) {
+            e.printStackTrace();
+            Assert.fail("Should not throw IOException");
+        }
+
+    }
+
+    @Test
+    public void testConsumeAuthorizedTopic() {
+        PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
+        pullMessageRequestHeader.setTopic("topicB");
+        pullMessageRequestHeader.setConsumerGroup("groupB");
+        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
+        aclClient.doBeforeRequest("", remotingCommand);
+        ByteBuffer buf = remotingCommand.encodeHeader();
+        buf.getInt();
+        buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+        buf.position(0);
+        try {
+            PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
+            plainAccessValidator.validate(accessResource);
+        } catch (RemotingCommandException e) {
+            e.printStackTrace();
+            Assert.fail("Should not throw IOException");
+        }
+    }
+
+    @Test(expected = AclException.class)
+    public void testConsumeInDeniedGroup() {
+        PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
+        pullMessageRequestHeader.setTopic("topicB");
+        pullMessageRequestHeader.setConsumerGroup("groupD");
+        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
+        aclClient.doBeforeRequest("", remotingCommand);
+        ByteBuffer buf = remotingCommand.encodeHeader();
+        buf.getInt();
+        buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+        buf.position(0);
+        try {
+            PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
+            plainAccessValidator.validate(accessResource);
+        } catch (RemotingCommandException e) {
+            e.printStackTrace();
+            Assert.fail("Should not throw IOException");
+        }
+    }
+
+    @Test
+    public void testConsumeInAuthorizedGroup() {
+        PullMessageRequestHeader pullMessageRequestHeader = new PullMessageRequestHeader();
+        pullMessageRequestHeader.setTopic("topicB");
+        pullMessageRequestHeader.setConsumerGroup("groupB");
+        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pullMessageRequestHeader);
+        aclClient.doBeforeRequest("", remotingCommand);
+        ByteBuffer buf = remotingCommand.encodeHeader();
+        buf.getInt();
+        buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+        buf.position(0);
+        try {
+            PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "123.4.5.6");
+            plainAccessValidator.validate(accessResource);
+        } catch (RemotingCommandException e) {
+            e.printStackTrace();
+            Assert.fail("Should not throw IOException");
+        }
+    }
+
+}
diff --git a/acl/src/test/resources/access_acl_conf/acl/plain_acl.yml b/acl/src/test/resources/access_acl_conf/acl/plain_acl.yml
new file mode 100644
index 000000000..28a8c4888
--- /dev/null
+++ b/acl/src/test/resources/access_acl_conf/acl/plain_acl.yml
@@ -0,0 +1,31 @@
+# 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.
+
+accounts:
+  - accessKey: rocketmq3
+    secretKey: 12345678
+    admin: false
+    defaultTopicPerm: DENY
+    defaultGroupPerm: DENY
+    topicPerms:
+      - topicA=PUB
+      - topicB=SUB
+      - topicC=PUB|SUB
+      - topicD=DENY
+    groupPerms:
+      - groupB=SUB
+      - groupC=PUB|SUB
+      - groupD=DENY
+
diff --git a/acl/src/test/resources/conf/acl/plain_acl.yml b/acl/src/test/resources/conf/acl/plain_acl.yml
index 5641a94bf..34e46696d 100644
--- a/acl/src/test/resources/conf/acl/plain_acl.yml
+++ b/acl/src/test/resources/conf/acl/plain_acl.yml
@@ -41,4 +41,3 @@ accounts:
     whiteRemoteAddress: 192.168.1.*
     # if it is admin, it could access all resources
     admin: true
-
diff --git a/pom.xml b/pom.xml
index 12bc2dbd5..4d5dd1dec 100644
--- a/pom.xml
+++ b/pom.xml
@@ -147,6 +147,7 @@
         <awaitility.version>4.1.0</awaitility.version>
         <truth.version>0.30</truth.version>
         <s3mock-junit4.version>2.11.0</s3mock-junit4.version>
+        <rocketmq-client-java.version>5.0.5</rocketmq-client-java.version>
 
         <!-- Build plugin dependencies -->
         <versions-maven-plugin.version>2.2</versions-maven-plugin.version>
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
index ea13bb808..06d5f4525 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
@@ -29,11 +29,14 @@ import org.apache.commons.cli.DefaultParser;
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.Options;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.AccessValidator;
+import org.apache.rocketmq.acl.plain.PlainAccessValidator;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.BrokerStartup;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
+import org.apache.rocketmq.common.utils.ServiceProvider;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 import org.apache.rocketmq.common.utils.AbstractStartAndShutdown;
@@ -75,16 +78,17 @@ public class ProxyStartup {
 
             MessagingProcessor messagingProcessor = createMessagingProcessor();
 
+            List<AccessValidator> accessValidators = loadAccessValidators();
             // create grpcServer
             GrpcServer grpcServer = GrpcServerBuilder.newBuilder(executor, ConfigurationManager.getProxyConfig().getGrpcServerPort())
                 .addService(createServiceProcessor(messagingProcessor))
                 .addService(ChannelzService.newInstance(100))
                 .addService(ProtoReflectionService.newInstance())
-                .configInterceptor()
+                .configInterceptor(accessValidators)
                 .build();
             PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(grpcServer);
 
-            RemotingProtocolServer remotingServer = new RemotingProtocolServer(messagingProcessor);
+            RemotingProtocolServer remotingServer = new RemotingProtocolServer(messagingProcessor, accessValidators);
             PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(remotingServer);
 
             // start servers one by one.
@@ -109,6 +113,15 @@ public class ProxyStartup {
         log.info(new Date() + " rocketmq-proxy startup successfully");
     }
 
+    protected static List<AccessValidator> loadAccessValidators() {
+        List<AccessValidator> accessValidators = ServiceProvider.load(AccessValidator.class);
+        if (accessValidators.isEmpty()) {
+            log.info("ServiceProvider loaded no AccessValidator, using default org.apache.rocketmq.acl.plain.PlainAccessValidator");
+            accessValidators.add(new PlainAccessValidator());
+        }
+        return accessValidators;
+    }
+
     protected static void initConfiguration(CommandLineArgument commandLineArgument) throws Exception {
         if (StringUtils.isNotBlank(commandLineArgument.getProxyConfigPath())) {
             System.setProperty(Configuration.CONFIG_PATH_PROPERTY, commandLineArgument.getProxyConfigPath());
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java
index 437b9216b..9cddd3013 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java
@@ -28,9 +28,7 @@ import java.util.List;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.acl.AccessValidator;
-import org.apache.rocketmq.acl.plain.PlainAccessValidator;
 import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.common.utils.ServiceProvider;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
@@ -98,14 +96,8 @@ public class GrpcServerBuilder {
         return new GrpcServer(this.serverBuilder.build());
     }
 
-    public GrpcServerBuilder configInterceptor() {
+    public GrpcServerBuilder configInterceptor(List<AccessValidator> accessValidators) {
         // grpc interceptors, including acl, logging etc.
-        List<AccessValidator> accessValidators = ServiceProvider.load(AccessValidator.class);
-        if (accessValidators.isEmpty()) {
-            log.info("ServiceProvider loaded no AccessValidator, using default org.apache.rocketmq.acl.plain.PlainAccessValidator");
-            accessValidators.add(new PlainAccessValidator());
-        }
-
         this.serverBuilder.intercept(new AuthenticationInterceptor(accessValidators));
 
         this.serverBuilder
diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java
index f08094c16..bcc9edd09 100644
--- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java
+++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java
@@ -19,7 +19,6 @@ package org.apache.rocketmq.proxy.remoting;
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.netty.channel.Channel;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
@@ -28,15 +27,14 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.acl.AccessValidator;
-import org.apache.rocketmq.acl.plain.PlainAccessValidator;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.future.FutureTaskExt;
 import org.apache.rocketmq.common.thread.ThreadPoolMonitor;
 import org.apache.rocketmq.common.thread.ThreadPoolStatusMonitor;
+import org.apache.rocketmq.common.utils.StartAndShutdown;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
-import org.apache.rocketmq.common.utils.StartAndShutdown;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
 import org.apache.rocketmq.proxy.config.ProxyConfig;
 import org.apache.rocketmq.proxy.processor.MessagingProcessor;
@@ -86,11 +84,11 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu
     protected final ThreadPoolExecutor defaultExecutor;
     protected final ScheduledExecutorService timerExecutor;
 
-    public RemotingProtocolServer(MessagingProcessor messagingProcessor) {
+    public RemotingProtocolServer(MessagingProcessor messagingProcessor, List<AccessValidator> accessValidators) {
         this.messagingProcessor = messagingProcessor;
         this.remotingChannelManager = new RemotingChannelManager(this, messagingProcessor.getProxyRelayService());
 
-        RequestPipeline pipeline = createRequestPipeline();
+        RequestPipeline pipeline = createRequestPipeline(accessValidators);
         this.getTopicRouteActivity = new GetTopicRouteActivity(pipeline, messagingProcessor);
         this.clientManagerActivity = new ClientManagerActivity(pipeline, messagingProcessor, remotingChannelManager);
         this.consumerManagerActivity = new ConsumerManagerActivity(pipeline, messagingProcessor);
@@ -254,15 +252,12 @@ public class RemotingProtocolServer implements StartAndShutdown, RemotingProxyOu
         return future;
     }
 
-    protected RequestPipeline createRequestPipeline() {
+    protected RequestPipeline createRequestPipeline(List<AccessValidator> accessValidators) {
         RequestPipeline pipeline = (ctx, request, context) -> {
         };
-
-        List<AccessValidator> accessValidatorList = new ArrayList<>();
-        accessValidatorList.add(new PlainAccessValidator());
         // add pipeline
         // the last pipe add will execute at the first
-        return pipeline.pipe(new AuthenticationPipeline(accessValidatorList));
+        return pipeline.pipe(new AuthenticationPipeline(accessValidators));
     }
 
     protected class ThreadPoolHeadSlowTimeMillsMonitor implements ThreadPoolStatusMonitor {
-- 
2.32.0.windows.2


From 440be1ed4ce2af0ab58af6c3019de7075c09c20f Mon Sep 17 00:00:00 2001
From: fuyou001 <yubao.fyb@alibaba-inc.com>
Date: Mon, 17 Jul 2023 19:23:23 +0800
Subject: [PATCH 8/8] [ISSUE #7031] fix Pop caused broker memory leak bug
 (#7032)

---
 .../broker/processor/PopBufferMergeService.java | 17 ++++++++++++++++-
 .../broker/processor/PopMessageProcessor.java   | 11 ++++++-----
 2 files changed, 22 insertions(+), 6 deletions(-)

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
index d7bc7c694..b7ba8ad4a 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopBufferMergeService.java
@@ -429,9 +429,16 @@ public class PopBufferMergeService extends ServiceThread {
      * @param nextBeginOffset
      * @return
      */
-    public void addCkJustOffset(PopCheckPoint point, int reviveQueueId, long reviveQueueOffset, long nextBeginOffset) {
+    public boolean addCkJustOffset(PopCheckPoint point, int reviveQueueId, long reviveQueueOffset, long nextBeginOffset) {
         PopCheckPointWrapper pointWrapper = new PopCheckPointWrapper(reviveQueueId, reviveQueueOffset, point, nextBeginOffset, true);
 
+        if (this.buffer.containsKey(pointWrapper.getMergeKey())) {
+            // when mergeKey conflict
+            // will cause PopBufferMergeService.scanCommitOffset cannot poll PopCheckPointWrapper
+            POP_LOGGER.warn("[PopBuffer]mergeKey conflict when add ckJustOffset. ck:{}, mergeKey:{}", pointWrapper, pointWrapper.getMergeKey());
+            return false;
+        }
+
         this.putCkToStore(pointWrapper, !checkQueueOk(pointWrapper));
 
         putOffsetQueue(pointWrapper);
@@ -440,6 +447,7 @@ public class PopBufferMergeService extends ServiceThread {
         if (brokerController.getBrokerConfig().isEnablePopLog()) {
             POP_LOGGER.info("[PopBuffer]add ck just offset, {}", pointWrapper);
         }
+        return  true;
     }
 
     public void addCkMock(String group, String topic, int queueId, long startOffset, long invisibleTime,
@@ -492,6 +500,13 @@ public class PopBufferMergeService extends ServiceThread {
             return false;
         }
 
+        if (this.buffer.containsKey(pointWrapper.getMergeKey())) {
+            // when mergeKey conflict
+            // will cause PopBufferMergeService.scanCommitOffset cannot poll PopCheckPointWrapper
+            POP_LOGGER.warn("[PopBuffer]mergeKey conflict when add ck. ck:{}, mergeKey:{}", pointWrapper, pointWrapper.getMergeKey());
+            return false;
+        }
+
         putOffsetQueue(pointWrapper);
         this.buffer.put(pointWrapper.getMergeKey(), pointWrapper);
         this.counter.incrementAndGet();
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
index 28549bfed..464f8f4fd 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
@@ -570,7 +570,9 @@ public class PopMessageProcessor implements NettyRequestProcessor {
                         this.brokerController.getConsumerOffsetManager().commitOffset(channel.remoteAddress().toString(),
                             requestHeader.getConsumerGroup(), topic, queueId, finalOffset);
                     } else {
-                        appendCheckPoint(requestHeader, topic, reviveQid, queueId, finalOffset, result, popTime, this.brokerController.getBrokerConfig().getBrokerName());
+                        if (!appendCheckPoint(requestHeader, topic, reviveQid, queueId, finalOffset, result, popTime, this.brokerController.getBrokerConfig().getBrokerName())) {
+                            return atomicRestNum.get() + result.getMessageCount();
+                        }
                     }
                     ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, isRetry, queueId, finalOffset);
                     ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, isRetry, queueId,
@@ -685,7 +687,7 @@ public class PopMessageProcessor implements NettyRequestProcessor {
         return msgInner;
     }
 
-    private void appendCheckPoint(final PopMessageRequestHeader requestHeader,
+    private boolean appendCheckPoint(final PopMessageRequestHeader requestHeader,
         final String topic, final int reviveQid, final int queueId, final long offset,
         final GetMessageResult getMessageTmpResult, final long popTime, final String brokerName) {
         // add check point msg to revive log
@@ -708,10 +710,9 @@ public class PopMessageProcessor implements NettyRequestProcessor {
         );
 
         if (addBufferSuc) {
-            return;
+            return true;
         }
-
-        this.popBufferMergeService.addCkJustOffset(
+        return this.popBufferMergeService.addCkJustOffset(
             ck, reviveQid, -1, getMessageTmpResult.getNextBeginOffset()
         );
     }
-- 
2.32.0.windows.2