Repository: risingwavelabs/risingwave (language: rust)

Issue:
cherry-pick fix(sink): fix mogodb write error handling (#19869) to branch release-2.0
Cherry picking \#19869 onto branch release-2.0,

This PR/issue was created by cherry-pick action from commit 1b9b5a797396ce19e08b55c0f111e6e30310fb89.,


Discussion / hints:

Fix patch:
diff --git a/src/connector/src/sink/mongodb.rs b/src/connector/src/sink/mongodb.rs
index d09d44b0de9dc..1d741ee94cd35 100644
--- a/src/connector/src/sink/mongodb.rs
+++ b/src/connector/src/sink/mongodb.rs
@@ -728,6 +728,12 @@ impl MongodbPayloadWriter {
             )))
         })?;
 
+        if let Ok(ok) = result.get_i32("ok")
+            && ok != 1
+        {
+            return Err(SinkError::Mongodb(anyhow!("bulk write write errors")));
+        }
+
         if let Ok(write_errors) = result.get_array("writeErrors") {
             return Err(SinkError::Mongodb(anyhow!(
                 "bulk write respond with write errors: {:?}",
@@ -735,15 +741,10 @@ impl MongodbPayloadWriter {
             )));
         }
 
-        let n = result.get_i32("n").map_err(|err| {
-            SinkError::Mongodb(
-                anyhow!(err).context("can't extract field n from bulk write response"),
-            )
-        })?;
-        if n < 1 {
+        if let Ok(write_concern_error) = result.get_array("writeConcernError") {
             return Err(SinkError::Mongodb(anyhow!(
-                "bulk write respond with an abnormal state, n = {}",
-                n
+                "bulk write respond with write errors: {:?}",
+                write_concern_error,
             )));
         }
 


Test patch: