Fix MQTT crashes when subscriber throws exception (#3719)

Signed-off-by: Jan N. Klug <github@klug.nrw>

bundles/org.openhab.core.io.transport.mqtt/src/main/java/org/openhab/core/io/transport/mqtt/internal/Subscription.java

@@ -18,6 +18,9 @@
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
 import org.openhab.core.io.transport.mqtt.MqttMessageSubscriber;
+import org.openhab.core.util.HexUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish;
 import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
@@ -31,6 +34,7 @@
  */
 @NonNullByDefault
 public class Subscription {
+    private final Logger logger = LoggerFactory.getLogger(Subscription.class);
     private final Map<String, byte[]> retainedMessages = new ConcurrentHashMap<>();
     private final Collection<MqttMessageSubscriber> subscribers = ConcurrentHashMap.newKeySet();
 
@@ -81,10 +85,15 @@ public void messageArrived(String topic, byte[] payload, boolean retain) {
         if (retain || retainedMessages.containsKey(topic)) {
             retainedMessages.put(topic, payload);
         }
-        subscribers.stream().forEach(subscriber -> processMessage(subscriber, topic, payload));
+        subscribers.forEach(subscriber -> processMessage(subscriber, topic, payload));
     }
 
     private void processMessage(MqttMessageSubscriber subscriber, String topic, byte[] payload) {
-        subscriber.processMessage(topic, payload);
+        try {
+            subscriber.processMessage(topic, payload);
+        } catch (RuntimeException e) {
+            logger.warn("A subscriber of type '{}' failed to process message '{}' to topic '{}'.",
+                    subscriber.getClass(), HexUtils.bytesToHex(payload), topic);
+        }
     }
 }