opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix LEAK: HAProxyMessage.release() was not called before it's garbage-collected (#7025)

Browse Source 
Call HAProxyMessage.release() after reading it.

---------

Co-authored-by: 徒钟 <shuangxi.dsx@alibaba-inc.com>
This commit is contained in:
ShuangxiDing
2023-07-14 16:46:40 +08:00
committed by GitHub
parent 33cb22e1c0
commit 70a66eda2c
2 changed files with 53 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/ProxyAndTlsProtocolNegotiator.java
+28 -24
View File
@@ -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());
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();
}
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());
}
}
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
+25 -21
View File
@@ -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();
}
}
}