You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2020/06/10 03:11:59 UTC
[flink] branch release-1.11 updated: [FLINK-16681][jdbc] Fix JDBC
source/sink lost connection after a long time idle
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.11 by this push:
new da677ec [FLINK-16681][jdbc] Fix JDBC source/sink lost connection after a long time idle
da677ec is described below
commit da677ec4338fdcbafa84420c3899e883489fca6c
Author: Lijie Wang <wa...@126.com>
AuthorDate: Mon Jun 1 18:57:57 2020 +0800
[FLINK-16681][jdbc] Fix JDBC source/sink lost connection after a long time idle
This closes #12427
---
.../jdbc/internal/AbstractJdbcOutputFormat.java | 8 +-
.../jdbc/internal/JdbcBatchingOutputFormat.java | 15 ++-
.../jdbc/internal/TableJdbcUpsertOutputFormat.java | 4 +-
.../connection/JdbcConnectionProvider.java | 2 +
.../connection/SimpleJdbcConnectionProvider.java | 18 +++
.../executor/InsertOrUpdateJdbcExecutor.java | 14 ++-
.../executor/JdbcBatchStatementExecutor.java | 8 +-
.../executor/KeyedBatchStatementExecutor.java | 13 ++-
.../executor/SimpleBatchStatementExecutor.java | 16 +--
.../jdbc/internal/options/JdbcOptions.java | 2 +
.../connector/jdbc/table/JdbcLookupFunction.java | 24 +++-
.../jdbc/table/JdbcRowDataLookupFunction.java | 24 +++-
.../jdbc/table/JdbcRowDataOutputFormat.java | 4 +-
.../connector/jdbc/JdbcLookupFunctionTest.java | 126 ++++++++++++++++++++
.../flink/connector/jdbc/JdbcOutputFormatTest.java | 45 +++++++
.../flink/connector/jdbc/JdbcTestFixture.java | 4 +
.../jdbc/table/JdbcLookupTableITCase.java | 78 +------------
.../connector/jdbc/table/JdbcLookupTestBase.java | 105 +++++++++++++++++
.../jdbc/table/JdbcRowDataLookupFunctionTest.java | 129 +++++++++++++++++++++
.../jdbc/table/JdbcRowDataOutputFormatTest.java | 61 ++++++++++
20 files changed, 591 insertions(+), 109 deletions(-)
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/AbstractJdbcOutputFormat.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/AbstractJdbcOutputFormat.java
index 2316401..72422ba 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/AbstractJdbcOutputFormat.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/AbstractJdbcOutputFormat.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.jdbc.internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.io.RichOutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.internal.connection.JdbcConnectionProvider;
@@ -42,7 +43,7 @@ public abstract class AbstractJdbcOutputFormat<T> extends RichOutputFormat<T> im
private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcOutputFormat.class);
protected transient Connection connection;
- private final JdbcConnectionProvider connectionProvider;
+ protected final JdbcConnectionProvider connectionProvider;
public AbstractJdbcOutputFormat(JdbcConnectionProvider connectionProvider) {
this.connectionProvider = Preconditions.checkNotNull(connectionProvider);
@@ -85,4 +86,9 @@ public abstract class AbstractJdbcOutputFormat<T> extends RichOutputFormat<T> im
@Override
public void flush() throws IOException {
}
+
+ @VisibleForTesting
+ public Connection getConnection() {
+ return connection;
+ }
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java
index 5d6c7bb..af41ed1 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/JdbcBatchingOutputFormat.java
@@ -47,6 +47,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
+import static org.apache.flink.connector.jdbc.internal.options.JdbcOptions.CONNECTION_CHECK_TIMEOUT_SECONDS;
import static org.apache.flink.connector.jdbc.utils.JdbcUtils.setRecordToStatement;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -129,7 +130,7 @@ public class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStat
private JdbcExec createAndOpenStatementExecutor(StatementExecutorFactory<JdbcExec> statementExecutorFactory) throws IOException {
JdbcExec exec = statementExecutorFactory.apply(getRuntimeContext());
try {
- exec.open(connection);
+ exec.prepareStatements(connection);
} catch (SQLException e) {
throw new IOException("unable to open JDBC writer", e);
}
@@ -176,6 +177,16 @@ public class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStat
throw new IOException(e);
}
try {
+ if (!connection.isValid(CONNECTION_CHECK_TIMEOUT_SECONDS)) {
+ connection = connectionProvider.reestablishConnection();
+ jdbcStatementExecutor.closeStatements();
+ jdbcStatementExecutor.prepareStatements(connection);
+ }
+ } catch (Exception excpetion) {
+ LOG.error("JDBC connection is not valid, and reestablish connection failed.", excpetion);
+ throw new IOException("Reestablish JDBC connection failed", excpetion);
+ }
+ try {
Thread.sleep(1000 * i);
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
@@ -215,7 +226,7 @@ public class JdbcBatchingOutputFormat<In, JdbcIn, JdbcExec extends JdbcBatchStat
try {
if (jdbcStatementExecutor != null) {
- jdbcStatementExecutor.close();
+ jdbcStatementExecutor.closeStatements();
}
} catch (SQLException e) {
LOG.warn("Close JDBC writer failed.", e);
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/TableJdbcUpsertOutputFormat.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/TableJdbcUpsertOutputFormat.java
index 05b5315..1865ebc 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/TableJdbcUpsertOutputFormat.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/TableJdbcUpsertOutputFormat.java
@@ -54,7 +54,7 @@ class TableJdbcUpsertOutputFormat extends JdbcBatchingOutputFormat<Tuple2<Boolea
super.open(taskNumber, numTasks);
deleteExecutor = createDeleteExecutor();
try {
- deleteExecutor.open(connection);
+ deleteExecutor.prepareStatements(connection);
} catch (SQLException e) {
throw new IOException(e);
}
@@ -83,7 +83,7 @@ class TableJdbcUpsertOutputFormat extends JdbcBatchingOutputFormat<Tuple2<Boolea
super.close();
} finally {
try {
- deleteExecutor.close();
+ deleteExecutor.closeStatements();
} catch (SQLException e) {
LOG.warn("unable to close delete statement runner", e);
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/JdbcConnectionProvider.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/JdbcConnectionProvider.java
index bba5748..f5a5e70 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/JdbcConnectionProvider.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/JdbcConnectionProvider.java
@@ -27,4 +27,6 @@ import java.sql.Connection;
@Internal
public interface JdbcConnectionProvider {
Connection getConnection() throws Exception;
+
+ Connection reestablishConnection() throws Exception;
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java
index a319a54..d9bac06 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProvider.java
@@ -19,6 +19,9 @@ package org.apache.flink.connector.jdbc.internal.connection;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -29,6 +32,8 @@ import java.sql.SQLException;
*/
public class SimpleJdbcConnectionProvider implements JdbcConnectionProvider, Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(SimpleJdbcConnectionProvider.class);
+
private static final long serialVersionUID = 1L;
private final JdbcConnectionOptions jdbcOptions;
@@ -55,4 +60,17 @@ public class SimpleJdbcConnectionProvider implements JdbcConnectionProvider, Ser
}
return connection;
}
+
+ @Override
+ public Connection reestablishConnection() throws SQLException, ClassNotFoundException {
+ try {
+ connection.close();
+ } catch (SQLException e) {
+ LOG.info("JDBC connection close failed.", e);
+ } finally {
+ connection = null;
+ }
+ connection = getConnection();
+ return connection;
+ }
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/InsertOrUpdateJdbcExecutor.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/InsertOrUpdateJdbcExecutor.java
index 593d121..14b758e 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/InsertOrUpdateJdbcExecutor.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/InsertOrUpdateJdbcExecutor.java
@@ -20,6 +20,9 @@ package org.apache.flink.connector.jdbc.internal.executor;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nonnull;
import java.sql.Connection;
@@ -40,6 +43,8 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
@Internal
public final class InsertOrUpdateJdbcExecutor<R, K, V> implements JdbcBatchStatementExecutor<R> {
+ private static final Logger LOG = LoggerFactory.getLogger(InsertOrUpdateJdbcExecutor.class);
+
private final String existSQL;
private final String insertSQL;
private final String updateSQL;
@@ -51,10 +56,11 @@ public final class InsertOrUpdateJdbcExecutor<R, K, V> implements JdbcBatchState
private final Function<R, K> keyExtractor;
private final Function<R, V> valueMapper;
+ private final Map<K, V> batch;
+
private transient PreparedStatement existStatement;
private transient PreparedStatement insertStatement;
private transient PreparedStatement updateStatement;
- private transient Map<K, V> batch = new HashMap<>();
public InsertOrUpdateJdbcExecutor(
@Nonnull String existSQL,
@@ -73,11 +79,11 @@ public final class InsertOrUpdateJdbcExecutor<R, K, V> implements JdbcBatchState
this.updateSetter = checkNotNull(updateSetter);
this.keyExtractor = checkNotNull(keyExtractor);
this.valueMapper = checkNotNull(valueExtractor);
+ this.batch = new HashMap<>();
}
@Override
- public void open(Connection connection) throws SQLException {
- batch = new HashMap<>();
+ public void prepareStatements(Connection connection) throws SQLException {
existStatement = connection.prepareStatement(existSQL);
insertStatement = connection.prepareStatement(insertSQL);
updateStatement = connection.prepareStatement(updateSQL);
@@ -118,7 +124,7 @@ public final class InsertOrUpdateJdbcExecutor<R, K, V> implements JdbcBatchState
}
@Override
- public void close() throws SQLException {
+ public void closeStatements() throws SQLException {
for (PreparedStatement s : Arrays.asList(existStatement, insertStatement, updateStatement)) {
if (s != null) {
s.close();
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/JdbcBatchStatementExecutor.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/JdbcBatchStatementExecutor.java
index 8b77d32..8e856c5 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/JdbcBatchStatementExecutor.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/JdbcBatchStatementExecutor.java
@@ -32,9 +32,9 @@ import java.util.function.Function;
public interface JdbcBatchStatementExecutor<T> {
/**
- * Open the writer by JDBC Connection. It can create Statement from Connection.
+ * Create statements from connection.
*/
- void open(Connection connection) throws SQLException;
+ void prepareStatements(Connection connection) throws SQLException;
void addToBatch(T record) throws SQLException;
@@ -44,9 +44,9 @@ public interface JdbcBatchStatementExecutor<T> {
void executeBatch() throws SQLException;
/**
- * Close JDBC related statements and other classes.
+ * Close JDBC related statements.
*/
- void close() throws SQLException;
+ void closeStatements() throws SQLException;
static <T, K> JdbcBatchStatementExecutor<T> keyed(String sql, Function<T, K> keyExtractor, JdbcStatementBuilder<K> statementBuilder) {
return new KeyedBatchStatementExecutor<>(sql, keyExtractor, statementBuilder);
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/KeyedBatchStatementExecutor.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/KeyedBatchStatementExecutor.java
index 785cd50..bff52d1 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/KeyedBatchStatementExecutor.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/KeyedBatchStatementExecutor.java
@@ -20,6 +20,9 @@ package org.apache.flink.connector.jdbc.internal.executor;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
@@ -32,11 +35,13 @@ import java.util.function.Function;
*/
class KeyedBatchStatementExecutor<T, K> implements JdbcBatchStatementExecutor<T> {
+ private static final Logger LOG = LoggerFactory.getLogger(KeyedBatchStatementExecutor.class);
+
private final String sql;
private final JdbcStatementBuilder<K> parameterSetter;
private final Function<T, K> keyExtractor;
+ private final Set<K> batch;
- private transient Set<K> batch = new HashSet<>();
private transient PreparedStatement st;
/**
@@ -46,11 +51,11 @@ class KeyedBatchStatementExecutor<T, K> implements JdbcBatchStatementExecutor<T>
this.parameterSetter = statementBuilder;
this.keyExtractor = keyExtractor;
this.sql = sql;
+ this.batch = new HashSet<>();
}
@Override
- public void open(Connection connection) throws SQLException {
- batch = new HashSet<>();
+ public void prepareStatements(Connection connection) throws SQLException {
st = connection.prepareStatement(sql);
}
@@ -72,7 +77,7 @@ class KeyedBatchStatementExecutor<T, K> implements JdbcBatchStatementExecutor<T>
}
@Override
- public void close() throws SQLException {
+ public void closeStatements() throws SQLException {
if (st != null) {
st.close();
st = null;
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/SimpleBatchStatementExecutor.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/SimpleBatchStatementExecutor.java
index d829fa5..974423d 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/SimpleBatchStatementExecutor.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/executor/SimpleBatchStatementExecutor.java
@@ -20,6 +20,9 @@ package org.apache.flink.connector.jdbc.internal.executor;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
@@ -32,22 +35,24 @@ import java.util.function.Function;
*/
class SimpleBatchStatementExecutor<T, V> implements JdbcBatchStatementExecutor<T> {
+ private static final Logger LOG = LoggerFactory.getLogger(SimpleBatchStatementExecutor.class);
+
private final String sql;
private final JdbcStatementBuilder<V> parameterSetter;
private final Function<T, V> valueTransformer;
+ private final List<V> batch;
private transient PreparedStatement st;
- private transient List<V> batch;
SimpleBatchStatementExecutor(String sql, JdbcStatementBuilder<V> statementBuilder, Function<T, V> valueTransformer) {
this.sql = sql;
this.parameterSetter = statementBuilder;
this.valueTransformer = valueTransformer;
+ this.batch = new ArrayList<>();
}
@Override
- public void open(Connection connection) throws SQLException {
- this.batch = new ArrayList<>();
+ public void prepareStatements(Connection connection) throws SQLException {
this.st = connection.prepareStatement(sql);
}
@@ -69,13 +74,10 @@ class SimpleBatchStatementExecutor<T, V> implements JdbcBatchStatementExecutor<T
}
@Override
- public void close() throws SQLException {
+ public void closeStatements() throws SQLException {
if (st != null) {
st.close();
st = null;
}
- if (batch != null) {
- batch.clear();
- }
}
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
index 6e25320..827b0cb 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/options/JdbcOptions.java
@@ -34,6 +34,8 @@ public class JdbcOptions extends JdbcConnectionOptions {
private static final long serialVersionUID = 1L;
+ public static final int CONNECTION_CHECK_TIMEOUT_SECONDS = 60;
+
private String tableName;
private JdbcDialect dialect;
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java
index 11803d0..9a4598e 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcLookupFunction.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.jdbc.table;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.internal.options.JdbcLookupOptions;
@@ -45,6 +46,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import static org.apache.flink.connector.jdbc.internal.options.JdbcOptions.CONNECTION_CHECK_TIMEOUT_SECONDS;
import static org.apache.flink.connector.jdbc.utils.JdbcUtils.getFieldFromResultSet;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -116,8 +118,7 @@ public class JdbcLookupFunction extends TableFunction<Row> {
@Override
public void open(FunctionContext context) throws Exception {
try {
- establishConnection();
- statement = dbConn.prepareStatement(query);
+ establishConnectionAndStatement();
this.cache = cacheMaxSize == -1 || cacheExpireMs == -1 ? null : CacheBuilder.newBuilder()
.expireAfterWrite(cacheExpireMs, TimeUnit.MILLISECONDS)
.maximumSize(cacheMaxSize)
@@ -171,6 +172,17 @@ public class JdbcLookupFunction extends TableFunction<Row> {
}
try {
+ if (!dbConn.isValid(CONNECTION_CHECK_TIMEOUT_SECONDS)) {
+ statement.close();
+ dbConn.close();
+ establishConnectionAndStatement();
+ }
+ } catch (SQLException | ClassNotFoundException excpetion) {
+ LOG.error("JDBC connection is not valid, and reestablish connection failed", excpetion);
+ throw new RuntimeException("Reestablish JDBC connection failed", excpetion);
+ }
+
+ try {
Thread.sleep(1000 * retry);
} catch (InterruptedException e1) {
throw new RuntimeException(e1);
@@ -187,13 +199,14 @@ public class JdbcLookupFunction extends TableFunction<Row> {
return row;
}
- private void establishConnection() throws SQLException, ClassNotFoundException {
+ private void establishConnectionAndStatement() throws SQLException, ClassNotFoundException {
Class.forName(drivername);
if (username == null) {
dbConn = DriverManager.getConnection(dbURL);
} else {
dbConn = DriverManager.getConnection(dbURL, username, password);
}
+ statement = dbConn.prepareStatement(query);
}
@Override
@@ -223,6 +236,11 @@ public class JdbcLookupFunction extends TableFunction<Row> {
}
}
+ @VisibleForTesting
+ public Connection getDbConnection() {
+ return dbConn;
+ }
+
@Override
public TypeInformation<Row> getResultType() {
return new RowTypeInfo(fieldTypes, fieldNames);
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java
index a075aee..3c3ecf2 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunction.java
@@ -19,6 +19,7 @@
package org.apache.flink.connector.jdbc.table;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.connector.jdbc.dialect.JdbcDialect;
import org.apache.flink.connector.jdbc.dialect.JdbcDialects;
import org.apache.flink.connector.jdbc.internal.converter.JdbcRowConverter;
@@ -49,6 +50,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import static org.apache.flink.connector.jdbc.internal.options.JdbcOptions.CONNECTION_CHECK_TIMEOUT_SECONDS;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -115,8 +117,7 @@ public class JdbcRowDataLookupFunction extends TableFunction<RowData> {
@Override
public void open(FunctionContext context) throws Exception {
try {
- establishConnection();
- statement = dbConn.prepareStatement(query);
+ establishConnectionAndStatement();
this.cache = cacheMaxSize == -1 || cacheExpireMs == -1 ? null : CacheBuilder.newBuilder()
.expireAfterWrite(cacheExpireMs, TimeUnit.MILLISECONDS)
.maximumSize(cacheMaxSize)
@@ -172,6 +173,17 @@ public class JdbcRowDataLookupFunction extends TableFunction<RowData> {
}
try {
+ if (!dbConn.isValid(CONNECTION_CHECK_TIMEOUT_SECONDS)) {
+ statement.close();
+ dbConn.close();
+ establishConnectionAndStatement();
+ }
+ } catch (SQLException | ClassNotFoundException excpetion) {
+ LOG.error("JDBC connection is not valid, and reestablish connection failed", excpetion);
+ throw new RuntimeException("Reestablish JDBC connection failed", excpetion);
+ }
+
+ try {
Thread.sleep(1000 * retry);
} catch (InterruptedException e1) {
throw new RuntimeException(e1);
@@ -180,13 +192,14 @@ public class JdbcRowDataLookupFunction extends TableFunction<RowData> {
}
}
- private void establishConnection() throws SQLException, ClassNotFoundException {
+ private void establishConnectionAndStatement() throws SQLException, ClassNotFoundException {
Class.forName(drivername);
if (username == null) {
dbConn = DriverManager.getConnection(dbURL);
} else {
dbConn = DriverManager.getConnection(dbURL, username, password);
}
+ statement = dbConn.prepareStatement(query);
}
@Override
@@ -215,4 +228,9 @@ public class JdbcRowDataLookupFunction extends TableFunction<RowData> {
}
}
}
+
+ @VisibleForTesting
+ public Connection getDbConnection() {
+ return dbConn;
+ }
}
diff --git a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataOutputFormat.java b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataOutputFormat.java
index b78eaca..4ce940f8 100644
--- a/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataOutputFormat.java
+++ b/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcRowDataOutputFormat.java
@@ -102,7 +102,7 @@ public class JdbcRowDataOutputFormat extends JdbcBatchingOutputFormat<RowData, R
deleteExecutor = createDeleteExecutor();
super.open(taskNumber, numTasks);
try {
- deleteExecutor.open(connection);
+ deleteExecutor.prepareStatements(connection);
} catch (SQLException e) {
throw new IOException(e);
}
@@ -140,7 +140,7 @@ public class JdbcRowDataOutputFormat extends JdbcBatchingOutputFormat<RowData, R
} finally {
try {
if (deleteExecutor != null) {
- deleteExecutor.close();
+ deleteExecutor.closeStatements();
}
} catch (SQLException e) {
LOG.warn("unable to close delete statement runner", e);
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcLookupFunctionTest.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcLookupFunctionTest.java
new file mode 100644
index 0000000..fecafd7
--- /dev/null
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcLookupFunctionTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.flink.connector.jdbc;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.jdbc.internal.options.JdbcLookupOptions;
+import org.apache.flink.connector.jdbc.internal.options.JdbcOptions;
+import org.apache.flink.connector.jdbc.table.JdbcLookupFunction;
+import org.apache.flink.connector.jdbc.table.JdbcLookupTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test suite for {@link JdbcLookupFunction}.
+ */
+public class JdbcLookupFunctionTest extends JdbcLookupTestBase {
+
+ public static final String DB_URL = "jdbc:derby:memory:lookup";
+ public static final String LOOKUP_TABLE = "lookup_table";
+ public static final String DB_DRIVER = "org.apache.derby.jdbc.EmbeddedDriver";
+
+ private static String[] fieldNames = new String[] {"id1", "id2", "comment1", "comment2"};
+ private static TypeInformation[] fieldTypes = new TypeInformation[] {
+ BasicTypeInfo.INT_TYPE_INFO,
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicTypeInfo.STRING_TYPE_INFO,
+ BasicTypeInfo.STRING_TYPE_INFO
+ };
+
+ private static String[] lookupKeys = new String[] {"id1", "id2"};
+
+ @Test
+ public void testEval() throws Exception {
+
+ JdbcLookupFunction lookupFunction = buildLookupFunction();
+ ListOutputCollector collector = new ListOutputCollector();
+ lookupFunction.setCollector(collector);
+
+ lookupFunction.open(null);
+
+ lookupFunction.eval(1, "1");
+
+ // close connection
+ lookupFunction.getDbConnection().close();
+
+ lookupFunction.eval(2, "3");
+
+ List<String> result = Lists.newArrayList(collector.getOutputs()).stream()
+ .map(Row::toString)
+ .sorted()
+ .collect(Collectors.toList());
+
+ List<String> expected = new ArrayList<>();
+ expected.add("1,1,11-c1-v1,11-c2-v1");
+ expected.add("1,1,11-c1-v2,11-c2-v2");
+ expected.add("2,3,null,23-c2");
+ Collections.sort(expected);
+
+ assertEquals(expected, result);
+ }
+
+ private JdbcLookupFunction buildLookupFunction() {
+ JdbcOptions jdbcOptions = JdbcOptions.builder()
+ .setDriverName(DB_DRIVER)
+ .setDBUrl(DB_URL)
+ .setTableName(LOOKUP_TABLE)
+ .build();
+
+ JdbcLookupOptions lookupOptions = JdbcLookupOptions.builder().build();
+
+ JdbcLookupFunction lookupFunction = JdbcLookupFunction.builder()
+ .setOptions(jdbcOptions)
+ .setLookupOptions(lookupOptions)
+ .setFieldTypes(fieldTypes)
+ .setFieldNames(fieldNames)
+ .setKeyNames(lookupKeys)
+ .build();
+
+ return lookupFunction;
+ }
+
+ private static final class ListOutputCollector implements Collector<Row> {
+
+ private final List<Row> output = new ArrayList<>();
+
+ @Override
+ public void collect(Row row) {
+ this.output.add(row);
+ }
+
+ @Override
+ public void close() {}
+
+ public List<Row> getOutputs() {
+ return output;
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcOutputFormatTest.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcOutputFormatTest.java
index 519b357..226d86f 100644
--- a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcOutputFormatTest.java
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcOutputFormatTest.java
@@ -38,8 +38,10 @@ import static org.apache.flink.connector.jdbc.JdbcTestFixture.INPUT_TABLE;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.INSERT_TEMPLATE;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.OUTPUT_TABLE;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.OUTPUT_TABLE_2;
+import static org.apache.flink.connector.jdbc.JdbcTestFixture.OUTPUT_TABLE_3;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.SELECT_ALL_NEWBOOKS;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.SELECT_ALL_NEWBOOKS_2;
+import static org.apache.flink.connector.jdbc.JdbcTestFixture.SELECT_ALL_NEWBOOKS_3;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.TEST_DATA;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.TestEntry;
import static org.apache.flink.util.ExceptionUtils.findThrowable;
@@ -294,6 +296,49 @@ public class JdbcOutputFormatTest extends JdbcDataTestBase {
}
}
+ @Test
+ public void testInvalidConnectionInJdbcOutputFormat() throws IOException, SQLException {
+ jdbcOutputFormat = JdbcOutputFormat.buildJdbcOutputFormat()
+ .setDrivername(DERBY_EBOOKSHOP_DB.getDriverClass())
+ .setDBUrl(DERBY_EBOOKSHOP_DB.getUrl())
+ .setQuery(String.format(INSERT_TEMPLATE, OUTPUT_TABLE_3))
+ .finish();
+ setRuntimeContext(jdbcOutputFormat, true);
+ jdbcOutputFormat.open(0, 1);
+
+ // write records
+ for (int i = 0; i < 3; i++) {
+ jdbcOutputFormat.writeRecord(toRow(TEST_DATA[i]));
+ }
+
+ // close connection
+ jdbcOutputFormat.getConnection().close();
+
+ for (int i = 3; i < TEST_DATA.length; i++) {
+ jdbcOutputFormat.writeRecord(toRow(TEST_DATA[i]));
+ }
+
+ jdbcOutputFormat.close();
+
+ try (
+ Connection dbConn = DriverManager.getConnection(DERBY_EBOOKSHOP_DB.getUrl());
+ PreparedStatement statement = dbConn.prepareStatement(SELECT_ALL_NEWBOOKS_3);
+ ResultSet resultSet = statement.executeQuery()
+ ) {
+ int recordCount = 0;
+ while (resultSet.next()) {
+ assertEquals(TEST_DATA[recordCount].id, resultSet.getObject("id"));
+ assertEquals(TEST_DATA[recordCount].title, resultSet.getObject("title"));
+ assertEquals(TEST_DATA[recordCount].author, resultSet.getObject("author"));
+ assertEquals(TEST_DATA[recordCount].price, resultSet.getObject("price"));
+ assertEquals(TEST_DATA[recordCount].qty, resultSet.getObject("qty"));
+
+ recordCount++;
+ }
+ assertEquals(TEST_DATA.length, recordCount);
+ }
+ }
+
@After
public void clearOutputTable() throws Exception {
Class.forName(DERBY_EBOOKSHOP_DB.getDriverClass());
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcTestFixture.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcTestFixture.java
index e4d7415..eb57ebc 100644
--- a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcTestFixture.java
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/JdbcTestFixture.java
@@ -40,10 +40,12 @@ public class JdbcTestFixture {
public static final String INPUT_TABLE = "books";
public static final String OUTPUT_TABLE = "newbooks";
public static final String OUTPUT_TABLE_2 = "newbooks2";
+ public static final String OUTPUT_TABLE_3 = "newbooks3";
public static final String SELECT_ALL_BOOKS = "select * from " + INPUT_TABLE;
public static final String SELECT_ID_BOOKS = "select id from " + INPUT_TABLE;
public static final String SELECT_ALL_NEWBOOKS = "select * from " + OUTPUT_TABLE;
public static final String SELECT_ALL_NEWBOOKS_2 = "select * from " + OUTPUT_TABLE_2;
+ public static final String SELECT_ALL_NEWBOOKS_3 = "select * from " + OUTPUT_TABLE_3;
public static final String SELECT_EMPTY = "select * from books WHERE QTY < 0";
public static final String INSERT_TEMPLATE = "insert into %s (id, title, author, price, qty) values (?,?,?,?,?)";
public static final String SELECT_ALL_BOOKS_SPLIT_BY_ID = SELECT_ALL_BOOKS + " WHERE id BETWEEN ? AND ?";
@@ -144,6 +146,7 @@ public class JdbcTestFixture {
createTable(conn, JdbcTestFixture.INPUT_TABLE);
createTable(conn, OUTPUT_TABLE);
createTable(conn, OUTPUT_TABLE_2);
+ createTable(conn, OUTPUT_TABLE_3);
}
}
@@ -174,6 +177,7 @@ public class JdbcTestFixture {
stat.executeUpdate("DROP TABLE " + INPUT_TABLE);
stat.executeUpdate("DROP TABLE " + OUTPUT_TABLE);
stat.executeUpdate("DROP TABLE " + OUTPUT_TABLE_2);
+ stat.executeUpdate("DROP TABLE " + OUTPUT_TABLE_3);
}
}
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcLookupTableITCase.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcLookupTableITCase.java
index 783ebdf..8babd64 100644
--- a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcLookupTableITCase.java
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcLookupTableITCase.java
@@ -19,7 +19,6 @@
package org.apache.flink.connector.jdbc.table;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.connector.jdbc.JdbcTestFixture;
import org.apache.flink.connector.jdbc.internal.options.JdbcLookupOptions;
import org.apache.flink.connector.jdbc.internal.options.JdbcOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -28,21 +27,14 @@ import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.types.DataType;
-import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.types.Row;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
-import org.junit.After;
-import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.SQLException;
-import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -51,7 +43,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.stream.Collectors;
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.DERBY_EBOOKSHOP_DB;
import static org.apache.flink.table.api.Expressions.$;
import static org.junit.Assert.assertEquals;
@@ -59,10 +50,7 @@ import static org.junit.Assert.assertEquals;
* IT case for lookup source of JDBC connector.
*/
@RunWith(Parameterized.class)
-public class JdbcLookupTableITCase extends AbstractTestBase {
-
- public static final String DB_URL = "jdbc:derby:memory:lookup";
- public static final String LOOKUP_TABLE = "lookup_table";
+public class JdbcLookupTableITCase extends JdbcLookupTestBase {
private final String tableFactory;
private final boolean useCache;
@@ -82,70 +70,6 @@ public class JdbcLookupTableITCase extends AbstractTestBase {
{"dynamicFactory", false}});
}
- @Before
- public void before() throws ClassNotFoundException, SQLException {
- System.setProperty("derby.stream.error.field", JdbcTestFixture.class.getCanonicalName() + ".DEV_NULL");
-
- Class.forName(DERBY_EBOOKSHOP_DB.getDriverClass());
- try (
- Connection conn = DriverManager.getConnection(DB_URL + ";create=true");
- Statement stat = conn.createStatement()) {
- stat.executeUpdate("CREATE TABLE " + LOOKUP_TABLE + " (" +
- "id1 INT NOT NULL DEFAULT 0," +
- "id2 VARCHAR(20) NOT NULL," +
- "comment1 VARCHAR(1000)," +
- "comment2 VARCHAR(1000))");
-
- Object[][] data = new Object[][] {
- new Object[] {1, "1", "11-c1-v1", "11-c2-v1"},
- new Object[] {1, "1", "11-c1-v2", "11-c2-v2"},
- new Object[] {2, "3", null, "23-c2"},
- new Object[] {2, "5", "25-c1", "25-c2"},
- new Object[] {3, "8", "38-c1", "38-c2"}
- };
- boolean[] surroundedByQuotes = new boolean[] {
- false, true, true, true
- };
-
- StringBuilder sqlQueryBuilder = new StringBuilder(
- "INSERT INTO " + LOOKUP_TABLE + " (id1, id2, comment1, comment2) VALUES ");
- for (int i = 0; i < data.length; i++) {
- sqlQueryBuilder.append("(");
- for (int j = 0; j < data[i].length; j++) {
- if (data[i][j] == null) {
- sqlQueryBuilder.append("null");
- } else {
- if (surroundedByQuotes[j]) {
- sqlQueryBuilder.append("'");
- }
- sqlQueryBuilder.append(data[i][j]);
- if (surroundedByQuotes[j]) {
- sqlQueryBuilder.append("'");
- }
- }
- if (j < data[i].length - 1) {
- sqlQueryBuilder.append(", ");
- }
- }
- sqlQueryBuilder.append(")");
- if (i < data.length - 1) {
- sqlQueryBuilder.append(", ");
- }
- }
- stat.execute(sqlQueryBuilder.toString());
- }
- }
-
- @After
- public void clearOutputTable() throws Exception {
- Class.forName(DERBY_EBOOKSHOP_DB.getDriverClass());
- try (
- Connection conn = DriverManager.getConnection(DB_URL);
- Statement stat = conn.createStatement()) {
- stat.execute("DROP TABLE " + LOOKUP_TABLE);
- }
- }
-
@Test
public void testLookup() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcLookupTestBase.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcLookupTestBase.java
new file mode 100644
index 0000000..5523ae2
--- /dev/null
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcLookupTestBase.java
@@ -0,0 +1,105 @@
+/*
+ * 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.flink.connector.jdbc.table;
+
+import org.apache.flink.connector.jdbc.JdbcTestFixture;
+import org.apache.flink.test.util.AbstractTestBase;
+
+import org.junit.After;
+import org.junit.Before;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.apache.flink.connector.jdbc.JdbcTestFixture.DERBY_EBOOKSHOP_DB;
+
+/**
+ * Base class for JDBC lookup test.
+ */
+public class JdbcLookupTestBase extends AbstractTestBase {
+
+ public static final String DB_URL = "jdbc:derby:memory:lookup";
+ public static final String LOOKUP_TABLE = "lookup_table";
+
+ @Before
+ public void before() throws ClassNotFoundException, SQLException {
+ System.setProperty("derby.stream.error.field", JdbcTestFixture.class.getCanonicalName() + ".DEV_NULL");
+
+ Class.forName(DERBY_EBOOKSHOP_DB.getDriverClass());
+ try (
+ Connection conn = DriverManager.getConnection(DB_URL + ";create=true");
+ Statement stat = conn.createStatement()) {
+ stat.executeUpdate("CREATE TABLE " + LOOKUP_TABLE + " (" +
+ "id1 INT NOT NULL DEFAULT 0," +
+ "id2 VARCHAR(20) NOT NULL," +
+ "comment1 VARCHAR(1000)," +
+ "comment2 VARCHAR(1000))");
+
+ Object[][] data = new Object[][] {
+ new Object[] {1, "1", "11-c1-v1", "11-c2-v1"},
+ new Object[] {1, "1", "11-c1-v2", "11-c2-v2"},
+ new Object[] {2, "3", null, "23-c2"},
+ new Object[] {2, "5", "25-c1", "25-c2"},
+ new Object[] {3, "8", "38-c1", "38-c2"}
+ };
+ boolean[] surroundedByQuotes = new boolean[] {
+ false, true, true, true
+ };
+
+ StringBuilder sqlQueryBuilder = new StringBuilder(
+ "INSERT INTO " + LOOKUP_TABLE + " (id1, id2, comment1, comment2) VALUES ");
+ for (int i = 0; i < data.length; i++) {
+ sqlQueryBuilder.append("(");
+ for (int j = 0; j < data[i].length; j++) {
+ if (data[i][j] == null) {
+ sqlQueryBuilder.append("null");
+ } else {
+ if (surroundedByQuotes[j]) {
+ sqlQueryBuilder.append("'");
+ }
+ sqlQueryBuilder.append(data[i][j]);
+ if (surroundedByQuotes[j]) {
+ sqlQueryBuilder.append("'");
+ }
+ }
+ if (j < data[i].length - 1) {
+ sqlQueryBuilder.append(", ");
+ }
+ }
+ sqlQueryBuilder.append(")");
+ if (i < data.length - 1) {
+ sqlQueryBuilder.append(", ");
+ }
+ }
+ stat.execute(sqlQueryBuilder.toString());
+ }
+ }
+
+ @After
+ public void clearOutputTable() throws Exception {
+ Class.forName(DERBY_EBOOKSHOP_DB.getDriverClass());
+ try (
+ Connection conn = DriverManager.getConnection(DB_URL);
+ Statement stat = conn.createStatement()) {
+ stat.execute("DROP TABLE " + LOOKUP_TABLE);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunctionTest.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunctionTest.java
new file mode 100644
index 0000000..5dc4398
--- /dev/null
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataLookupFunctionTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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.flink.connector.jdbc.table;
+
+import org.apache.flink.connector.jdbc.internal.options.JdbcLookupOptions;
+import org.apache.flink.connector.jdbc.internal.options.JdbcOptions;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Collector;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.connector.jdbc.JdbcTestFixture.DERBY_EBOOKSHOP_DB;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test suite for {@link JdbcRowDataLookupFunction}.
+ */
+public class JdbcRowDataLookupFunctionTest extends JdbcLookupTestBase {
+
+ private static String[] fieldNames = new String[] {"id1", "id2", "comment1", "comment2"};
+ private static DataType[] fieldDataTypes = new DataType[] {
+ DataTypes.INT(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ };
+
+ private static String[] lookupKeys = new String[] {"id1", "id2"};
+
+ @Test
+ public void testEval() throws Exception {
+
+ JdbcRowDataLookupFunction lookupFunction = buildRowDataLookupFunction();
+
+ ListOutputCollector collector = new ListOutputCollector();
+ lookupFunction.setCollector(collector);
+
+ lookupFunction.open(null);
+
+ lookupFunction.eval(1, StringData.fromString("1"));
+
+ // close connection
+ lookupFunction.getDbConnection().close();
+
+ lookupFunction.eval(2, StringData.fromString("3"));
+
+ List<String> result = Lists.newArrayList(collector.getOutputs()).stream()
+ .map(RowData::toString)
+ .sorted()
+ .collect(Collectors.toList());
+
+ List<String> expected = new ArrayList<>();
+ expected.add("+I(1,1,11-c1-v1,11-c2-v1)");
+ expected.add("+I(1,1,11-c1-v2,11-c2-v2)");
+ expected.add("+I(2,3,null,23-c2)");
+ Collections.sort(expected);
+
+ assertEquals(expected, result);
+ }
+
+ private JdbcRowDataLookupFunction buildRowDataLookupFunction() {
+ JdbcOptions jdbcOptions = JdbcOptions.builder()
+ .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass())
+ .setDBUrl(DB_URL)
+ .setTableName(LOOKUP_TABLE)
+ .build();
+
+ JdbcLookupOptions lookupOptions = JdbcLookupOptions.builder().build();
+
+ RowType rowType = RowType.of(Arrays.stream(fieldDataTypes).
+ map(DataType::getLogicalType).toArray(LogicalType[]::new), fieldNames);
+
+ JdbcRowDataLookupFunction lookupFunction = new JdbcRowDataLookupFunction(
+ jdbcOptions,
+ lookupOptions,
+ fieldNames,
+ fieldDataTypes,
+ lookupKeys,
+ rowType);
+
+ return lookupFunction;
+ }
+
+ private static final class ListOutputCollector implements Collector<RowData> {
+
+ private final List<RowData> output = new ArrayList<>();
+
+ @Override
+ public void collect(RowData row) {
+ this.output.add(row);
+ }
+
+ @Override
+ public void close() {}
+
+ public List<RowData> getOutputs() {
+ return output;
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataOutputFormatTest.java b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataOutputFormatTest.java
index 21f12e0..4b3de68 100644
--- a/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataOutputFormatTest.java
+++ b/flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/JdbcRowDataOutputFormatTest.java
@@ -45,8 +45,10 @@ import static org.apache.flink.connector.jdbc.JdbcTestFixture.DERBY_EBOOKSHOP_DB
import static org.apache.flink.connector.jdbc.JdbcTestFixture.INPUT_TABLE;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.OUTPUT_TABLE;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.OUTPUT_TABLE_2;
+import static org.apache.flink.connector.jdbc.JdbcTestFixture.OUTPUT_TABLE_3;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.SELECT_ALL_NEWBOOKS;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.SELECT_ALL_NEWBOOKS_2;
+import static org.apache.flink.connector.jdbc.JdbcTestFixture.SELECT_ALL_NEWBOOKS_3;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.TEST_DATA;
import static org.apache.flink.connector.jdbc.JdbcTestFixture.TestEntry;
import static org.apache.flink.util.ExceptionUtils.findThrowable;
@@ -359,6 +361,65 @@ public class JdbcRowDataOutputFormatTest extends JdbcDataTestBase {
}
}
+ @Test
+ public void testInvalidConnectionInJdbcOutputFormat() throws IOException, SQLException {
+ JdbcOptions jdbcOptions = JdbcOptions.builder()
+ .setDriverName(DERBY_EBOOKSHOP_DB.getDriverClass())
+ .setDBUrl(DERBY_EBOOKSHOP_DB.getUrl())
+ .setTableName(OUTPUT_TABLE_3)
+ .build();
+ JdbcDmlOptions dmlOptions = JdbcDmlOptions.builder()
+ .withTableName(jdbcOptions.getTableName())
+ .withDialect(jdbcOptions.getDialect())
+ .withFieldNames(fieldNames)
+ .build();
+
+ outputFormat = JdbcRowDataOutputFormat.dynamicOutputFormatBuilder()
+ .setJdbcOptions(jdbcOptions)
+ .setFieldDataTypes(fieldDataTypes)
+ .setJdbcDmlOptions(dmlOptions)
+ .setJdbcExecutionOptions(JdbcExecutionOptions.builder().build())
+ .setRowDataTypeInfo(rowDataTypeInfo)
+ .build();
+ setRuntimeContext(outputFormat, true);
+ outputFormat.open(0, 1);
+
+ // write records
+ for (int i = 0; i < 3; i++) {
+ TestEntry entry = TEST_DATA[i];
+ outputFormat.writeRecord(buildGenericData(entry.id, entry.title, entry.author, entry.price, entry.qty));
+ }
+
+ // close connection
+ outputFormat.getConnection().close();
+
+ // continue to write rest records
+ for (int i = 3; i < TEST_DATA.length; i++) {
+ TestEntry entry = TEST_DATA[i];
+ outputFormat.writeRecord(buildGenericData(entry.id, entry.title, entry.author, entry.price, entry.qty));
+ }
+
+ outputFormat.close();
+
+ try (
+ Connection dbConn = DriverManager.getConnection(DERBY_EBOOKSHOP_DB.getUrl());
+ PreparedStatement statement = dbConn.prepareStatement(SELECT_ALL_NEWBOOKS_3);
+ ResultSet resultSet = statement.executeQuery()
+ ) {
+ int recordCount = 0;
+ while (resultSet.next()) {
+ assertEquals(TEST_DATA[recordCount].id, resultSet.getObject("id"));
+ assertEquals(TEST_DATA[recordCount].title, resultSet.getObject("title"));
+ assertEquals(TEST_DATA[recordCount].author, resultSet.getObject("author"));
+ assertEquals(TEST_DATA[recordCount].price, resultSet.getObject("price"));
+ assertEquals(TEST_DATA[recordCount].qty, resultSet.getObject("qty"));
+
+ recordCount++;
+ }
+ assertEquals(TEST_DATA.length, recordCount);
+ }
+ }
+
@After
public void clearOutputTable() throws Exception {
Class.forName(DERBY_EBOOKSHOP_DB.getDriverClass());