OpenSearch Sink: Add log messages when there is no exception (#3532)

Add log messages when there is exception Signed-off-by: Kondaka <krishkdk@amazon.com>
opensearch-project · Oct 31, 2023 · a30fdb5 · a30fdb5
1 parent c560e1f
commit a30fdb5
Showing 1 changed file with 14 additions and 1 deletion.
diff --git a/...h/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategy.java b/...h/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategy.java
@@ -238,6 +238,13 @@ private BulkOperationRequestResponse handleRetriesAndFailures(final Accumulating
         if (doRetry) {
             if (retryCount % 5 == 0) {
                 LOG.warn("Bulk Operation Failed. Number of retries {}. Retrying... ", retryCount, e);
+                if (e == null) {
+                    for (final BulkResponseItem bulkItemResponse : bulkResponse.items()) {
+                        if (Objects.nonNull(bulkItemResponse.error())) {
+                            LOG.warn("operation = {}, error = {}", bulkItemResponse.operationType(), bulkItemResponse.error());
+                        }
+                    }
+                }
             }
             bulkRequestNumberOfRetries.increment();
             return new BulkOperationRequestResponse(bulkRequestForRetry, bulkResponse);
@@ -248,7 +255,13 @@ private BulkOperationRequestResponse handleRetriesAndFailures(final Accumulating
     }
 
     private void handleFailures(final AccumulatingBulkRequest<BulkOperationWrapper, BulkRequest> bulkRequest, final BulkResponse bulkResponse, final Throwable failure) {
-        if (Objects.isNull(failure)) {
+        LOG.warn("Bulk Operation Failed.", failure);
+        if (failure == null) {
+            for (final BulkResponseItem bulkItemResponse : bulkResponse.items()) {
+                if (Objects.nonNull(bulkItemResponse.error())) {
+                    LOG.warn("operation = {}, error = {}", bulkItemResponse.operationType(), bulkItemResponse.error());
+                }
+            }
             handleFailures(bulkRequest, bulkResponse.items());
         } else {
             handleFailures(bulkRequest, failure);