#586 send connection failure message to parent if kafka producer stre…

…am fails Signed-off-by: Johannes Schneider <johannes.schneider@bosch.io>
eclipse-ditto · Jun 15, 2021 · 20b3b40 · 20b3b40
1 parent 2640ee2
commit 20b3b40
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/...main/java/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaPublisherActor.java b/...main/java/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaPublisherActor.java
@@ -45,6 +45,7 @@
 import org.eclipse.ditto.connectivity.service.messaging.BasePublisherActor;
 import org.eclipse.ditto.connectivity.service.messaging.ExceptionToAcknowledgementConverter;
 import org.eclipse.ditto.connectivity.service.messaging.SendResult;
+import org.eclipse.ditto.connectivity.service.messaging.internal.ImmutableConnectionFailure;
 import org.eclipse.ditto.json.JsonField;
 import org.eclipse.ditto.json.JsonObject;
 import org.eclipse.ditto.json.JsonObjectBuilder;
@@ -335,7 +336,9 @@ private KafkaProducerStream(final KafkaConfig config, final Materializer materia
                 if (t == null) {
                     logger.debug("Publisher source queue done.");
                 } else {
-                    logger.debug("Error occurred in publisher source queue.", t);
+                    final String msg = "Error occurred in publisher source queue.";
+                    logger.debug(msg, t);
+                    getContext().parent().tell(new ImmutableConnectionFailure(getSelf(), t, msg), getSelf());
                 }
             });