fix (log): print records in jdbc flush error message (#57)

oceanbase · Mar 11, 2024 · b1eaac1 · b1eaac1
1 parent 90c3284
commit b1eaac1
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 5 deletions.
diff --git a/...or-oceanbase-base/src/main/java/com/oceanbase/connector/flink/table/DataChangeRecord.java b/...or-oceanbase-base/src/main/java/com/oceanbase/connector/flink/table/DataChangeRecord.java
@@ -75,4 +75,9 @@ public boolean isUpsert() {
     public Object getFieldValue(String fieldName) {
         return data.getValue(table.getFieldIndex(fieldName));
     }
+
+    @Override
+    public String toString() {
+        return "DataChangeRecord{" + "table=" + table + ", type=" + type + ", data=" + data + '}';
+    }
 }
diff --git a/...ceanbase-base/src/main/java/com/oceanbase/connector/flink/table/DataChangeRecordData.java b/...ceanbase-base/src/main/java/com/oceanbase/connector/flink/table/DataChangeRecordData.java
@@ -47,4 +47,9 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Arrays.deepHashCode(this.values);
     }
+
+    @Override
+    public String toString() {
+        return Arrays.toString(values);
+    }
 }
diff --git a/...or-oceanbase/src/main/java/com/oceanbase/connector/flink/sink/OceanBaseRecordFlusher.java b/...or-oceanbase/src/main/java/com/oceanbase/connector/flink/sink/OceanBaseRecordFlusher.java
@@ -263,13 +263,23 @@ private void flush(String sql, List<String> statementFields, List<DataChangeReco
         try (Connection connection = connectionProvider.getConnection();
                 PreparedStatement statement = connection.prepareStatement(sql)) {
             for (List<DataChangeRecord> groupRecords : group.values()) {
-                for (DataChangeRecord record : groupRecords) {
-                    for (int i = 0; i < statementFields.size(); i++) {
-                        statement.setObject(i + 1, record.getFieldValue(statementFields.get(i)));
+                try {
+                    for (DataChangeRecord record : groupRecords) {
+                        for (int i = 0; i < statementFields.size(); i++) {
+                            statement.setObject(
+                                    i + 1, record.getFieldValue(statementFields.get(i)));
+                        }
+                        statement.addBatch();
                     }
-                    statement.addBatch();
+                    statement.executeBatch();
+                } catch (SQLException e) {
+                    throw new RuntimeException(
+                            "Failed to execute batch with sql: "
+                                    + sql
+                                    + ", records: "
+                                    + groupRecords,
+                            e);
                 }
-                statement.executeBatch();
             }
         }
     }