(hop) branch main updated: Bulk Doris Loader Plugin - Add server error handling and better output and rejected rows count (#6834)

[hansva]

This is an automated email from the ASF dual-hosted git repository.

hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git


The following commit(s) were added to refs/heads/main by this push:
     new 0aa820a4b0 Bulk Doris Loader Plugin - Add server error handling and 
better output and rejected rows count (#6834)
0aa820a4b0 is described below

commit 0aa820a4b0057d43f7560af9092fdaf25e40d5b9
Author: Davide Villa <davi...@gepinformatica.it>
AuthorDate: Thu Mar 19 14:20:40 2026 +0100

    Bulk Doris Loader Plugin - Add server error handling and better output and 
rejected rows count (#6834)
    
    * Update Output and Rejected Lines
    
    * Managed Doris Server Errors
    
    ---------
    
    Co-authored-by: davidev <davidev@tuxedo-davidev>
---
 .../dorisbulkloader/DorisBulkLoader.java           | 28 ++++++++++++++++++++--
 .../dorisbulkloader/ResponseContent.java           | 16 +++++++++++++
 2 files changed, 42 insertions(+), 2 deletions(-)

diff --git 
a/plugins/transforms/dorisbulkloader/src/main/java/org/apache/hop/pipeline/transforms/dorisbulkloader/DorisBulkLoader.java
 
b/plugins/transforms/dorisbulkloader/src/main/java/org/apache/hop/pipeline/transforms/dorisbulkloader/DorisBulkLoader.java
index 55fef2379f..c20cb9b51e 100644
--- 
a/plugins/transforms/dorisbulkloader/src/main/java/org/apache/hop/pipeline/transforms/dorisbulkloader/DorisBulkLoader.java
+++ 
b/plugins/transforms/dorisbulkloader/src/main/java/org/apache/hop/pipeline/transforms/dorisbulkloader/DorisBulkLoader.java
@@ -75,11 +75,11 @@ public class DorisBulkLoader extends 
BaseTransform<DorisBulkLoaderMeta, DorisBul
       processStreamLoad(rowString, first);
 
       if (checkFeedback(getLinesRead()) && isDetailed()) {
-        logDetailed(BaseMessages.getString(PKG, 
"DorisBulkLoader.Log.LineNumber") + getLinesRead());
+        logDetailed(
+            BaseMessages.getString(PKG, "DorisBulkLoader.Log.LineNumber") + " 
: " + getLinesRead());
       }
 
       putRow(getInputRowMeta(), r);
-      incrementLinesOutput();
 
       if (first) {
         first = false;
@@ -138,6 +138,8 @@ public class DorisBulkLoader extends 
BaseTransform<DorisBulkLoaderMeta, DorisBul
                 BaseMessages.getString(
                     PKG, "DorisBulkLoader.Log.StreamLoadResult", 
responseContent.toString()));
           }
+          // Set Real Rows Loaded and Rejected
+          updateLoadedRows(responseContent);
           // close doris http client
           data.dorisStreamLoad.close();
         }
@@ -165,6 +167,8 @@ public class DorisBulkLoader extends 
BaseTransform<DorisBulkLoaderMeta, DorisBul
               BaseMessages.getString(
                   PKG, "DorisBulkLoader.Log.StreamLoadResult", 
responseContent.toString()));
         }
+        // Set Real Rows Loaded and Rejected
+        updateLoadedRows(responseContent);
         data.dorisStreamLoad.startWritingIntoBuffer();
         if (data.dorisStreamLoad.canWrite(record.length)) {
           data.dorisStreamLoad.writeRecord(record);
@@ -182,6 +186,26 @@ public class DorisBulkLoader extends 
BaseTransform<DorisBulkLoaderMeta, DorisBul
     }
   }
 
+  /**
+   * Update Stats on Output/Rejected Rows
+   *
+   * @param responseContent
+   */
+  private void updateLoadedRows(ResponseContent responseContent) throws 
DorisStreamLoadException {
+
+    if (!responseContent.getStatus().equalsIgnoreCase("Success")) {
+      setErrors(getLinesWritten() - getLinesOutput());
+      throw new DorisStreamLoadException(
+          responseContent.getStatus() + " - " + responseContent.getMessage());
+    }
+    // Set Real Rows Loaded and Rejected
+    setLinesOutput(getLinesOutput() + responseContent.getNumberLoadedRows());
+    setLinesRejected(
+        getLinesRejected()
+            + responseContent.getNumberTotalRows()
+            - responseContent.getNumberLoadedRows());
+  }
+
   /**
    * init stream load
    *
diff --git 
a/plugins/transforms/dorisbulkloader/src/main/java/org/apache/hop/pipeline/transforms/dorisbulkloader/ResponseContent.java
 
b/plugins/transforms/dorisbulkloader/src/main/java/org/apache/hop/pipeline/transforms/dorisbulkloader/ResponseContent.java
index 85128d4380..5d36398f5a 100644
--- 
a/plugins/transforms/dorisbulkloader/src/main/java/org/apache/hop/pipeline/transforms/dorisbulkloader/ResponseContent.java
+++ 
b/plugins/transforms/dorisbulkloader/src/main/java/org/apache/hop/pipeline/transforms/dorisbulkloader/ResponseContent.java
@@ -105,6 +105,22 @@ public class ResponseContent {
     return ErrorURL;
   }
 
+  public long getNumberTotalRows() {
+    return NumberTotalRows;
+  }
+
+  public long getNumberLoadedRows() {
+    return NumberLoadedRows;
+  }
+
+  public int getNumberFilteredRows() {
+    return NumberFilteredRows;
+  }
+
+  public int getNumberUnselectedRows() {
+    return NumberUnselectedRows;
+  }
+
   @Override
   public String toString() {
     ObjectMapper mapper = HopJson.newMapper();