[improve][es-sink] Add error log for failed bulk records in ES sink (a…

…pache#16177) (cherry picked from commit c327b63)
datastax · Jun 9, 2023 · c8d7cc6 · c8d7cc6
1 parent 42f3668
commit c8d7cc6
Showing 1 changed file with 11 additions and 2 deletions.
diff --git a/.../elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java b/.../elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java
@@ -76,7 +76,7 @@ public void afterBulk(long executionId, List<BulkProcessor.BulkOperationRequest>
                     final Record record = bulkOperationList.get(index++).getPulsarRecord();
                     if (result.isError()) {
                         record.fail();
-                        checkForIrrecoverableError(result);
+                        checkForIrrecoverableError(record, result);
                     } else {
                         record.ack();
                     }
@@ -106,13 +106,15 @@ boolean isFailed() {
         return irrecoverableError.get() != null;
     }
 
-    void checkForIrrecoverableError(BulkProcessor.BulkOperationResult result) {
+    void checkForIrrecoverableError(Record<?> record, BulkProcessor.BulkOperationResult result) {
         if (!result.isError()) {
             return;
         }
         final String errorCause = result.getError();
+        boolean isMalformed = false;
         for (String error : MALFORMED_ERRORS) {
             if (errorCause.contains(error)) {
+                isMalformed = true;
                 switch (config.getMalformedDocAction()) {
                     case IGNORE:
                         break;
@@ -132,6 +134,13 @@ void checkForIrrecoverableError(BulkProcessor.BulkOperationResult result) {
                 }
             }
         }
+        if (!isMalformed) {
+            log.warn("Bulk request failed, message id=[{}] index={} error={}",
+                    record.getMessage()
+                            .map(m -> m.getMessageId().toString())
+                            .orElse(""),
+                    result.getIndex(), result.getError());
+        }
     }
 
     public void bulkIndex(Record record, Pair<String, String> idAndDoc) throws Exception {