slinkydeveloper commented on a change in pull request #18653:
URL: https://github.com/apache/flink/pull/18653#discussion_r803604728
##########
File path:
flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/UnsignedTypeConversionITCase.java
##########
@@ -94,70 +103,76 @@
.withDatabaseName(DEFAULT_DB_NAME)
.withLogConsumer(new Slf4jLogConsumer(LOGGER));
- private Connection connection;
private StreamTableEnvironment tEnv;
@Test
public void testUnsignedType() throws Exception {
prepare();
// write data to db
- tEnv.executeSql(String.format("insert into jdbc_sink select %s from
data", COLUMNS))
+ tEnv.executeSql(
+ String.format(
+ "insert into jdbc_sink select %s from data",
+ String.join(",", COLUMNS)))
.await();
// read data from db using jdbc connection and compare
- PreparedStatement query =
- connection.prepareStatement(
- String.format("select %s from %s", COLUMNS,
TABLE_NAME));
- ResultSet resultSet = query.executeQuery();
- while (resultSet.next()) {
-
Assertions.assertThat(resultSet.getObject("tiny_c")).isEqualTo(127);
-
Assertions.assertThat(resultSet.getObject("tiny_un_c")).isEqualTo(255);
-
Assertions.assertThat(resultSet.getObject("small_c")).isEqualTo(32767);
-
Assertions.assertThat(resultSet.getObject("small_un_c")).isEqualTo(65535);
-
Assertions.assertThat(resultSet.getObject("int_c")).isEqualTo(2147483647);
-
Assertions.assertThat(resultSet.getObject("int_un_c")).isEqualTo(4294967295L);
-
Assertions.assertThat(resultSet.getObject("big_c")).isEqualTo(9223372036854775807L);
- Assertions.assertThat(resultSet.getObject("big_un_c"))
- .isEqualTo(new BigInteger("18446744073709551615"));
+ try (Connection con =
+
DriverManager.getConnection(MYSQL_CONTAINER.getJdbcUrl(), USER, PASSWORD);
+ PreparedStatement ps =
+ con.prepareStatement(
+ String.format(
+ "select %s from %s",
+ String.join(",", COLUMNS),
TABLE_NAME))) {
+ ResultSet resultSet = ps.executeQuery();
+ while (resultSet.next()) {
+ assertThat(resultSet.getObject("tiny_c")).isEqualTo(127);
+ assertThat(resultSet.getObject("tiny_un_c")).isEqualTo(255);
+ assertThat(resultSet.getObject("small_c")).isEqualTo(32767);
+ assertThat(resultSet.getObject("small_un_c")).isEqualTo(65535);
+ assertThat(resultSet.getObject("int_c")).isEqualTo(2147483647);
+
assertThat(resultSet.getObject("int_un_c")).isEqualTo(4294967295L);
+
assertThat(resultSet.getObject("big_c")).isEqualTo(9223372036854775807L);
+ assertThat(resultSet.getObject("big_un_c"))
+ .isEqualTo(new BigInteger("18446744073709551615"));
+ }
}
// read data from db using flink and compare
Iterator<Row> collected =
- tEnv.executeSql(String.format("select %s from jdbc_source",
COLUMNS)).collect();
+ tEnv.executeSql(
+ String.format(
+ "select %s from jdbc_source",
String.join(",", COLUMNS)))
+ .collect();
List<Row> result = CollectionUtil.iteratorToList(collected);
- List<Row> expected =
Collections.singletonList(Row.ofKind(RowKind.INSERT, ROW));
- Assertions.assertThat(result).isEqualTo(expected);
-
- connection.close();
+ assertThat(result).containsOnly(Row.ofKind(RowKind.INSERT, ROW));
}
private void prepare() throws Exception {
- MYSQL_CONTAINER.start();
-
- connection = DriverManager.getConnection(MYSQL_CONTAINER.getJdbcUrl(),
USER, PASSWORD);
tEnv =
StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment());
-
createMysqlTable();
createFlinkTable();
prepareData();
}
private void createMysqlTable() throws SQLException {
- PreparedStatement ddlStatement =
- connection.prepareStatement(
- "create table "
- + TABLE_NAME
- + " ("
- + " tiny_c TINYINT,"
- + " tiny_un_c TINYINT UNSIGNED,"
- + " small_c SMALLINT,"
- + " small_un_c SMALLINT UNSIGNED,"
- + " int_c INTEGER ,"
- + " int_un_c INTEGER UNSIGNED,"
- + " big_c BIGINT,"
- + " big_un_c BIGINT UNSIGNED);");
- ddlStatement.execute();
+ try (Connection con =
Review comment:
Get the connection as parameter here rather than creating a new one?
##########
File path:
flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/table/UnsignedTypeConversionITCase.java
##########
@@ -94,70 +103,76 @@
.withDatabaseName(DEFAULT_DB_NAME)
.withLogConsumer(new Slf4jLogConsumer(LOGGER));
- private Connection connection;
private StreamTableEnvironment tEnv;
@Test
public void testUnsignedType() throws Exception {
prepare();
// write data to db
- tEnv.executeSql(String.format("insert into jdbc_sink select %s from
data", COLUMNS))
+ tEnv.executeSql(
+ String.format(
+ "insert into jdbc_sink select %s from data",
+ String.join(",", COLUMNS)))
.await();
// read data from db using jdbc connection and compare
- PreparedStatement query =
- connection.prepareStatement(
- String.format("select %s from %s", COLUMNS,
TABLE_NAME));
- ResultSet resultSet = query.executeQuery();
- while (resultSet.next()) {
-
Assertions.assertThat(resultSet.getObject("tiny_c")).isEqualTo(127);
-
Assertions.assertThat(resultSet.getObject("tiny_un_c")).isEqualTo(255);
-
Assertions.assertThat(resultSet.getObject("small_c")).isEqualTo(32767);
-
Assertions.assertThat(resultSet.getObject("small_un_c")).isEqualTo(65535);
-
Assertions.assertThat(resultSet.getObject("int_c")).isEqualTo(2147483647);
-
Assertions.assertThat(resultSet.getObject("int_un_c")).isEqualTo(4294967295L);
-
Assertions.assertThat(resultSet.getObject("big_c")).isEqualTo(9223372036854775807L);
- Assertions.assertThat(resultSet.getObject("big_un_c"))
- .isEqualTo(new BigInteger("18446744073709551615"));
+ try (Connection con =
+
DriverManager.getConnection(MYSQL_CONTAINER.getJdbcUrl(), USER, PASSWORD);
+ PreparedStatement ps =
+ con.prepareStatement(
+ String.format(
+ "select %s from %s",
+ String.join(",", COLUMNS),
TABLE_NAME))) {
+ ResultSet resultSet = ps.executeQuery();
+ while (resultSet.next()) {
+ assertThat(resultSet.getObject("tiny_c")).isEqualTo(127);
+ assertThat(resultSet.getObject("tiny_un_c")).isEqualTo(255);
+ assertThat(resultSet.getObject("small_c")).isEqualTo(32767);
+ assertThat(resultSet.getObject("small_un_c")).isEqualTo(65535);
+ assertThat(resultSet.getObject("int_c")).isEqualTo(2147483647);
+
assertThat(resultSet.getObject("int_un_c")).isEqualTo(4294967295L);
+
assertThat(resultSet.getObject("big_c")).isEqualTo(9223372036854775807L);
+ assertThat(resultSet.getObject("big_un_c"))
+ .isEqualTo(new BigInteger("18446744073709551615"));
+ }
}
// read data from db using flink and compare
Iterator<Row> collected =
- tEnv.executeSql(String.format("select %s from jdbc_source",
COLUMNS)).collect();
+ tEnv.executeSql(
+ String.format(
+ "select %s from jdbc_source",
String.join(",", COLUMNS)))
+ .collect();
List<Row> result = CollectionUtil.iteratorToList(collected);
- List<Row> expected =
Collections.singletonList(Row.ofKind(RowKind.INSERT, ROW));
- Assertions.assertThat(result).isEqualTo(expected);
-
- connection.close();
+ assertThat(result).containsOnly(Row.ofKind(RowKind.INSERT, ROW));
}
private void prepare() throws Exception {
- MYSQL_CONTAINER.start();
-
- connection = DriverManager.getConnection(MYSQL_CONTAINER.getJdbcUrl(),
USER, PASSWORD);
tEnv =
StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment());
Review comment:
Can you also make this a local variable? It's always better in tests to
have methods with parameters, rather than having test class fields. This is
also in order to simplify future JUnit 5 test porting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
