From 5da80314c1777790fdc438254741e9c32b34bc13 Mon Sep 17 00:00:00 2001
From: Evaldas Buinauskas <evaldas.buinauskas@vinted.com>
Date: Tue, 24 Oct 2023 16:09:48 +0300
Subject: [PATCH] Change DLQ reporting

---
 README.md                                                 | 4 ++--
 pom.xml                                                   | 2 +-
 .../kafka/connect/vespa/feeders/VespaFeederHandler.java   | 8 +++++++-
 3 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/README.md b/README.md
index ccb2bc9..404d155 100644
--- a/README.md
+++ b/README.md
@@ -10,11 +10,11 @@ This connector has not yet been published to Confluent Hub. To install it, downl
 install it using `confluent-hub` command line tool.
 
 ```sh
-wget https://github.com/vinted/kafka-connect-vespa/releases/download/v1.0.7/vinted-kafka-connect-vespa-1.0.7-SNAPSHOT.zip -O /tmp/vinted-kafka-connect-vespa-1.0.7-SNAPSHOT.zip -q
+wget https://github.com/vinted/kafka-connect-vespa/releases/download/v1.0.8/vinted-kafka-connect-vespa-1.0.8-SNAPSHOT.zip -O /tmp/vinted-kafka-connect-vespa-1.0.8-SNAPSHOT.zip -q
 ```
 
 ```sh
-confluent-hub install --no-prompt /tmp/vinted-kafka-connect-vespa-1.0.7-SNAPSHOT.zip
+confluent-hub install --no-prompt /tmp/vinted-kafka-connect-vespa-1.0.8-SNAPSHOT.zip
 ```
 
 ### Operational modes
diff --git a/pom.xml b/pom.xml
index 6a5ca3b..67e7f40 100644
--- a/pom.xml
+++ b/pom.xml
@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.vinted.kafka.connect.vespa</groupId>
     <artifactId>kafka-connect-vespa</artifactId>
-    <version>1.0.7-SNAPSHOT</version>
+    <version>1.0.8-SNAPSHOT</version>
     <name>kafka-connect-vespa</name>
     <description>The Vespa Sink Connector is used to write data from Kafka to a Vespa search engine.</description>
     <url>https://github.com/vinted/kafka-connect-vespa</url>
diff --git a/src/main/java/com/vinted/kafka/connect/vespa/feeders/VespaFeederHandler.java b/src/main/java/com/vinted/kafka/connect/vespa/feeders/VespaFeederHandler.java
index 434bd0e..ff2de41 100644
--- a/src/main/java/com/vinted/kafka/connect/vespa/feeders/VespaFeederHandler.java
+++ b/src/main/java/com/vinted/kafka/connect/vespa/feeders/VespaFeederHandler.java
@@ -29,13 +29,15 @@ public CompletableFuture<Result> handle(SinkRecord record, CompletableFuture<Res
 
         future.whenComplete((result, throwable) -> {
             if (result == null) {
-                reporter.report(record, throwable);
+                // An exception occurred while indexing the document
 
                 if (!isMalformed(throwable)) {
                     log.error(errorMessage(record), throwable);
 
                     promise.completeExceptionally(throwable);
                 } else {
+                    reporter.report(record, throwable);
+
                     switch (config.behaviorOnMalformedDoc) {
                         case IGNORE:
                             log.info(ignoreMessage(record), throwable);
@@ -53,8 +55,12 @@ public CompletableFuture<Result> handle(SinkRecord record, CompletableFuture<Res
                     }
                 }
             } else if (result.type() == Result.Type.success) {
+                // Document was indexed successfully
                 promise.complete(result);
             } else {
+                // Document could not be indexed. Vespa responded with an error message.
+                // This usually is a condition not met error.
+
                 Throwable resultThrowable = new Throwable(result.toString());
 
                 reporter.report(record, resultThrowable);