Provide all counters in BatchUpdateException

This commit updates JbcTemplate#batchUpdate to provide additional
information when one batch fails. Previously, the raw
BatchUpdateException was thrown with no way to know what had completed
thus far.

This commit creates an AggregatedBatchUpdateException that wraps the
original BatchUpdateException, yet providing the counters of the batches
that ran prior to the exception. In essence, this represents the same
state as the return value of the method if no batch fails.

AggregateBatchUpdateException exposes the original BatchUpdateException
in advanced case, such as checking for a sub-class that may contain
additional information.

Closes gh-23867

spring-jdbc/src/test/java/org/springframework/jdbc/core/JdbcTemplateTests.java

@@ -32,13 +32,16 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Consumer;
 
 import javax.sql.DataSource;
 
+import org.assertj.core.data.Index;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.dao.DataAccessException;
+import org.springframework.dao.DuplicateKeyException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.jdbc.BadSqlGrammarException;
 import org.springframework.jdbc.CannotGetJdbcConnectionException;
@@ -799,6 +802,55 @@ class JdbcTemplateTests {
 		verify(this.connection, atLeastOnce()).close();
 	}
 
+	@Test
+	void testBatchUpdateWithBatchFailingHasUpdateCounts() throws Exception {
+		test3BatchesOf2ItemsFailing(exception -> assertThat(exception).cause()
+				.isInstanceOfSatisfying(AggregatedBatchUpdateException.class, ex -> {
+					assertThat(ex.getSuccessfulUpdateCounts()).hasDimensions(1, 2)
+							.contains(new int[] { 1, 1 }, Index.atIndex(0));
+					assertThat(ex.getUpdateCounts()).contains(-3, -3);
+				}));
+	}
+
+	@Test
+	void testBatchUpdateWithBatchFailingMatchesOriginalException() throws Exception {
+		test3BatchesOf2ItemsFailing(exception -> assertThat(exception).cause()
+				.isInstanceOfSatisfying(AggregatedBatchUpdateException.class, ex -> {
+					BatchUpdateException originalException = ex.getOriginalException();
+					assertThat(ex.getMessage()).isEqualTo(originalException.getMessage());
+					assertThat(ex.getCause()).isEqualTo(originalException.getCause());
+					assertThat(ex.getSQLState()).isEqualTo(originalException.getSQLState());
+					assertThat(ex.getErrorCode()).isEqualTo(originalException.getErrorCode());
+					assertThat((Exception) ex.getNextException()).isSameAs(originalException.getNextException());
+					assertThat(ex.getSuppressed()).isEqualTo(originalException.getSuppressed());
+				}));
+	}
+
+	void test3BatchesOf2ItemsFailing(Consumer<Exception> exception) throws Exception {
+		String sql = "INSERT INTO NOSUCHTABLE values (?)";
+		List<Integer> ids = Arrays.asList(1, 2, 3, 2, 4, 5);
+		int[] rowsAffected = new int[] {1, 1};
+
+		given(this.preparedStatement.executeBatch()).willReturn(rowsAffected).willThrow(new BatchUpdateException(
+				"duplicate key value violates unique constraint \"NOSUCHTABLE_pkey\" Detail: Key (id)=(2) already exists.",
+				"23505", 0, new int[] { -3, -3 }));
+		mockDatabaseMetaData(true);
+
+		ParameterizedPreparedStatementSetter<Integer> setter = (ps, argument) -> ps.setInt(1, argument);
+		JdbcTemplate template = new JdbcTemplate(this.dataSource, false);
+
+		assertThatExceptionOfType(DuplicateKeyException.class)
+				.isThrownBy(() -> template.batchUpdate(sql, ids, 2, setter))
+				.satisfies(exception);
+		verify(this.preparedStatement, times(4)).addBatch();
+		verify(this.preparedStatement).setInt(1, 1);
+		verify(this.preparedStatement, times(2)).setInt(1, 2);
+		verify(this.preparedStatement).setInt(1, 3);
+		verify(this.preparedStatement, times(2)).executeBatch();
+		verify(this.preparedStatement).close();
+		verify(this.connection, atLeastOnce()).close();
+	}
+
 	@Test
 	void testCouldNotGetConnectionForOperationOrExceptionTranslator() throws SQLException {
 		SQLException sqlException = new SQLException("foo", "07xxx");