[CONJ-1071] Error during Bulk execution might result in connection wr…

…ong state

src/main/java/org/mariadb/jdbc/client/impl/StandardClient.java

@@ -543,7 +543,7 @@ public List<Completion> executePipeline(
       boolean canRedo)
       throws SQLException {
     List<Completion> results = new ArrayList<>();
-
+    int perMsgCounter = 0;
     int readCounter = 0;
     int[] responseMsg = new int[messages.length];
     try {
@@ -566,7 +566,7 @@ public List<Completion> executePipeline(
         }
         while (readCounter < messages.length) {
           readCounter++;
-          for (int j = 0; j < responseMsg[readCounter - 1]; j++) {
+          for (perMsgCounter = 0; perMsgCounter < responseMsg[readCounter - 1]; perMsgCounter++) {
             results.addAll(
                 readResponse(
                     stmt,
@@ -583,6 +583,23 @@ public List<Completion> executePipeline(
     } catch (SQLException sqlException) {
       if (!closed) {
         // read remaining results
+        perMsgCounter++;
+        for (; perMsgCounter < responseMsg[readCounter - 1]; perMsgCounter++) {
+          try {
+            results.addAll(
+                readResponse(
+                    stmt,
+                    messages[readCounter - 1],
+                    fetchSize,
+                    maxRows,
+                    resultSetConcurrency,
+                    resultSetType,
+                    closeOnCompletion));
+          } catch (SQLException e) {
+            // eat
+          }
+        }
+
         for (int i = readCounter; i < messages.length; i++) {
           for (int j = 0; j < responseMsg[i]; j++) {
             try {

src/test/java/org/mariadb/jdbc/integration/BulkStmtSplitError.java

@@ -0,0 +1,49 @@
+package org.mariadb.jdbc.integration;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Types;
+import org.junit.jupiter.api.*;
+import org.mariadb.jdbc.Statement;
+
+public class BulkStmtSplitError extends Common {
+
+  @AfterAll
+  public static void drop() throws SQLException {
+    Statement stmt = sharedConn.createStatement();
+    stmt.execute("DROP TABLE IF EXISTS BulkStmtSplitError");
+  }
+
+  @BeforeAll
+  public static void beforeAll2() throws SQLException {
+    drop();
+    Statement stmt = sharedConn.createStatement();
+    stmt.execute(
+        "CREATE TABLE BulkStmtSplitError (t1 int not null primary key, field varchar(300))");
+    stmt.execute("FLUSH TABLES");
+  }
+
+  @Test
+  public void BulkStmtSplitError() throws SQLException {
+    Statement stmt = sharedConn.createStatement();
+    stmt.execute("INSERT INTO BulkStmtSplitError VALUES (1, 'tt'), (2, 'tt2')");
+
+    try (PreparedStatement prep =
+        sharedConn.prepareStatement("insert into BulkStmtSplitError values (?, ?)")) {
+      prep.setInt(1, 1);
+      prep.setNull(2, Types.VARCHAR);
+      prep.addBatch();
+      prep.setInt(1, 2);
+      prep.setString(2, "Kenny");
+      prep.addBatch();
+      prep.executeBatch();
+    } catch (SQLException e) {
+      // eat
+    }
+
+    ResultSet rs = stmt.executeQuery("SELECT count(*) FROM agent");
+    Assertions.assertTrue(rs.next());
+    Assertions.assertEquals(2, rs.getInt(1));
+  }
+}