Feature/6233 implementation

common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java

@@ -27,6 +27,7 @@ public class MqttDeviceProfileTransportConfiguration implements DeviceProfileTra
     @NoXss
     private String deviceAttributesTopic = MqttTopics.DEVICE_ATTRIBUTES_TOPIC;
     private TransportPayloadTypeConfiguration transportPayloadTypeConfiguration;
+    private boolean sendPubAckOnValidationException;
 
     @Override
     public DeviceTransportType getType() {

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java

@@ -359,10 +359,10 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
             }
         } catch (RuntimeException e) {
             log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
-            ctx.close();
+            sendPubAckOrCloseSession(ctx, topicName, msgId);
         } catch (AdaptorException e) {
             log.debug("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
-            ctx.close();
+            sendPubAckOrCloseSession(ctx, topicName, msgId);
         }
     }
 
@@ -451,6 +451,15 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
             }
         } catch (AdaptorException e) {
             log.debug("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
+            sendPubAckOrCloseSession(ctx, topicName, msgId);
+        }
+    }
+
+    private void sendPubAckOrCloseSession(ChannelHandlerContext ctx, String topicName, int msgId) {
+        if (deviceSessionCtx.isSendPubAckOnValidationException() && msgId > 0) {
+            log.info("[{}] Send pub ack on invalid publish msg [{}][{}]", sessionId, topicName, msgId);
+            ctx.writeAndFlush(createMqttPubAckMsg(msgId));
+        } else {
             log.info("[{}] Closing current session due to invalid publish msg [{}][{}]", sessionId, topicName, msgId);
             ctx.close();
         }

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java

@@ -84,6 +84,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
     private volatile MqttTransportAdaptor adaptor;
     private volatile boolean jsonPayloadFormatCompatibilityEnabled;
     private volatile boolean useJsonPayloadFormatForDefaultDownlinkTopics;
+    private volatile boolean sendPubAckOnValidationException;
 
     @Getter
     @Setter
@@ -115,6 +116,10 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
         return payloadType.equals(TransportPayloadType.JSON);
     }
 
+    public boolean isSendPubAckOnValidationException() {
+        return sendPubAckOnValidationException;
+    }
+
     public Descriptors.Descriptor getTelemetryDynamicMsgDescriptor() {
         return telemetryDynamicMessageDescriptor;
     }
@@ -152,6 +157,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
             payloadType = transportPayloadTypeConfiguration.getTransportPayloadType();
             telemetryTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceTelemetryTopic());
             attributesTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceAttributesTopic());
+            sendPubAckOnValidationException = mqttConfig.isSendPubAckOnValidationException();
             if (TransportPayloadType.PROTOBUF.equals(payloadType)) {
                 ProtoTransportPayloadConfiguration protoTransportPayloadConfig = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
                 updateDynamicMessageDescriptors(protoTransportPayloadConfig);
@@ -162,6 +168,7 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
             telemetryTopicFilter = MqttTopicFilterFactory.getDefaultTelemetryFilter();
             attributesTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter();
             payloadType = TransportPayloadType.JSON;
+            sendPubAckOnValidationException = false;
         }
         updateAdaptor();
     }

ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.html

@@ -134,4 +134,10 @@
       </div>
     </fieldset>
   </section>
+  <div style="padding-bottom: 20px">
+    <mat-checkbox formControlName="sendPubAckOnValidationException">
+      {{ 'device-profile.mqtt-send-pub-ack-on-validation-exception' | translate }}
+    </mat-checkbox>
+    <div class="tb-hint" innerHTML="{{ 'device-profile.mqtt-send-pub-ack-on-validation-exception-hint' | translate }}"></div>
+  </div>
 </form>

ui-ngx/src/app/modules/home/components/profile/device/mqtt-device-profile-transport-configuration.component.ts

@@ -92,6 +92,7 @@ export class MqttDeviceProfileTransportConfigurationComponent implements Control
     this.mqttDeviceProfileTransportConfigurationFormGroup = this.fb.group({
         deviceAttributesTopic: [null, [Validators.required, this.validationMQTTTopic()]],
         deviceTelemetryTopic: [null, [Validators.required, this.validationMQTTTopic()]],
+        sendPubAckOnValidationException: [false, Validators.required],
         transportPayloadTypeConfiguration: this.fb.group({
           transportPayloadType: [TransportPayloadType.JSON, Validators.required],
           deviceTelemetryProtoSchema: [defaultTelemetrySchema, Validators.required],

ui-ngx/src/app/shared/models/device.models.ts

@@ -242,6 +242,7 @@ export interface DefaultDeviceProfileTransportConfiguration {
 export interface MqttDeviceProfileTransportConfiguration {
   deviceTelemetryTopic?: string;
   deviceAttributesTopic?: string;
+  sendPubAckOnValidationException?: boolean;
   transportPayloadTypeConfiguration?: {
     transportPayloadType?: TransportPayloadType;
     enableCompatibilityWithJsonPayloadFormat?: boolean;
@@ -358,6 +359,7 @@ export function createDeviceProfileTransportConfiguration(type: DeviceTransportT
         const mqttTransportConfiguration: MqttDeviceProfileTransportConfiguration = {
           deviceTelemetryTopic: 'v1/devices/me/telemetry',
           deviceAttributesTopic: 'v1/devices/me/attributes',
+          sendPubAckOnValidationException: false,
           transportPayloadTypeConfiguration: {
             transportPayloadType: TransportPayloadType.JSON,
             enableCompatibilityWithJsonPayloadFormat: false,

ui-ngx/src/assets/locale/locale.constant-en_US.json

@@ -1135,6 +1135,8 @@
         "mqtt-enable-compatibility-with-json-payload-format-hint": "When enabled, the platform will use a Protobuf payload format by default. If parsing fails, the platform will attempt to use JSON payload format. Useful for backward compatibility during firmware updates. For example, the initial release of the firmware uses Json, while the new release uses Protobuf. During the process of firmware update for the fleet of devices, it is required to support both Protobuf and JSON simultaneously. The compatibility mode introduces slight performance degradation, so it is recommended to disable this mode once all devices are updated.",
         "mqtt-use-json-format-for-default-downlink-topics": "Use Json format for default downlink topics",
         "mqtt-use-json-format-for-default-downlink-topics-hint": "When enabled, the platform will use Json payload format to push attributes and RPC via the following topics: <b>v1/devices/me/attributes/response/$request_id</b>, <b>v1/devices/me/attributes</b>, <b>v1/devices/me/rpc/request/$request_id</b>, <b>v1/devices/me/rpc/response/$request_id</b>. This setting does not impact attribute and rpc subscriptions sent using new (v2) topics: <b>v2/a/res/$request_id</b>, <b>v2/a</b>, <b>v2/r/req/$request_id</b>, <b>v2/r/res/$request_id</b>. Where <b>$request_id</b> is an integer request identifier.",
+        "mqtt-send-pub-ack-on-validation-exception": "Send PUBACK on PUBLISH message validation failure",
+        "mqtt-send-pub-ack-on-validation-exception-hint": "When enabled, the MQTT transport service will send publish acknowledgment on publish message validation failure, otherwise, the MQTT transport service will close the MQTT session.",
         "snmp-add-mapping": "Add SNMP mapping",
         "snmp-mapping-not-configured": "No mapping for OID to timeseries/telemetry configured",
         "snmp-timseries-or-attribute-name": "Timeseries/attribute name for mapping",