Include specific SQL statements in batch exception

Refine the SQL statements contained in exceptions thrown from batch
updates based on BatchUpdateException.getUpdateCounts().

Issue: SPR-10677

spring-jdbc/src/main/java/org/springframework/jdbc/core/JdbcTemplate.java

@@ -20,6 +20,7 @@ import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
+import java.sql.BatchUpdateException;
 import java.sql.CallableStatement;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -564,11 +565,24 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 
 				if (JdbcUtils.supportsBatchUpdates(stmt.getConnection())) {
 					for (String sqlStmt : sql) {
-						this.currSql = (StringUtils.isEmpty(this.currSql) ? sqlStmt
-								: this.currSql + "; " + sqlStmt);
+						this.currSql = appendSql(this.currSql, sqlStmt);
 						stmt.addBatch(sqlStmt);
 					}
-					rowsAffected = stmt.executeBatch();
+					try {
+						rowsAffected = stmt.executeBatch();
+					}
+					catch (BatchUpdateException ex) {
+						String batchExceptionSql = null;
+						for (int i = 0; i < ex.getUpdateCounts().length; i++) {
+							if (ex.getUpdateCounts()[i] == Statement.EXECUTE_FAILED) {
+								batchExceptionSql = appendSql(batchExceptionSql, sql[i]);
+							}
+						}
+						if (StringUtils.hasLength(batchExceptionSql)) {
+							this.currSql = batchExceptionSql;
+						}
+						throw ex;
+					}
 				}
 				else {
 					for (int i = 0; i < sql.length; i++) {
@@ -583,6 +597,11 @@ public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
 				}
 				return rowsAffected;
 			}
+
+			private String appendSql(String sql, String statement) {
+				return (StringUtils.isEmpty(sql) ? statement : sql + "; " + statement);
+			}
+
 			@Override
 			public String getSql() {
 				return this.currSql;

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

@@ -16,6 +16,7 @@
 
 package org.springframework.jdbc.core;
 
+import java.sql.BatchUpdateException;
 import java.sql.CallableStatement;
 import java.sql.Connection;
 import java.sql.DatabaseMetaData;
@@ -458,6 +459,24 @@ public class JdbcTemplateTests {
 		verify(this.connection, atLeastOnce()).close();
 	}
 
+	@Test
+	public void testBatchUpdateWithBatchFailure() throws Exception {
+		final String[] sql = {"A", "B", "C", "D"};
+		given(this.statement.executeBatch()).willThrow(
+				new BatchUpdateException(new int[] { 1, Statement.EXECUTE_FAILED, 1,
+					Statement.EXECUTE_FAILED }));
+		mockDatabaseMetaData(true);
+		given(this.connection.createStatement()).willReturn(this.statement);
+
+		JdbcTemplate template = new JdbcTemplate(this.dataSource, false);
+		try {
+			template.batchUpdate(sql);
+		}
+		catch (UncategorizedSQLException ex) {
+			assertThat(ex.getSql(), equalTo("B; D"));
+		}
+	}
+
 	@Test
 	public void testBatchUpdateWithNoBatchSupport() throws Exception {
 		final String[] sql = {"UPDATE NOSUCHTABLE SET DATE_DISPATCHED = SYSDATE WHERE ID = 1",