Change DLQ reporting (#16)

vinted · Oct 24, 2023 · 8d35955 · 8d35955
1 parent a7ee4fd
commit 8d35955
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 4 deletions.
diff --git 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
@@ -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
@@ -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);