You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "SteNicholas (via GitHub)" <gi...@apache.org> on 2025/04/30 06:57:04 UTC
[PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
SteNicholas opened a new pull request, #3239:
URL: https://github.com/apache/celeborn/pull/3239
### What changes were proposed in this pull request?
Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat.
Follow up #1457.
### Why are the changes needed?
In Flink batch jobs, the following exception is caused by closed connection:
```
2025-04-27 23:30:28
java.io.IOException: Client /:9093 is lost, notify related stream 805472050177
at org.apache.celeborn.plugin.flink.RemoteBufferStreamReader.errorReceived(RemoteBufferStreamReader.java:146)
at org.apache.celeborn.plugin.flink.RemoteBufferStreamReader.lambda$new$0(RemoteBufferStreamReader.java:77)
at org.apache.celeborn.plugin.flink.network.ReadClientHandler.processMessageInternal(ReadClientHandler.java:64)
at org.apache.celeborn.plugin.flink.network.ReadClientHandler.lambda$channelInactive$0(ReadClientHandler.java:145)
at java.base/java.util.concurrent.ConcurrentHashMap.forEach(ConcurrentHashMap.java:1603)
at org.apache.celeborn.plugin.flink.network.ReadClientHandler.channelInactive(ReadClientHandler.java:136)
at org.apache.celeborn.common.network.server.TransportRequestHandler.channelInactive(TransportRequestHandler.java:74)
at org.apache.celeborn.common.network.server.TransportChannelHandler.channelInactive(TransportChannelHandler.java:141)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
at org.apache.celeborn.shaded.io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81)
at org.apache.celeborn.common.network.client.ReconnectHandler.scheduleReconnect(ReconnectHandler.java:93)
at org.apache.celeborn.common.network.client.ReconnectHandler.channelInactive(ReconnectHandler.java:63)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
at org.apache.celeborn.shaded.io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81)
at org.apache.celeborn.shaded.io.netty.handler.timeout.IdleStateHandler.channelInactive(IdleStateHandler.java:280)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
at org.apache.celeborn.shaded.io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81)
at org.apache.celeborn.plugin.flink.network.TransportFrameDecoderWithBufferSupplier.channelInactive(TransportFrameDecoderWithBufferSupplier.java:207)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
at org.apache.celeborn.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1405)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:301)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
at org.apache.celeborn.shaded.io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:901)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:813)
at org.apache.celeborn.shaded.io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
at org.apache.celeborn.shaded.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
at org.apache.celeborn.shaded.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
at org.apache.celeborn.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
at org.apache.celeborn.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at org.apache.celeborn.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at org.apache.celeborn.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:991)
```
The closed connection is caused by reading idleness of worker which enables heartbeat with troubleshooting via debug mode of log.
```
2025-04-27 23:30:32,341 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] WRITE: MessageWithHeader [headerLength: 17, bodyLength: 26]
2025-04-27 23:30:32,341 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] FLUSH
2025-04-27 23:30:32,380 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] READ 38B
2025-04-27 23:30:32,380 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] READ COMPLETE
2025-04-27 23:31:31,813 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] WRITE 10B
2025-04-27 23:31:31,813 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] FLUSH
2025-04-27 23:32:31,823 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] WRITE 10B
2025-04-27 23:32:31,824 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] FLUSH
2025-04-27 23:33:31,826 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] WRITE 10B
2025-04-27 23:33:31,826 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] FLUSH
2025-04-27 23:34:31,826 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] WRITE 10B
2025-04-27 23:34:31,826 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] FLUSH
2025-04-27 23:34:32,380 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 - R:/33.133.79.187:44862] CLOSE
2025-04-27 23:34:32,380 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 ! R:/33.133.79.187:44862] INACTIVE
2025-04-27 23:34:32,380 [fetch-server-11-5] DEBUG util.NettyLogger - [id: 0x2dc85987, L:/:9093 ! R:/33.133.79.187:44862] UNREGISTERED
```
The reading idleness of worker which enables heartbeat is resulted via one-way heartbeat from worker to client, which only keeps the channel of client active. Client should handle heartbeat to keep the channel of worker active via sending heartbeat to worker.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
`HeartbeatTest`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3239:
URL: https://github.com/apache/celeborn/pull/3239#discussion_r2071167489
##########
common/src/main/java/org/apache/celeborn/common/network/server/TransportChannelHandler.java:
##########
@@ -152,8 +153,18 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
@Override
public void channelRead(ChannelHandlerContext ctx, Object request) throws Exception {
- if (request instanceof RequestMessage && !(request instanceof Heartbeat)) {
- requestHandler.handle((RequestMessage) request);
+ if (request instanceof RequestMessage) {
+ if (request instanceof Heartbeat) {
+ logger.debug("Received heartbeat from {}.", NettyUtils.getRemoteAddress(ctx.channel()));
+ if (!enableHeartbeat) {
+ // Which disables heartbeat should send heartbeat for processing heartbeat to avoid
+ // reading idleness of which enables heartbeat.
+ requestHandler.processHeartbeat();
Review Comment:
@mridulm, there is no case that both sides constantly sending Heartbeat's to each other. I could give the following situations:
1. worker disables heartbeat and client disables heartbeat: the heartbeat message does not send.
2. worker enables heartbeat and client disables heartbeat: client invokes `processHeartbeat` and `fireChannelRead`, but worker only invoke `fireChannelRead`.
3. worker disables heartbeat and client enables heartbeat: worker invokes `processHeartbeat` and `fireChannelRead`, but client only invoke `fireChannelRead`.
4. worker enables heartbeat and client enables heartbeat: worker and client only invoke `fireChannelRead`.
Therefore, the heartbeat is only used to keep both sides of conection not read idle and does not constantly send heartbeat.
BTW, the reason why `processHeartbeat` sends `Heartbeat` instead of `HeartbeatResponse` is that there is compatibility problem in sending `HeartbeatResponse`. `TransportResponseHandler` handles `HeartbeatResponse` with `IllegalStateException` for unknown response type.
PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3239:
URL: https://github.com/apache/celeborn/pull/3239#discussion_r2071167489
##########
common/src/main/java/org/apache/celeborn/common/network/server/TransportChannelHandler.java:
##########
@@ -152,8 +153,18 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
@Override
public void channelRead(ChannelHandlerContext ctx, Object request) throws Exception {
- if (request instanceof RequestMessage && !(request instanceof Heartbeat)) {
- requestHandler.handle((RequestMessage) request);
+ if (request instanceof RequestMessage) {
+ if (request instanceof Heartbeat) {
+ logger.debug("Received heartbeat from {}.", NettyUtils.getRemoteAddress(ctx.channel()));
+ if (!enableHeartbeat) {
+ // Which disables heartbeat should send heartbeat for processing heartbeat to avoid
+ // reading idleness of which enables heartbeat.
+ requestHandler.processHeartbeat();
Review Comment:
@mridulm, there is no case that both sides constantly sending Heartbeat's to each other. I could give the following situations:
1. worker disables heartbeat and client disables heartbeat: the heartbeat message does not send.
2. worker enables heartbeat and client disables heartbeat: client invokes `processHeartbeat` and `fireChannelRead`, but worker only invoke `fireChannelRead`.
3. worker disables heartbeat and client enables heartbeat: worker invokes `processHeartbeat` and `fireChannelRead`, but client only invoke `fireChannelRead`.
4. worker enables heartbeat and client enables heartbeat: worker and client only invoke `fireChannelRead`.
Therefore, the heartbeat is only used to keep both sides of conection not read idle and does not constantly send heartbeat. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
Posted by "codenohup (via GitHub)" <gi...@apache.org>.
codenohup commented on PR #3239:
URL: https://github.com/apache/celeborn/pull/3239#issuecomment-2841113543
Hi, @SteNicholas
Thank you for your contribution@ Our Celeborn cluster has also encountered this problem. We hope that this PR can solve them. Overall, this PR is fine. I have a small question, PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #3239:
URL: https://github.com/apache/celeborn/pull/3239#discussion_r2069077363
##########
common/src/main/java/org/apache/celeborn/common/network/server/TransportChannelHandler.java:
##########
@@ -152,8 +153,18 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
@Override
public void channelRead(ChannelHandlerContext ctx, Object request) throws Exception {
- if (request instanceof RequestMessage && !(request instanceof Heartbeat)) {
- requestHandler.handle((RequestMessage) request);
+ if (request instanceof RequestMessage) {
+ if (request instanceof Heartbeat) {
+ logger.debug("Received heartbeat from {}.", NettyUtils.getRemoteAddress(ctx.channel()));
+ if (!enableHeartbeat) {
+ // Which disables heartbeat should send heartbeat for processing heartbeat to avoid
+ // reading idleness of which enables heartbeat.
Review Comment:
This is a bit confusing.
Perhaps something like this ?
```suggestion
// When heartbeat is disabled, we should still response to a heartbeat if peer has
// heartbeat enabled - to present reading idleness
```
##########
common/src/main/java/org/apache/celeborn/common/network/server/TransportChannelHandler.java:
##########
@@ -152,8 +153,18 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
@Override
public void channelRead(ChannelHandlerContext ctx, Object request) throws Exception {
- if (request instanceof RequestMessage && !(request instanceof Heartbeat)) {
- requestHandler.handle((RequestMessage) request);
+ if (request instanceof RequestMessage) {
+ if (request instanceof Heartbeat) {
+ logger.debug("Received heartbeat from {}.", NettyUtils.getRemoteAddress(ctx.channel()));
+ if (!enableHeartbeat) {
+ // Which disables heartbeat should send heartbeat for processing heartbeat to avoid
+ // reading idleness of which enables heartbeat.
+ requestHandler.processHeartbeat();
Review Comment:
`processHeartbeat` always results in sending a `Heartbeat` as response (not `HeartbeatResponse`).
Are we sure this does not result in both sides constantly sending Heartbeat's to each other ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
Posted by "codenohup (via GitHub)" <gi...@apache.org>.
codenohup commented on code in PR #3239:
URL: https://github.com/apache/celeborn/pull/3239#discussion_r2068080441
##########
common/src/main/java/org/apache/celeborn/common/network/server/TransportChannelHandler.java:
##########
@@ -113,7 +113,7 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.executor()
.scheduleWithFixedDelay(
() -> {
- logger.debug("send heartbeat");
+ logger.debug("Send heartbeat to {}.", ctx.channel().remoteAddress());
Review Comment:
`NettyUtils.getRemoteAddress(ctx.channel())`?
##########
common/src/main/java/org/apache/celeborn/common/network/server/TransportRequestHandler.java:
##########
@@ -95,6 +95,18 @@ public void handle(RequestMessage request) {
}
}
+ protected void processHeartbeat() {
+ try {
+ logger.trace("Process heartbeat from {}.", NettyUtils.getRemoteAddress(channel));
+ respond(new Heartbeat());
+ } catch (Exception e) {
+ logger.error(
+ "Error while invoking handler#receive() on heartbeat from {}.",
Review Comment:
Confusing error message
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3239:
URL: https://github.com/apache/celeborn/pull/3239#issuecomment-2846492202
@mridulm, thanks for review. I have addressed above comments. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #3239:
URL: https://github.com/apache/celeborn/pull/3239#discussion_r2071174660
##########
common/src/main/java/org/apache/celeborn/common/network/server/TransportChannelHandler.java:
##########
@@ -152,8 +153,18 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
@Override
public void channelRead(ChannelHandlerContext ctx, Object request) throws Exception {
- if (request instanceof RequestMessage && !(request instanceof Heartbeat)) {
- requestHandler.handle((RequestMessage) request);
+ if (request instanceof RequestMessage) {
+ if (request instanceof Heartbeat) {
+ logger.debug("Received heartbeat from {}.", NettyUtils.getRemoteAddress(ctx.channel()));
+ if (!enableHeartbeat) {
+ // Which disables heartbeat should send heartbeat for processing heartbeat to avoid
+ // reading idleness of which enables heartbeat.
+ requestHandler.processHeartbeat();
Review Comment:
I was trying to understand if case 4 had corner cases. Makes sense, thanks for clarifying !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3239:
URL: https://github.com/apache/celeborn/pull/3239#issuecomment-2841353164
Ping @FMX, @RexXiong.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat [celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX closed pull request #3239: [CELEBORN-1912] Client should send heartbeat to worker for processing heartbeat to avoid reading idleness of worker which enables heartbeat
URL: https://github.com/apache/celeborn/pull/3239
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org