[camel-kafka-connector] 01/01: Throw ConnectException due to failed exchanges

[oa...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

oalsafi pushed a commit to branch add-exception
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git

commit dc0dc2596cdb31787e2a9ee8c0b03883b0d915b3
Author: Omar Al-Safi <om...@gmail.com>
AuthorDate: Fri Jun 12 14:25:08 2020 +0200

    Throw ConnectException due to failed exchanges
---
 core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
index d8fef18..422d3ed 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
@@ -127,6 +127,10 @@ public class CamelSinkTask extends SinkTask {
 
             LOG.debug("Sending exchange {} to {}", exchange.getExchangeId(), LOCAL_URL);
             producer.send(LOCAL_URL, exchange);
+
+            if (exchange.isFailed()) {
+                throw new ConnectException("Exchange delivery has failed!", exchange.getException());
+            }
         }
     }