Merge pull request #8346 from volodymyr-babak/keep-alive-updates

Improved Keep Alive Functionality between Edge and Cloud to Prevent Data Loss

application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java

@@ -98,11 +98,14 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
     private String privateKeyResource;
     @Value("${edges.state.persistToTelemetry:false}")
     private boolean persistToTelemetry;
-    @Value("${edges.rpc.client_max_keep_alive_time_sec}")
+    @Value("${edges.rpc.client_max_keep_alive_time_sec:1}")
     private int clientMaxKeepAliveTimeSec;
     @Value("${edges.rpc.max_inbound_message_size:4194304}")
     private int maxInboundMessageSize;
-
+    @Value("${edges.rpc.keep_alive_time_sec:10}")
+    private int keepAliveTimeSec;
+    @Value("${edges.rpc.keep_alive_timeout_sec:5}")
+    private int keepAliveTimeoutSec;
     @Value("${edges.scheduler_pool_size}")
     private int schedulerPoolSize;
 
@@ -131,6 +134,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
         log.info("Initializing Edge RPC service!");
         NettyServerBuilder builder = NettyServerBuilder.forPort(rpcPort)
                 .permitKeepAliveTime(clientMaxKeepAliveTimeSec, TimeUnit.SECONDS)
+                .keepAliveTime(keepAliveTimeSec, TimeUnit.SECONDS)
+                .keepAliveTimeout(keepAliveTimeoutSec, TimeUnit.SECONDS)
+                .permitKeepAliveWithoutCalls(true)
                 .maxInboundMessageSize(maxInboundMessageSize)
                 .addService(this);
         if (sslEnabled) {

application/src/main/resources/thingsboard.yml

@@ -949,7 +949,9 @@ edges:
   enabled: "${EDGES_ENABLED:true}"
   rpc:
     port: "${EDGES_RPC_PORT:7070}"
-    client_max_keep_alive_time_sec: "${EDGES_RPC_CLIENT_MAX_KEEP_ALIVE_TIME_SEC:300}"
+    client_max_keep_alive_time_sec: "${EDGES_RPC_CLIENT_MAX_KEEP_ALIVE_TIME_SEC:1}"
+    keep_alive_time_sec: "${EDGES_RPC_KEEP_ALIVE_TIME_SEC:10}"
+    keep_alive_timeout_sec: "${EDGES_RPC_KEEP_ALIVE_TIMEOUT_SEC:5}"
     ssl:
       # Enable/disable SSL support
       enabled: "${EDGES_RPC_SSL_ENABLED:false}"

application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java

@@ -108,6 +108,7 @@ public class EdgeImitator {
         updateEdgeClientFields("rpcPort", port);
         updateEdgeClientFields("timeoutSecs", 3);
         updateEdgeClientFields("keepAliveTimeSec", 300);
+        updateEdgeClientFields("keepAliveTimeoutSec", 5);
         updateEdgeClientFields("maxInboundMessageSize", 4194304);
     }
 

common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java

@@ -57,8 +57,10 @@ public class EdgeGrpcClient implements EdgeRpcClient {
     private int rpcPort;
     @Value("${cloud.rpc.timeout}")
     private int timeoutSecs;
-    @Value("${cloud.rpc.keep_alive_time_sec}")
+    @Value("${cloud.rpc.keep_alive_time_sec:10}")
     private int keepAliveTimeSec;
+    @Value("${cloud.rpc.keep_alive_timeout_sec:5}")
+    private int keepAliveTimeoutSec;
     @Value("${cloud.rpc.ssl.enabled}")
     private boolean sslEnabled;
     @Value("${cloud.rpc.ssl.cert:}")
@@ -83,7 +85,9 @@ public class EdgeGrpcClient implements EdgeRpcClient {
                         Consumer<Exception> onError) {
         NettyChannelBuilder builder = NettyChannelBuilder.forAddress(rpcHost, rpcPort)
                 .maxInboundMessageSize(maxInboundMessageSize)
-                .keepAliveTime(keepAliveTimeSec, TimeUnit.SECONDS);
+                .keepAliveTime(keepAliveTimeSec, TimeUnit.SECONDS)
+                .keepAliveTimeout(keepAliveTimeoutSec, TimeUnit.SECONDS)
+                .keepAliveWithoutCalls(true);
         if (sslEnabled) {
             try {
                 SslContextBuilder sslContextBuilder = GrpcSslContexts.forClient();