Skip to content

Commit

Permalink
1. add channel id in log
Browse files Browse the repository at this point in the history 
2. fix go away
  • Loading branch information
@qianye1001
qianye1001 committed Sep 12, 2024
1 parent 581f08f commit de3187a
Showing 1 changed file with 21 additions and 19 deletions.
40 changes: 21 additions & 19 deletions remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -415,13 +415,13 @@ public void closeChannel(final String addr, final Channel channel) {
boolean removeItemFromTable = true;
final ChannelWrapper prevCW = this.channelTables.get(addrRemote);

LOGGER.info("closeChannel: begin close the channel[{}], channelId={} Found: {}", addrRemote, channel.id(), prevCW != null);
LOGGER.info("closeChannel: begin close the channel[addr={}, id={}] Found: {}", addrRemote, channel.id(), prevCW != null);

if (null == prevCW) {
LOGGER.info("closeChannel: the channel[{}], channelId={} has been removed from the channel table before", addrRemote, channel.id());
LOGGER.info("closeChannel: the channel[addr={}, id={}] has been removed from the channel table before", addrRemote, channel.id());
removeItemFromTable = false;
} else if (prevCW.isWrapperOf(channel)) {
LOGGER.info("closeChannel: the channel[{}], channelId={} has been closed before, and has been created again, nothing to do.",
LOGGER.info("closeChannel: the channel[addr={}, id={}] has been closed before, and has been created again, nothing to do.",
addrRemote, channel.id());
removeItemFromTable = false;
}
Expand All @@ -431,7 +431,7 @@ public void closeChannel(final String addr, final Channel channel) {
if (channelWrapper != null && channelWrapper.tryClose(channel)) {
this.channelTables.remove(addrRemote);
}
LOGGER.info("closeChannel: the channel[{}], channelId={}, was removed from channel table", addrRemote, channel.id());
LOGGER.info("closeChannel: the channel[addr={}, id={}] was removed from channel table", addrRemote, channel.id());
}

RemotingHelper.closeChannel(channel);
Expand All @@ -456,6 +456,7 @@ public void closeChannel(final Channel channel) {
try {
if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
try {
boolean removeItemFromTable = true;
ChannelWrapper prevCW = null;
String addrRemote = null;
for (Map.Entry<String, ChannelWrapper> entry : channelTables.entrySet()) {
Expand All @@ -470,18 +471,19 @@ public void closeChannel(final Channel channel) {

if (null == prevCW) {
LOGGER.info("eventCloseChannel: the channel[addr={}, id={}] has been removed from the channel table before", RemotingHelper.parseChannelRemoteAddr(channel), channel.id());
return;
}

ChannelWrapper channelWrapper = this.channelWrapperTables.remove(channel);
if (channelWrapper != null && channelWrapper.tryClose(channel)) {
this.channelTables.remove(addrRemote);
removeItemFromTable = false;
}
LOGGER.info("closeChannel: the channel[addr={}, id={}] was removed from channel table", addrRemote, channel.id());
RemotingHelper.closeChannel(channel);

if (removeItemFromTable) {
ChannelWrapper channelWrapper = this.channelWrapperTables.remove(channel);
if (channelWrapper != null && channelWrapper.tryClose(channel)) {
this.channelTables.remove(addrRemote);
}
LOGGER.info("closeChannel: the channel[addr={}, id={}] was removed from channel table", addrRemote, channel.id());
RemotingHelper.closeChannel(channel);
}
} catch (Exception e) {
LOGGER.error("closeChannel: close the channel exception", e);
LOGGER.error("closeChannel: close the channel[id={}] exception", channel.id(), e);
} finally {
this.lockChannelTables.unlock();
}
Expand Down Expand Up @@ -559,9 +561,9 @@ public RemotingCommand invokeSync(String addr, final RemotingCommand request, lo
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, {}, channelId={}", timeoutMillis, channelRemoteAddr, channel.id());
LOGGER.warn("invokeSync: close socket because of timeout, {}ms, channel[addr={}, id={}]", timeoutMillis, channelRemoteAddr, channel.id());
}
LOGGER.warn("invokeSync: wait response timeout exception, the channel[{}], channelId={}", channelRemoteAddr, channel.id());
LOGGER.warn("invokeSync: wait response timeout exception, the channel[addr={}, id={}]", channelRemoteAddr, channel.id());
throw e;
}
} else {
Expand Down Expand Up @@ -1179,7 +1181,7 @@ public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws
@Override
public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
LOGGER.info("NETTY CLIENT PIPELINE: CLOSE {}, channelId={}", remoteAddress, ctx.channel().id());
LOGGER.info("NETTY CLIENT PIPELINE: CLOSE channel[addr={}, id={}]", remoteAddress, ctx.channel().id());
closeChannel(ctx.channel());
super.close(ctx, promise);
NettyRemotingClient.this.failFast(ctx.channel());
Expand All @@ -1191,7 +1193,7 @@ public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exce
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
LOGGER.info("NETTY CLIENT PIPELINE: channelInactive, the channel[{}], channelId={}", remoteAddress, ctx.channel().id());
LOGGER.info("NETTY CLIENT PIPELINE: channelInactive, the channel[addr={}, id={}]", remoteAddress, ctx.channel().id());
closeChannel(ctx.channel());
super.channelInactive(ctx);
}
Expand All @@ -1202,7 +1204,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
IdleStateEvent event = (IdleStateEvent) evt;
if (event.state().equals(IdleState.ALL_IDLE)) {
final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
LOGGER.warn("NETTY CLIENT PIPELINE: IDLE exception [{}]", remoteAddress);
LOGGER.warn("NETTY CLIENT PIPELINE: IDLE exception channel[addr={}, id={}]", remoteAddress, ctx.channel().id());
closeChannel(ctx.channel());
if (NettyRemotingClient.this.channelEventListener != null) {
NettyRemotingClient.this
Expand All @@ -1217,7 +1219,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
LOGGER.warn("NETTY CLIENT PIPELINE: exceptionCaught {}, channelId={}", remoteAddress, ctx.channel().id(), cause);
LOGGER.warn("NETTY CLIENT PIPELINE: exceptionCaught channel[addr={}, id={}]", remoteAddress, ctx.channel().id(), cause);
closeChannel(ctx.channel());
if (NettyRemotingClient.this.channelEventListener != null) {
NettyRemotingClient.this.putNettyEvent(new NettyEvent(NettyEventType.EXCEPTION, remoteAddress, ctx.channel()));
Expand Down

0 comments on commit de3187a

Please sign in to comment.