fix handling weak ack for built-in ack labels in StreamingSessionActor

* before an error to the WS session was sent declaring that ack-label 'live-response' was not declared if a WS subscribed for live messages, but filtered them out based on an RQL filter
* now this situation is handled correctly by dropping the weak ack

Signed-off-by: Thomas Jäckle <thomas.jaeckle@beyonnex.io>

gateway/service/src/main/java/org/eclipse/ditto/gateway/service/streaming/actors/StreamingSessionActor.java

@@ -27,6 +27,7 @@
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabel;
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabelNotDeclaredException;
 import org.eclipse.ditto.base.model.acks.AcknowledgementLabelNotUniqueException;
+import org.eclipse.ditto.base.model.acks.DittoAcknowledgementLabel;
 import org.eclipse.ditto.base.model.acks.FatalPubSubException;
 import org.eclipse.ditto.base.model.auth.AuthorizationContext;
 import org.eclipse.ditto.base.model.entity.id.NamespacedEntityId;
@@ -268,6 +269,7 @@ private Receive createIncomingSignalBehavior() {
                 .build();
 
         final Receive signalBehavior = ReceiveBuilder.create()
+                .match(Acknowledgement.class, this::isWeakAckForBuiltInAckLabel, this::dropWeakAckForBuiltInAckLabelAcknowledgement)
                 .match(Acknowledgement.class, this::hasUndeclaredAckLabel, this::ackLabelNotDeclared)
                 .match(Acknowledgement.class, this::forwardAcknowledgementOrLiveCommandResponse)
                 .match(CommandResponse.class, CommandResponse::isLiveCommandResponse, liveCommandResponse ->
@@ -493,6 +495,16 @@ private static Receive addPreprocessors(final List<PartialFunction<Object, Objec
                 .orElse(receive);
     }
 
+    private boolean isWeakAckForBuiltInAckLabel(final Acknowledgement acknowledgement) {
+        return acknowledgement.isWeak() &&
+                List.of(DittoAcknowledgementLabel.values()).contains(acknowledgement.getLabel());
+    }
+
+    private void dropWeakAckForBuiltInAckLabelAcknowledgement(final Acknowledgement ack) {
+        logger.withCorrelationId(ack)
+                .info("Dropping weak ack for built-in ack label <{}>", ack.getLabel());
+    }
+
     private boolean hasUndeclaredAckLabel(final Acknowledgement acknowledgement) {
         return !declaredAcks.contains(acknowledgement.getLabel());
     }