[FLINK-34014][jdbc] Avoid executeBatch when buffer is empty

Author: jeyhunkarimov

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/TableBufferReducedStatementExecutor.java

@@ -86,17 +86,19 @@ private boolean changeFlag(RowKind rowKind) {
 
     @Override
     public void executeBatch() throws SQLException {
-        for (Map.Entry<RowData, Tuple2<Boolean, RowData>> entry : reduceBuffer.entrySet()) {
-            if (entry.getValue().f0) {
-                upsertExecutor.addToBatch(entry.getValue().f1);
-            } else {
-                // delete by key
-                deleteExecutor.addToBatch(entry.getKey());
+        if (!reduceBuffer.isEmpty()) {
+            for (Map.Entry<RowData, Tuple2<Boolean, RowData>> entry : reduceBuffer.entrySet()) {
+                if (entry.getValue().f0) {
+                    upsertExecutor.addToBatch(entry.getValue().f1);
+                } else {
+                    // delete by key
+                    deleteExecutor.addToBatch(entry.getKey());
+                }
             }
+            upsertExecutor.executeBatch();
+            deleteExecutor.executeBatch();
+            reduceBuffer.clear();
         }
-        upsertExecutor.executeBatch();
-        deleteExecutor.executeBatch();
-        reduceBuffer.clear();
     }
 
     @Override

flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/TableBufferedStatementExecutor.java

@@ -52,11 +52,13 @@ public void addToBatch(RowData record) throws SQLException {
 
     @Override
     public void executeBatch() throws SQLException {
-        for (RowData value : buffer) {
-            statementExecutor.addToBatch(value);
+        if (!buffer.isEmpty()) {
+            for (RowData value : buffer) {
+                statementExecutor.addToBatch(value);
+            }
+            statementExecutor.executeBatch();
+            buffer.clear();
         }
-        statementExecutor.executeBatch();
-        buffer.clear();
     }
 
     @Override