Fix process of messages that exceeds max inbound message size

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

@@ -174,7 +174,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
 
     @Override
     public StreamObserver<RequestMsg> handleMsgs(StreamObserver<ResponseMsg> outputStream) {
-        return new EdgeGrpcSession(ctx, outputStream, this::onEdgeConnect, this::onEdgeDisconnect, sendDownlinkExecutorService).getInputStream();
+        return new EdgeGrpcSession(ctx, outputStream, this::onEdgeConnect, this::onEdgeDisconnect, sendDownlinkExecutorService, this.maxInboundMessageSize).getInputStream();
     }
 
     @Override

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

@@ -105,16 +105,20 @@ public final class EdgeGrpcSession implements Closeable {
 
     private EdgeVersion edgeVersion;
 
+    private int maxInboundMessageSize;
+    private int clientMaxInboundMessageSize;
+
     private ScheduledExecutorService sendDownlinkExecutorService;
 
     EdgeGrpcSession(EdgeContextComponent ctx, StreamObserver<ResponseMsg> outputStream, BiConsumer<EdgeId, EdgeGrpcSession> sessionOpenListener,
-                    Consumer<EdgeId> sessionCloseListener, ScheduledExecutorService sendDownlinkExecutorService) {
+                    Consumer<EdgeId> sessionCloseListener, ScheduledExecutorService sendDownlinkExecutorService, int maxInboundMessageSize) {
         this.sessionId = UUID.randomUUID();
         this.ctx = ctx;
         this.outputStream = outputStream;
         this.sessionOpenListener = sessionOpenListener;
         this.sessionCloseListener = sessionCloseListener;
         this.sendDownlinkExecutorService = sendDownlinkExecutorService;
+        this.maxInboundMessageSize = maxInboundMessageSize;
         initInputStream();
     }
 
@@ -130,6 +134,8 @@ public final class EdgeGrpcSession implements Closeable {
                     if (ConnectResponseCode.ACCEPTED != responseMsg.getResponseCode()) {
                         outputStream.onError(new RuntimeException(responseMsg.getErrorMsg()));
                     } else {
+                        log.debug("[{}] Client max inbound message size: {}", sessionId, requestMsg.getConnectRequestMsg().getMaxInboundMessageSize());
+                        clientMaxInboundMessageSize = requestMsg.getConnectRequestMsg().getMaxInboundMessageSize();
                         connected = true;
                     }
                 }
@@ -408,9 +414,17 @@ public final class EdgeGrpcSession implements Closeable {
                     }
                     log.trace("[{}] [{}] downlink msg(s) are going to be send.", this.sessionId, copy.size());
                     for (DownlinkMsg downlinkMsg : copy) {
-                        sendDownlinkMsg(ResponseMsg.newBuilder()
-                                .setDownlinkMsg(downlinkMsg)
-                                .build());
+                        if (this.clientMaxInboundMessageSize != 0 && downlinkMsg.getSerializedSize() > this.clientMaxInboundMessageSize) {
+                            log.error("[{}] Downlink msg size [{}] exceeds client max inbound message size [{}]. Skipping this message. " +
+                                            "Please increase value of CLOUD_RPC_MAX_INBOUND_MESSAGE_SIZE env variable on the edge and restart it." +
+                                            "Message {}",
+                                    this.sessionId, downlinkMsg.getSerializedSize(), this.clientMaxInboundMessageSize, downlinkMsg);
+                            sessionState.getPendingMsgsMap().remove(downlinkMsg.getDownlinkMsgId());
+                        } else {
+                            sendDownlinkMsg(ResponseMsg.newBuilder()
+                                    .setDownlinkMsg(downlinkMsg)
+                                    .build());
+                        }
                     }
                     if (attempt < MAX_DOWNLINK_ATTEMPTS) {
                         scheduleDownlinkMsgsPackSend(attempt + 1);
@@ -638,7 +652,9 @@ public final class EdgeGrpcSession implements Closeable {
                     return ConnectResponseMsg.newBuilder()
                             .setResponseCode(ConnectResponseCode.ACCEPTED)
                             .setErrorMsg("")
-                            .setConfiguration(ctx.getEdgeMsgConstructor().constructEdgeConfiguration(edge)).build();
+                            .setConfiguration(ctx.getEdgeMsgConstructor().constructEdgeConfiguration(edge))
+                            .setMaxInboundMessageSize(maxInboundMessageSize)
+                            .build();
                 }
                 return ConnectResponseMsg.newBuilder()
                         .setResponseCode(ConnectResponseCode.BAD_CREDENTIALS)

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

@@ -20,6 +20,7 @@ import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
 import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
 import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder;
 import io.grpc.stub.StreamObserver;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
@@ -62,6 +63,10 @@ public class EdgeGrpcClient implements EdgeRpcClient {
     private boolean sslEnabled;
     @Value("${cloud.rpc.ssl.cert:}")
     private String certResource;
+    @Value("${cloud.rpc.max_inbound_message_size:4194304}")
+    private int maxInboundMessageSize;
+    @Getter
+    private int serverMaxInboundMessageSize;
 
     private ManagedChannel channel;
 
@@ -77,6 +82,7 @@ public class EdgeGrpcClient implements EdgeRpcClient {
                         Consumer<DownlinkMsg> onDownlink,
                         Consumer<Exception> onError) {
         NettyChannelBuilder builder = NettyChannelBuilder.forAddress(rpcHost, rpcPort)
+                .maxInboundMessageSize(maxInboundMessageSize)
                 .keepAliveTime(keepAliveTimeSec, TimeUnit.SECONDS);
         if (sslEnabled) {
             try {
@@ -102,6 +108,7 @@ public class EdgeGrpcClient implements EdgeRpcClient {
                         .setEdgeRoutingKey(edgeKey)
                         .setEdgeSecret(edgeSecret)
                         .setEdgeVersion(EdgeVersion.V_3_3_3)
+                        .setMaxInboundMessageSize(maxInboundMessageSize)
                         .build())
                 .build());
     }
@@ -117,6 +124,8 @@ public class EdgeGrpcClient implements EdgeRpcClient {
                 if (responseMsg.hasConnectResponseMsg()) {
                     ConnectResponseMsg connectResponseMsg = responseMsg.getConnectResponseMsg();
                     if (connectResponseMsg.getResponseCode().equals(ConnectResponseCode.ACCEPTED)) {
+                        log.debug("[{}] Server max inbound message size: {}", edgeKey, connectResponseMsg.getMaxInboundMessageSize());
+                        serverMaxInboundMessageSize = connectResponseMsg.getMaxInboundMessageSize();
                         log.info("[{}] Configuration received: {}", edgeKey, connectResponseMsg.getConfiguration());
                         onEdgeUpdate.accept(connectResponseMsg.getConfiguration());
                     } else {

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

@@ -41,4 +41,6 @@ public interface EdgeRpcClient {
     void sendUplinkMsg(UplinkMsg uplinkMsg);
 
     void sendDownlinkResponseMsg(DownlinkResponseMsg downlinkResponseMsg);
+
+    int getServerMaxInboundMessageSize();
 }

common/edge-api/src/main/proto/edge.proto

@@ -68,6 +68,7 @@ message ConnectRequestMsg {
   string edgeRoutingKey = 1;
   string edgeSecret = 2;
   EdgeVersion edgeVersion = 3;
+  int32 maxInboundMessageSize = 4;
 }
 
 enum ConnectResponseCode {
@@ -80,6 +81,7 @@ message ConnectResponseMsg {
   ConnectResponseCode responseCode = 1;
   string errorMsg = 2;
   EdgeConfiguration configuration = 3;
+  int32 maxInboundMessageSize = 4;
 }
 
 message SyncRequestMsg {