This is an automated email from the ASF dual-hosted git repository.
zhangshenghang pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 5b0f8a5f69 [Fix][Connector-V2] Fix JDBC sink reconnect for nested
batch exceptions (#10793)
5b0f8a5f69 is described below
commit 5b0f8a5f69388788e91ca9af8833588f6cb25972
Author: David Zollo <[email protected]>
AuthorDate: Tue Jul 21 23:10:50 2026 +0800
[Fix][Connector-V2] Fix JDBC sink reconnect for nested batch exceptions
(#10793)
---
.../seatunnel/jdbc/internal/JdbcOutputFormat.java | 71 ++++++++-
.../internal/JdbcOutputFormatReconnectTest.java | 175 +++++++++++++++++++++
2 files changed, 245 insertions(+), 1 deletion(-)
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcOutputFormat.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcOutputFormat.java
index d13462f07b..b709302836 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcOutputFormat.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcOutputFormat.java
@@ -30,6 +30,9 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.Serializable;
import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
import java.util.function.Supplier;
import static
org.apache.seatunnel.shade.com.google.common.base.Preconditions.checkNotNull;
@@ -137,7 +140,7 @@ public class JdbcOutputFormat<I, E extends
JdbcBatchStatementExecutor<I>> implem
CommonErrorCodeDeprecated.FLUSH_DATA_FAILED, e);
}
try {
- if (!connectionProvider.isConnectionValid()) {
+ if (shouldRefreshExecutor(findSqlExceptions(e))) {
updateExecutor(true);
}
} catch (Exception exception) {
@@ -228,6 +231,72 @@ public class JdbcOutputFormat<I, E extends
JdbcBatchStatementExecutor<I>> implem
: connectionProvider.getConnection());
}
+ private boolean shouldRefreshExecutor(List<SQLException> sqlExceptions)
throws SQLException {
+ // BatchUpdateException often stores the vendor SQLException in
nextException.
+ return hasConnectionErrorSqlState(sqlExceptions)
+ || isStatementClosed(sqlExceptions)
+ || !connectionProvider.isConnectionValid();
+ }
+
+ private boolean hasConnectionErrorSqlState(List<SQLException>
sqlExceptions) {
+ for (SQLException sqlException : sqlExceptions) {
+ String sqlState = sqlException.getSQLState();
+ if (sqlState != null && sqlState.startsWith("08")) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private List<SQLException> findSqlExceptions(Throwable throwable) {
+ List<SQLException> sqlExceptions = new ArrayList<>();
+ while (throwable != null) {
+ if (throwable instanceof SQLException) {
+ collectSqlExceptionChain((SQLException) throwable,
sqlExceptions);
+ }
+ throwable = throwable.getCause();
+ }
+ return sqlExceptions;
+ }
+
+ private void collectSqlExceptionChain(
+ SQLException sqlException, List<SQLException> sqlExceptions) {
+ SQLException current = sqlException;
+ while (current != null) {
+ sqlExceptions.add(current);
+ current = current.getNextException();
+ }
+ }
+
+ private boolean isStatementClosed(List<SQLException> sqlExceptions) {
+ for (SQLException sqlException : sqlExceptions) {
+ if (isStatementClosed(sqlException)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private boolean isStatementClosed(SQLException sqlException) {
+ String exceptionClassName =
+
sqlException.getClass().getSimpleName().toLowerCase(Locale.ROOT);
+ if (exceptionClassName.contains("statementisclosedexception")) {
+ return true;
+ }
+
+ String message = sqlException.getMessage();
+ if (message == null) {
+ return false;
+ }
+
+ String normalizedMessage = message.toLowerCase(Locale.ROOT);
+ // SQL Server may report a closed statement handle without using
"closed".
+ return normalizedMessage.contains("statement closed")
+ || normalizedMessage.contains("statement is closed")
+ || normalizedMessage.contains("statement handle is not
executing")
+ || normalizedMessage.contains("statement handle is closed");
+ }
+
/**
* A factory for creating {@link JdbcBatchStatementExecutor} instance.
*
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcOutputFormatReconnectTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcOutputFormatReconnectTest.java
new file mode 100644
index 0000000000..c197d499d3
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcOutputFormatReconnectTest.java
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.jdbc.internal;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcConnectionConfig;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.connection.JdbcConnectionProvider;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.executor.JdbcBatchStatementExecutor;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.sql.BatchUpdateException;
+import java.sql.Connection;
+import java.sql.SQLException;
+
+/** Tests JDBC output retry decisions for nested SQLExceptions thrown by batch
execution. */
+public class JdbcOutputFormatReconnectTest {
+
+ @Test
+ public void
testFlushRetryShouldReconnectWhenBatchNextExceptionHasConnectionSqlState()
+ throws Exception {
+ JdbcConnectionProvider provider =
Mockito.mock(JdbcConnectionProvider.class);
+ Connection connection = Mockito.mock(Connection.class);
+
Mockito.when(provider.getOrEstablishConnection()).thenReturn(connection);
+ Mockito.when(provider.getConnection()).thenReturn(connection);
+ Mockito.when(provider.reestablishConnection()).thenReturn(connection);
+
+ TrackingJdbcBatchExecutor executor =
+ new TrackingJdbcBatchExecutor(
+ batchException(
+ "batch failed",
+ "HY000",
+ new SQLException("connection dropped",
"08006")));
+
+ JdbcOutputFormat<SeaTunnelRow, TrackingJdbcBatchExecutor> outputFormat
=
+ new JdbcOutputFormat<>(provider, buildConnectionConfig(), ()
-> executor);
+ outputFormat.open();
+ outputFormat.writeRecord(new SeaTunnelRow(new Object[] {"AA"}));
+
+ outputFormat.flush();
+
+ Assertions.assertEquals(2, executor.prepareStatementsCalls);
+ Assertions.assertEquals(2, executor.executeBatchCalls);
+ Assertions.assertEquals(1, executor.closeStatementsCalls);
+ Mockito.verify(provider).reestablishConnection();
+ Mockito.verify(provider, Mockito.never()).isConnectionValid();
+ }
+
+ @Test
+ public void
testFlushRetryShouldReconnectWhenBatchNextExceptionIsStatementClosed()
+ throws Exception {
+ JdbcConnectionProvider provider =
Mockito.mock(JdbcConnectionProvider.class);
+ Connection connection = Mockito.mock(Connection.class);
+
Mockito.when(provider.getOrEstablishConnection()).thenReturn(connection);
+ Mockito.when(provider.getConnection()).thenReturn(connection);
+ Mockito.when(provider.reestablishConnection()).thenReturn(connection);
+
+ TrackingJdbcBatchExecutor executor =
+ new TrackingJdbcBatchExecutor(
+ batchException(
+ "batch failed",
+ "HY000",
+ new SQLException("No operations allowed after
statement closed.")));
+
+ JdbcOutputFormat<SeaTunnelRow, TrackingJdbcBatchExecutor> outputFormat
=
+ new JdbcOutputFormat<>(provider, buildConnectionConfig(), ()
-> executor);
+ outputFormat.open();
+ outputFormat.writeRecord(new SeaTunnelRow(new Object[] {"AA"}));
+
+ outputFormat.flush();
+
+ Assertions.assertEquals(2, executor.prepareStatementsCalls);
+ Assertions.assertEquals(2, executor.executeBatchCalls);
+ Assertions.assertEquals(1, executor.closeStatementsCalls);
+ Mockito.verify(provider).reestablishConnection();
+ Mockito.verify(provider, Mockito.never()).isConnectionValid();
+ }
+
+ @Test
+ public void
testFlushRetryShouldReconnectWhenSqlServerStatementHandleIsNotExecuting()
+ throws Exception {
+ JdbcConnectionProvider provider =
Mockito.mock(JdbcConnectionProvider.class);
+ Connection connection = Mockito.mock(Connection.class);
+
Mockito.when(provider.getOrEstablishConnection()).thenReturn(connection);
+ Mockito.when(provider.getConnection()).thenReturn(connection);
+ Mockito.when(provider.reestablishConnection()).thenReturn(connection);
+
+ TrackingJdbcBatchExecutor executor =
+ new TrackingJdbcBatchExecutor(
+ batchException(
+ "batch failed",
+ "HY000",
+ new SQLException("Statement handle is not
executing.")));
+
+ JdbcOutputFormat<SeaTunnelRow, TrackingJdbcBatchExecutor> outputFormat
=
+ new JdbcOutputFormat<>(provider, buildConnectionConfig(), ()
-> executor);
+ outputFormat.open();
+ outputFormat.writeRecord(new SeaTunnelRow(new Object[] {"AA"}));
+
+ outputFormat.flush();
+
+ Assertions.assertEquals(2, executor.prepareStatementsCalls);
+ Assertions.assertEquals(2, executor.executeBatchCalls);
+ Assertions.assertEquals(1, executor.closeStatementsCalls);
+ Mockito.verify(provider).reestablishConnection();
+ Mockito.verify(provider, Mockito.never()).isConnectionValid();
+ }
+
+ private JdbcConnectionConfig buildConnectionConfig() {
+ return JdbcConnectionConfig.builder()
+ .url("jdbc:postgresql://localhost:5432/test")
+ .maxRetries(1)
+ .batchSize(1024)
+ .build();
+ }
+
+ private static BatchUpdateException batchException(
+ String message, String sqlState, SQLException nextException) {
+ BatchUpdateException exception = new BatchUpdateException(message,
sqlState, new int[0]);
+ exception.setNextException(nextException);
+ return exception;
+ }
+
+ private static class TrackingJdbcBatchExecutor
+ implements JdbcBatchStatementExecutor<SeaTunnelRow> {
+ private final SQLException firstFailure;
+ private boolean failedOnce;
+ private int prepareStatementsCalls;
+ private int executeBatchCalls;
+ private int closeStatementsCalls;
+
+ private TrackingJdbcBatchExecutor(SQLException firstFailure) {
+ this.firstFailure = firstFailure;
+ }
+
+ @Override
+ public void prepareStatements(Connection connection) {
+ prepareStatementsCalls++;
+ }
+
+ @Override
+ public void addToBatch(SeaTunnelRow record) {}
+
+ @Override
+ public void executeBatch() throws SQLException {
+ executeBatchCalls++;
+ if (!failedOnce) {
+ failedOnce = true;
+ throw firstFailure;
+ }
+ }
+
+ @Override
+ public void closeStatements() {
+ closeStatementsCalls++;
+ }
+ }
+}
