Merge branch 'hotfix_1.10_4.0.x_30236' into '1.10_release_4.0.x'

fix bug 30236

See merge request dt-insight-engine/flinkStreamSQL!170

Author: zoudaokoulife

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/AbstractUpsertWriter.java

@@ -122,7 +122,9 @@ public void addRecord(Tuple2<Boolean, Row> record) throws SQLException {
         // we don't need perform a deep copy, because jdbc field are immutable object.
         Tuple2<Boolean, Row> tuple2 = objectReuse ? new Tuple2<>(record.f0, Row.copy(record.f1)) : record;
         // add records to buffer
-        keyToRows.put(getPrimaryKey(tuple2.f1), tuple2);
+        if (tuple2.f0) {
+            keyToRows.put(getPrimaryKey(tuple2.f1), tuple2);
+        }
     }
 
     @Override