[FIX_#3789][remote]cherry pick from dev to support netty heart beat (#3913)

* [FIX_#3789][remote]cherry pick from dev to support netty heart beat

* [FIX_#3789][remote]cherry pick from dev to support netty heart beat

Co-authored-by: Kirs <acm_master@163.com>

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingClient.java

@@ -18,10 +18,16 @@
 package org.apache.dolphinscheduler.remote;
 
 import io.netty.bootstrap.Bootstrap;
-import io.netty.channel.*;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
+
 import org.apache.dolphinscheduler.remote.codec.NettyDecoder;
 import org.apache.dolphinscheduler.remote.codec.NettyEncoder;
 import org.apache.dolphinscheduler.remote.command.Command;
@@ -38,6 +44,8 @@ import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
 import org.apache.dolphinscheduler.remote.utils.Host;
 import org.apache.dolphinscheduler.remote.utils.CallerThreadExecutePolicy;
 import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory;
+import org.apache.dolphinscheduler.remote.utils.NettyUtils;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -76,7 +84,7 @@ public class NettyRemotingClient {
     /**
      * worker group
      */
-    private final NioEventLoopGroup workerGroup;
+    private final EventLoopGroup workerGroup;
 
     /**
      * client config
@@ -105,10 +113,21 @@ public class NettyRemotingClient {
 
     /**
      * client init
+     *
      * @param clientConfig client config
      */
     public NettyRemotingClient(final NettyClientConfig clientConfig) {
         this.clientConfig = clientConfig;
+        if (NettyUtils.useEpoll()) {
+            this.workerGroup = new EpollEventLoopGroup(clientConfig.getWorkerThreads(), new ThreadFactory() {
+                private AtomicInteger threadIndex = new AtomicInteger(0);
+
+                @Override
+                public Thread newThread(Runnable r) {
+                    return new Thread(r, String.format("NettyClient_%d", this.threadIndex.incrementAndGet()));
+                }
+            });
+        } else {
             this.workerGroup = new NioEventLoopGroup(clientConfig.getWorkerThreads(), new ThreadFactory() {
                 private AtomicInteger threadIndex = new AtomicInteger(0);
 
@@ -117,6 +136,7 @@ public class NettyRemotingClient {
                     return new Thread(r, String.format("NettyClient_%d", this.threadIndex.incrementAndGet()));
                 }
             });
+        }
         this.callbackExecutor = new ThreadPoolExecutor(5, 10, 1, TimeUnit.MINUTES,
             new LinkedBlockingQueue<>(1000), new NamedThreadFactory("CallbackExecutor", 10),
             new CallerThreadExecutePolicy());
@@ -134,11 +154,12 @@ public class NettyRemotingClient {
 
         this.bootstrap
             .group(this.workerGroup)
-                .channel(NioSocketChannel.class)
+            .channel(NettyUtils.getSocketChannelClass())
             .option(ChannelOption.SO_KEEPALIVE, clientConfig.isSoKeepalive())
             .option(ChannelOption.TCP_NODELAY, clientConfig.isTcpNoDelay())
             .option(ChannelOption.SO_SNDBUF, clientConfig.getSendBufferSize())
             .option(ChannelOption.SO_RCVBUF, clientConfig.getReceiveBufferSize())
+            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfig.getConnectTimeoutMillis())
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
@@ -160,6 +181,7 @@ public class NettyRemotingClient {
 
     /**
      * async send
+     *
      * @param host           host
      * @param command        command
      * @param timeoutMillis  timeoutMillis
@@ -227,6 +249,7 @@ public class NettyRemotingClient {
 
     /**
      * sync send
+     *
      * @param host          host
      * @param command       command
      * @param timeoutMillis timeoutMillis
@@ -271,6 +294,7 @@ public class NettyRemotingClient {
 
     /**
      * send task
+     *
      * @param host    host
      * @param command command
      * @throws RemotingException
@@ -297,6 +321,7 @@ public class NettyRemotingClient {
 
     /**
      * register processor
+     *
      * @param commandType command type
      * @param processor   processor
      */
@@ -317,6 +342,7 @@ public class NettyRemotingClient {
 
     /**
      * get channel
+     *
      * @param host
      * @return
      */
@@ -330,6 +356,7 @@ public class NettyRemotingClient {
 
     /**
      * create channel
+     *
      * @param host   host
      * @param isSync sync flag
      * @return channel
@@ -389,6 +416,7 @@ public class NettyRemotingClient {
 
     /**
      * close channel
+     *
      * @param host host
      */
     public void closeChannel(Host host) {

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/NettyRemotingServer.java

@@ -17,14 +17,6 @@
 
 package org.apache.dolphinscheduler.remote;
 
-import io.netty.bootstrap.ServerBootstrap;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.ChannelPipeline;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
 import org.apache.dolphinscheduler.remote.codec.NettyDecoder;
 import org.apache.dolphinscheduler.remote.codec.NettyEncoder;
 import org.apache.dolphinscheduler.remote.command.CommandType;
@@ -32,8 +24,7 @@ import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
 import org.apache.dolphinscheduler.remote.handler.NettyServerHandler;
 import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
 import org.apache.dolphinscheduler.remote.utils.Constants;
-import org.slf4j.Logger;
+import org.apache.dolphinscheduler.remote.utils.NettyUtils;
-import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -41,6 +32,19 @@ import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+
 /**
  * remoting netty server
  */
@@ -66,12 +70,12 @@ public class NettyRemotingServer {
     /**
      * boss group
      */
-    private final NioEventLoopGroup bossGroup;
+    private final EventLoopGroup bossGroup;
 
     /**
      * worker group
      */
-    private final NioEventLoopGroup workGroup;
+    private final EventLoopGroup workGroup;
 
     /**
      * server config
@@ -95,7 +99,25 @@ public class NettyRemotingServer {
      */
     public NettyRemotingServer(final NettyServerConfig serverConfig) {
         this.serverConfig = serverConfig;
+        if (NettyUtils.useEpoll()) {
+            this.bossGroup = new EpollEventLoopGroup(1, new ThreadFactory() {
+                private AtomicInteger threadIndex = new AtomicInteger(0);
 
+                @Override
+                public Thread newThread(Runnable r) {
+                    return new Thread(r, String.format("NettyServerBossThread_%d", this.threadIndex.incrementAndGet()));
+                }
+            });
+
+            this.workGroup = new EpollEventLoopGroup(serverConfig.getWorkerThread(), new ThreadFactory() {
+                private AtomicInteger threadIndex = new AtomicInteger(0);
+
+                @Override
+                public Thread newThread(Runnable r) {
+                    return new Thread(r, String.format("NettyServerWorkerThread_%d", this.threadIndex.incrementAndGet()));
+                }
+            });
+        } else {
             this.bossGroup = new NioEventLoopGroup(1, new ThreadFactory() {
                 private AtomicInteger threadIndex = new AtomicInteger(0);
 
@@ -114,29 +136,26 @@ public class NettyRemotingServer {
                 }
             });
         }
+    }
 
     /**
      * server start
      */
     public void start() {
-
+        if (isStarted.compareAndSet(false, true)) {
-        if(this.isStarted.get()){
-            return;
-        }
-
             this.serverBootstrap
                 .group(this.bossGroup, this.workGroup)
-                .channel(NioServerSocketChannel.class)
+                .channel(NettyUtils.getServerSocketChannelClass())
                 .option(ChannelOption.SO_REUSEADDR, true)
                 .option(ChannelOption.SO_BACKLOG, serverConfig.getSoBacklog())
                 .childOption(ChannelOption.SO_KEEPALIVE, serverConfig.isSoKeepalive())
                 .childOption(ChannelOption.TCP_NODELAY, serverConfig.isTcpNoDelay())
                 .childOption(ChannelOption.SO_SNDBUF, serverConfig.getSendBufferSize())
                 .childOption(ChannelOption.SO_RCVBUF, serverConfig.getReceiveBufferSize())
-                .childHandler(new ChannelInitializer<NioSocketChannel>() {
+                .childHandler(new ChannelInitializer<SocketChannel>() {
 
                     @Override
-                    protected void initChannel(NioSocketChannel ch) throws Exception {
+                    protected void initChannel(SocketChannel ch) throws Exception {
                         initNettyChannel(ch);
                     }
                 });
@@ -155,16 +174,15 @@ public class NettyRemotingServer {
             } else {
                 throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()));
             }
-        //
+        }
-        isStarted.compareAndSet(false, true);
     }
 
     /**
      * init netty channel
+     *
      * @param ch socket channel
-     * @throws Exception
      */
-    private void initNettyChannel(NioSocketChannel ch) throws Exception{
+    private void initNettyChannel(SocketChannel ch) {
         ChannelPipeline pipeline = ch.pipeline();
         pipeline.addLast("encoder", encoder);
         pipeline.addLast("decoder", new NettyDecoder());
@@ -173,6 +191,7 @@ public class NettyRemotingServer {
 
     /**
      * register processor
+     *
      * @param commandType command type
      * @param processor   processor
      */
@@ -193,6 +212,7 @@ public class NettyRemotingServer {
 
     /**
      * get default thread executor
+     *
      * @return thread executor
      */
     public ExecutorService getDefaultExecutor() {

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/CommandType.java

@@ -98,6 +98,11 @@ public enum CommandType {
      */
     TASK_KILL_RESPONSE,
 
+    /**
+     * HEART_BEAT
+     */
+    HEART_BEAT,
+
     /**
      * ping
      */

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/config/NettyClientConfig.java

@@ -48,6 +48,11 @@ public class NettyClientConfig {
      */
     private int receiveBufferSize = 65535;
 
+    /**
+     * connect timeout millis
+     */
+    private int connectTimeoutMillis = 3000;
+
     public int getWorkerThreads() {
         return workerThreads;
     }
@@ -88,4 +93,11 @@ public class NettyClientConfig {
         this.receiveBufferSize = receiveBufferSize;
     }
 
+    public int getConnectTimeoutMillis() {
+        return connectTimeoutMillis;
+    }
+
+    public void setConnectTimeoutMillis(int connectTimeoutMillis) {
+        this.connectTimeoutMillis = connectTimeoutMillis;
+    }
 }

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyClientHandler.java

@@ -14,9 +14,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.dolphinscheduler.remote.handler;
 
-import io.netty.channel.*;
 import org.apache.dolphinscheduler.remote.NettyRemotingClient;
 import org.apache.dolphinscheduler.remote.command.Command;
 import org.apache.dolphinscheduler.remote.command.CommandType;
@@ -25,14 +25,22 @@ import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
 import org.apache.dolphinscheduler.remote.utils.ChannelUtils;
 import org.apache.dolphinscheduler.remote.utils.Constants;
 import org.apache.dolphinscheduler.remote.utils.Pair;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.timeout.IdleStateEvent;
+
 /**
  * netty client request handler
  */
@@ -46,6 +54,8 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
      */
     private final NettyRemotingClient nettyRemotingClient;
 
+    private static byte[] heartBeatData = "heart_beat".getBytes();
+
     /**
      * callback thread executor
      */
@@ -164,6 +174,7 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
 
     /**
      * caught exception
+     *
      * @param ctx   channel handler context
      * @param cause cause
      * @throws Exception
@@ -175,4 +186,18 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
         ctx.channel().close();
     }
 
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof IdleStateEvent) {
+            Command heartBeat = new Command();
+            heartBeat.setType(CommandType.HEART_BEAT);
+            heartBeat.setBody(heartBeatData);
+            ctx.writeAndFlush(heartBeat)
+                .addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
+
+        } else {
+            super.userEventTriggered(ctx, evt);
+        }
+    }
+
 }

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/handler/NettyServerHandler.java

@@ -17,20 +17,28 @@
 
 package org.apache.dolphinscheduler.remote.handler;
 
-import io.netty.channel.*;
 import org.apache.dolphinscheduler.remote.NettyRemotingServer;
 import org.apache.dolphinscheduler.remote.command.Command;
 import org.apache.dolphinscheduler.remote.command.CommandType;
 import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
 import org.apache.dolphinscheduler.remote.utils.ChannelUtils;
 import org.apache.dolphinscheduler.remote.utils.Pair;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.RejectedExecutionException;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelConfig;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.timeout.IdleStateEvent;
+
+
 /**
  * netty server request handler
  */
@@ -56,6 +64,7 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
     /**
      * When the current channel is not active,
      * the current channel has reached the end of its life cycle
+     *
      * @param ctx channel handler context
      * @throws Exception
      */
@@ -103,6 +112,7 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
 
     /**
      * process received logic
+     *
      * @param channel channel
      * @param msg     message
      */
@@ -170,4 +180,13 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
             config.setAutoRead(true);
         }
     }
+
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof IdleStateEvent) {
+            ctx.channel().close();
+        } else {
+            super.userEventTriggered(ctx, evt);
+        }
+    }
 }

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/utils/Constants.java

@@ -19,7 +19,6 @@ package org.apache.dolphinscheduler.remote.utils;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 
-
 /**
  *  constant
  */
@@ -29,6 +28,10 @@ public class Constants {
 
     public static final String SLASH = "/";
 
+    public static final int NETTY_SERVER_HEART_BEAT_TIME = 1000 * 60 * 3 + 1000;
+
+    public static final int NETTY_CLIENT_HEART_BEAT_TIME = 1000 * 60;
+
     /**
      *  charset
      */
@@ -42,4 +45,14 @@ public class Constants {
 
     public static final String LOCAL_ADDRESS = IPUtils.getFirstNoLoopbackIP4Address();
 
+    /**
+     * netty epoll enable switch
+     */
+    public static final String NETTY_EPOLL_ENABLE = System.getProperty("netty.epoll.enable", "true");
+
+    /**
+     * OS Name
+     */
+    public static final String OS_NAME = System.getProperty("os.name");
+
 }

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/utils/NettyUtils.java

@@ -0,0 +1,62 @@
+/*
+ * 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.dolphinscheduler.remote.utils;
+
+import io.netty.channel.epoll.Epoll;
+import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.channel.epoll.EpollSocketChannel;
+import io.netty.channel.socket.ServerSocketChannel;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+
+/**
+ * NettyUtils
+ */
+public class NettyUtils {
+
+    private NettyUtils() {
+    }
+
+    public static boolean useEpoll() {
+        String osName = Constants.OS_NAME;
+        if (!osName.toLowerCase().contains("linux")) {
+            return false;
+        }
+        if (!Epoll.isAvailable()) {
+            return false;
+        }
+        String enableNettyEpoll = Constants.NETTY_EPOLL_ENABLE;
+        return Boolean.parseBoolean(enableNettyEpoll);
+    }
+
+    public static Class<? extends ServerSocketChannel> getServerSocketChannelClass() {
+        if (useEpoll()) {
+            return EpollServerSocketChannel.class;
+        }
+        return NioServerSocketChannel.class;
+    }
+
+    public static Class<? extends SocketChannel> getSocketChannelClass() {
+        if (useEpoll()) {
+            return EpollSocketChannel.class;
+        }
+        return NioSocketChannel.class;
+    }
+
+}