reswqa commented on code in PR #21977:
URL: https://github.com/apache/flink/pull/21977#discussion_r1112634137
##########
flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSinkITCase.java:
##########
@@ -398,39 +399,38 @@ public void testStreamingSinkWithTimestampLtzWatermark()
throws Exception {
}
@Test
- public void testStreamingSinkWithoutCommitPolicy() throws Exception {
+ void testStreamingSinkWithoutCommitPolicy() {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv =
HiveTestUtils.createTableEnvInStreamingMode(env);
tableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);
tableEnv.useCatalog(hiveCatalog.getName());
tableEnv.executeSql("create database db1");
- try {
- tableEnv.useDatabase("db1");
- tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
- tableEnv.executeSql("create table dest(x int) partitioned by (p
string)");
-
- tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
- tableEnv.executeSql(
- "create table src (i int, p string) with ("
- + "'connector'='datagen',"
- + "'number-of-rows'='5')");
- tableEnv.executeSql("insert into dest select * from src").await();
- fail("Streaming write partitioned table without commit policy
should fail");
- } catch (FlinkHiveException e) {
- // expected
- assertThat(e.getMessage())
- .contains(
- String.format(
- "Streaming write to partitioned hive table
`%s`.`%s`.`%s` without providing a commit policy",
- hiveCatalog.getName(), "db1", "dest"));
- } finally {
- tableEnv.executeSql("drop database db1 cascade");
- }
+
+ tableEnv.useDatabase("db1");
+ tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
+ tableEnv.executeSql("create table dest(x int) partitioned by (p
string)");
+
+ tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
+ tableEnv.executeSql(
+ "create table src (i int, p string) with ("
+ + "'connector'='datagen',"
+ + "'number-of-rows'='5')");
+
+ assertThatThrownBy(
+ () -> tableEnv.executeSql("insert into dest select *
from src").await(),
+ "Streaming write partitioned table without commit
policy should fail")
+ .isInstanceOf(FlinkHiveException.class)
+ .hasMessageContaining(
+ String.format(
+ "Streaming write to partitioned hive table
`%s`.`%s`.`%s` without providing a commit policy",
+ hiveCatalog.getName(), "db1", "dest"));
+
+ tableEnv.executeSql("drop database db1 cascade");
Review Comment:
good catch, fixed.
--
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
