[improve][io] Add ElasticSearch Sink bulk debug logs (apache#179)

michaeljmarshall · May 4, 2023 · e2e2c4a · e2e2c4a
1 parent 4201d00
commit e2e2c4a
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 0 deletions.
diff --git a/...src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticBulkProcessor.java b/...src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticBulkProcessor.java
@@ -100,6 +100,10 @@ public void appendIndexRequest(BulkIndexRequest request) throws IOException {
         if (config.getBulkSizeInMb() > 0) {
             sourceLength = request.getDocumentSource().getBytes(StandardCharsets.UTF_8).length;
         }
+        if (log.isDebugEnabled()) {
+            log.debug("append index request id={}, type={}, source={}, length={}",
+                    request.getDocumentId(), config.getTypeName(), mapped, sourceLength);
+        }
         add(BulkOperationWithPulsarRecord.indexOperation(indexOperation, request.getRecord(), sourceLength));
     }
 
@@ -109,6 +113,9 @@ public void appendDeleteRequest(BulkDeleteRequest request) {
                 .index(request.getIndex())
                 .id(request.getDocumentId())
                 .build();
+        if (log.isDebugEnabled()) {
+            log.debug("append delete request id={}, type={}", request.getDocumentId(), config.getTypeName());
+        }
         add(BulkOperationWithPulsarRecord.deleteOperation(deleteOperation, request.getRecord()));
     }
 

diff --git a/...a/org/apache/pulsar/io/elasticsearch/client/opensearch/OpenSearchHighLevelRestClient.java b/...a/org/apache/pulsar/io/elasticsearch/client/opensearch/OpenSearchHighLevelRestClient.java
@@ -303,6 +303,10 @@ public void appendIndexRequest(BulkProcessor.BulkIndexRequest request) throws IO
         }
         indexRequest.type(config.getTypeName());
         indexRequest.source(request.getDocumentSource(), XContentType.JSON);
+        if (log.isDebugEnabled()) {
+            log.debug("append index request id={}, type={}, source={}", request.getDocumentId(), config.getTypeName(),
+                    request.getDocumentSource());
+        }
         internalBulkProcessor.add(indexRequest);
     }
 
@@ -311,6 +315,9 @@ public void appendDeleteRequest(BulkProcessor.BulkDeleteRequest request) throws
         DeleteRequest deleteRequest = new DeleteRequestWithPulsarRecord(request.getIndex(), request.getRecord());
         deleteRequest.id(request.getDocumentId());
         deleteRequest.type(config.getTypeName());
+        if (log.isDebugEnabled()) {
+            log.debug("append delete request id={}, type={}", request.getDocumentId(), config.getTypeName());
+        }
         internalBulkProcessor.add(deleteRequest);
     }