This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kudu.git
The following commit(s) were added to refs/heads/main by this push:
new e569fa7 [FLINK-37690] Fix Kudu test table create/cleanup
e569fa7 is described below
commit e569fa7b973783a600782d78723a79dfa656201d
Author: Ferenc Csaky <[email protected]>
AuthorDate: Tue Jun 23 13:32:57 2026 +0200
[FLINK-37690] Fix Kudu test table create/cleanup
---
.../connector/kudu/connector/KuduTestBase.java | 9 +++++
.../kudu/format/KuduRowDataInputFormatTest.java | 34 +++++++++++--------
.../kudu/format/KuduRowInputFormatTest.java | 34 +++++++++++--------
.../connector/kudu/table/KuduDynamicSinkTest.java | 25 ++++++++------
.../kudu/table/KuduDynamicSourceTest.java | 38 ++++++++++++----------
.../kudu/table/KuduRowDataLookupFunctionTest.java | 11 ++++---
.../kudu/table/KuduTableSourceITCase.java | 11 ++++---
7 files changed, 96 insertions(+), 66 deletions(-)
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/connector/KuduTestBase.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/connector/KuduTestBase.java
index 2b00efb..72150b2 100644
---
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/connector/KuduTestBase.java
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/connector/KuduTestBase.java
@@ -58,6 +58,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.UUID;
import java.util.stream.Collectors;
import static org.junit.Assert.assertFalse;
@@ -178,6 +179,14 @@ public class KuduTestBase {
return tableInfo;
}
+ protected static KuduTableInfo uniqueBooksTableInfo(boolean
createIfNotExist) {
+ return booksTableInfo(uniqueTableName("books"), createIfNotExist);
+ }
+
+ private static String uniqueTableName(String prefix) {
+ return prefix + "_" + UUID.randomUUID().toString().replace("-", "");
+ }
+
public static List<Tuple5<Integer, String, String, Double, Integer>>
booksDataTuple() {
return Arrays.stream(booksTableData)
.map(
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/format/KuduRowDataInputFormatTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/format/KuduRowDataInputFormatTest.java
index 8c4f648..21d3e58 100644
---
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/format/KuduRowDataInputFormatTest.java
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/format/KuduRowDataInputFormatTest.java
@@ -38,7 +38,7 @@ class KuduRowDataInputFormatTest extends KuduTestBase {
@Test
void testInvalidKuduMaster() {
- KuduTableInfo tableInfo = booksTableInfo("books", false);
+ KuduTableInfo tableInfo = uniqueBooksTableInfo(false);
Assertions.assertThrows(
NullPointerException.class,
() -> new KuduRowDataInputFormat(null, new
RowResultRowDataConverter(), tableInfo));
@@ -58,28 +58,34 @@ class KuduRowDataInputFormatTest extends KuduTestBase {
@Test
void testInputFormat() throws Exception {
- KuduTableInfo tableInfo = booksTableInfo("books", true);
- setUpDatabase(tableInfo);
+ KuduTableInfo tableInfo = uniqueBooksTableInfo(true);
+ try {
+ setUpDatabase(tableInfo);
- List<RowData> rows = readRowDatas(tableInfo);
- Assertions.assertEquals(5, rows.size());
+ List<RowData> rows = readRowDatas(tableInfo);
+ Assertions.assertEquals(5, rows.size());
- cleanDatabase(tableInfo);
+ } finally {
+ cleanDatabase(tableInfo);
+ }
}
@Test
void testInputFormatWithProjection() throws Exception {
- KuduTableInfo tableInfo = booksTableInfo("books", true);
- setUpDatabase(tableInfo);
+ KuduTableInfo tableInfo = uniqueBooksTableInfo(true);
+ try {
+ setUpDatabase(tableInfo);
- List<RowData> rows = readRowDatas(tableInfo, "title", "id");
- Assertions.assertEquals(5, rows.size());
+ List<RowData> rows = readRowDatas(tableInfo, "title", "id");
+ Assertions.assertEquals(5, rows.size());
- for (RowData row : rows) {
- Assertions.assertEquals(2, row.getArity());
- }
+ for (RowData row : rows) {
+ Assertions.assertEquals(2, row.getArity());
+ }
- cleanDatabase(tableInfo);
+ } finally {
+ cleanDatabase(tableInfo);
+ }
}
private List<RowData> readRowDatas(KuduTableInfo tableInfo, String...
fieldProjection)
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/format/KuduRowInputFormatTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/format/KuduRowInputFormatTest.java
index c5a9f46..50f4cfe 100644
---
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/format/KuduRowInputFormatTest.java
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/format/KuduRowInputFormatTest.java
@@ -35,7 +35,7 @@ class KuduRowInputFormatTest extends KuduTestBase {
@Test
void testInvalidKuduMaster() {
- KuduTableInfo tableInfo = booksTableInfo("books", false);
+ KuduTableInfo tableInfo = uniqueBooksTableInfo(false);
Assertions.assertThrows(
NullPointerException.class,
() -> new KuduRowInputFormat(null, new
RowResultRowConverter(), tableInfo));
@@ -53,28 +53,34 @@ class KuduRowInputFormatTest extends KuduTestBase {
@Test
void testInputFormat() throws Exception {
- KuduTableInfo tableInfo = booksTableInfo("books", true);
- setUpDatabase(tableInfo);
+ KuduTableInfo tableInfo = uniqueBooksTableInfo(true);
+ try {
+ setUpDatabase(tableInfo);
- List<Row> rows = readRows(tableInfo);
- Assertions.assertEquals(5, rows.size());
+ List<Row> rows = readRows(tableInfo);
+ Assertions.assertEquals(5, rows.size());
- cleanDatabase(tableInfo);
+ } finally {
+ cleanDatabase(tableInfo);
+ }
}
@Test
void testInputFormatWithProjection() throws Exception {
- KuduTableInfo tableInfo = booksTableInfo("books", true);
- setUpDatabase(tableInfo);
+ KuduTableInfo tableInfo = uniqueBooksTableInfo(true);
+ try {
+ setUpDatabase(tableInfo);
- List<Row> rows = readRows(tableInfo, "title", "id");
- Assertions.assertEquals(5, rows.size());
+ List<Row> rows = readRows(tableInfo, "title", "id");
+ Assertions.assertEquals(5, rows.size());
- for (Row row : rows) {
- Assertions.assertEquals(2, row.getArity());
- }
+ for (Row row : rows) {
+ Assertions.assertEquals(2, row.getArity());
+ }
- cleanDatabase(tableInfo);
+ } finally {
+ cleanDatabase(tableInfo);
+ }
}
private List<Row> readRows(KuduTableInfo tableInfo, String...
fieldProjection)
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduDynamicSinkTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduDynamicSinkTest.java
index 7020519..8382887 100644
---
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduDynamicSinkTest.java
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduDynamicSinkTest.java
@@ -33,13 +33,13 @@ import static
org.junit.jupiter.api.Assertions.assertNotNull;
/** Unit Tests for {@link
org.apache.flink.connector.kudu.table.KuduDynamicTableSink}. */
public class KuduDynamicSinkTest extends KuduTestBase {
- public static final String INPUT_TABLE = "books";
- public static StreamExecutionEnvironment env;
- public static TableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private TableEnvironment tEnv;
+ private KuduTableInfo tableInfo;
@BeforeEach
public void init() {
- KuduTableInfo tableInfo = booksTableInfo(INPUT_TABLE, true);
+ tableInfo = uniqueBooksTableInfo(true);
setUpDatabase(tableInfo);
env = StreamExecutionEnvironment.getExecutionEnvironment();
tEnv = StreamTableEnvironment.create(env);
@@ -47,15 +47,17 @@ public class KuduDynamicSinkTest extends KuduTestBase {
@AfterEach
public void clean() {
- KuduTableInfo tableInfo = booksTableInfo(INPUT_TABLE, true);
- cleanDatabase(tableInfo);
+ if (tableInfo != null) {
+ cleanDatabase(tableInfo);
+ tableInfo = null;
+ }
}
@Test
public void testKuduSink() {
String createSql =
"CREATE TABLE "
- + INPUT_TABLE
+ + tableInfo.getName()
+ "("
+ "id int PRIMARY KEY NOT ENFORCED,"
+ "title string,"
@@ -68,7 +70,7 @@ public class KuduDynamicSinkTest extends KuduTestBase {
+ 123245
+ "',"
+ " 'table-name'='"
- + INPUT_TABLE
+ + tableInfo.getName()
+ "','sink.max-buffer-size'='1024"
+ "','sink.flush-interval'='1000ms"
+ "','sink.operation-timeout'='500ms"
@@ -77,9 +79,12 @@ public class KuduDynamicSinkTest extends KuduTestBase {
+ ")";
tEnv.executeSql(createSql);
tEnv.executeSql(
- "insert into " + INPUT_TABLE + " values(1006,'test
title','test author',10.1,10)");
+ "insert into "
+ + tableInfo.getName()
+ + " values(1006,'test title','test author',10.1,10)");
CloseableIterator<Row> collected =
- tEnv.executeSql("select * from " + INPUT_TABLE + " where id
=1006").collect();
+ tEnv.executeSql("select * from " + tableInfo.getName() + "
where id =1006")
+ .collect();
assertNotNull(collected);
}
}
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduDynamicSourceTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduDynamicSourceTest.java
index d7fbfef..19e1025 100644
---
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduDynamicSourceTest.java
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduDynamicSourceTest.java
@@ -39,13 +39,13 @@ import static
org.junit.jupiter.api.Assertions.assertNotNull;
/** Unit Tests for {@link
org.apache.flink.connector.kudu.table.KuduDynamicTableSource}. */
public class KuduDynamicSourceTest extends KuduTestBase {
- public static final String INPUT_TABLE = "books";
- public static StreamExecutionEnvironment env;
- public static TableEnvironment tEnv;
+ private StreamExecutionEnvironment env;
+ private TableEnvironment tEnv;
+ private KuduTableInfo tableInfo;
@BeforeEach
public void init() {
- KuduTableInfo tableInfo = booksTableInfo(INPUT_TABLE, true);
+ tableInfo = uniqueBooksTableInfo(true);
setUpDatabase(tableInfo);
env = StreamExecutionEnvironment.getExecutionEnvironment();
tEnv = StreamTableEnvironment.create(env);
@@ -53,8 +53,10 @@ public class KuduDynamicSourceTest extends KuduTestBase {
@AfterEach
public void clean() {
- KuduTableInfo tableInfo = booksTableInfo(INPUT_TABLE, true);
- cleanDatabase(tableInfo);
+ if (tableInfo != null) {
+ cleanDatabase(tableInfo);
+ tableInfo = null;
+ }
}
@Test
@@ -62,7 +64,7 @@ public class KuduDynamicSourceTest extends KuduTestBase {
// "id", "title", "author", "price", "quantity"
tEnv.executeSql(
"CREATE TABLE "
- + INPUT_TABLE
+ + tableInfo.getName()
+ "("
+ "id int PRIMARY KEY NOT ENFORCED,"
+ "title string,"
@@ -75,12 +77,12 @@ public class KuduDynamicSourceTest extends KuduTestBase {
+ getMasterAddress()
+ "',"
+ " 'table-name'='"
- + INPUT_TABLE
+ + tableInfo.getName()
+ "',"
+ "'scan.row-size'='10'"
+ ")");
- Iterator<Row> collected = tEnv.executeSql("SELECT * FROM " +
INPUT_TABLE).collect();
+ Iterator<Row> collected = tEnv.executeSql("SELECT * FROM " +
tableInfo.getName()).collect();
assertNotNull(collected);
}
@@ -88,7 +90,7 @@ public class KuduDynamicSourceTest extends KuduTestBase {
public void testProject() throws Exception {
tEnv.executeSql(
"CREATE TABLE "
- + INPUT_TABLE
+ + tableInfo.getName()
+ "("
+ "id int PRIMARY KEY NOT ENFORCED,"
+ "title string,"
@@ -101,13 +103,13 @@ public class KuduDynamicSourceTest extends KuduTestBase {
+ getMasterAddress()
+ "',"
+ " 'table-name'='"
- + INPUT_TABLE
+ + tableInfo.getName()
+ "',"
+ "'scan.row-size'='10'"
+ ")");
Iterator<Row> collected =
- tEnv.executeSql("SELECT id,title,author FROM " +
INPUT_TABLE).collect();
+ tEnv.executeSql("SELECT id,title,author FROM " +
tableInfo.getName()).collect();
assertNotNull(collected);
List<String> result =
CollectionUtil.iteratorToList(collected).stream()
@@ -130,7 +132,7 @@ public class KuduDynamicSourceTest extends KuduTestBase {
public void testLimit() throws Exception {
tEnv.executeSql(
"CREATE TABLE "
- + INPUT_TABLE
+ + tableInfo.getName()
+ "("
+ "id int PRIMARY KEY NOT ENFORCED,"
+ "title string,"
@@ -143,13 +145,13 @@ public class KuduDynamicSourceTest extends KuduTestBase {
+ getMasterAddress()
+ "',"
+ " 'table-name'='"
- + INPUT_TABLE
+ + tableInfo.getName()
+ "',"
+ "'scan.row-size'='10'"
+ ")");
Iterator<Row> collected =
- tEnv.executeSql("SELECT * FROM " + INPUT_TABLE + " LIMIT
1").collect();
+ tEnv.executeSql("SELECT * FROM " + tableInfo.getName() + "
LIMIT 1").collect();
List<String> result =
CollectionUtil.iteratorToList(collected).stream()
.map(Row::toString)
@@ -162,7 +164,7 @@ public class KuduDynamicSourceTest extends KuduTestBase {
public void testLookupJoin() {
tEnv.executeSql(
"CREATE TABLE "
- + INPUT_TABLE
+ + tableInfo.getName()
+ "("
+ "id int PRIMARY KEY NOT ENFORCED,"
+ "title string,"
@@ -175,7 +177,7 @@ public class KuduDynamicSourceTest extends KuduTestBase {
+ getMasterAddress()
+ "',"
+ " 'table-name'='"
- + INPUT_TABLE
+ + tableInfo.getName()
+ "',"
+ "'scan.row-size'='10'"
+ ")");
@@ -201,7 +203,7 @@ public class KuduDynamicSourceTest extends KuduTestBase {
tEnv.executeSql(
"SELECT d.id, isbn, title FROM datagen as d"
+ " JOIN "
- + INPUT_TABLE
+ + tableInfo.getName()
+ " FOR SYSTEM_TIME AS OF d.proctime
AS k"
+ " ON d.id=k.id"
+ " WHERE k.title='Java for dummies'")
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduRowDataLookupFunctionTest.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduRowDataLookupFunctionTest.java
index c93391c..50e942a 100644
---
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduRowDataLookupFunctionTest.java
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduRowDataLookupFunctionTest.java
@@ -37,19 +37,20 @@ import static
org.junit.jupiter.api.Assertions.assertNotNull;
/** Unit Tests for {@link KuduRowDataLookupFunction}. */
public class KuduRowDataLookupFunctionTest extends KuduTestBase {
- public static final String INPUT_TABLE = "books";
- public static KuduTableInfo tableInfo;
+ private KuduTableInfo tableInfo;
@BeforeEach
public void init() {
- tableInfo = booksTableInfo(INPUT_TABLE, true);
+ tableInfo = uniqueBooksTableInfo(true);
setUpDatabase(tableInfo);
}
@AfterEach
public void clean() {
- KuduTableInfo tableInfo = booksTableInfo(INPUT_TABLE, true);
- cleanDatabase(tableInfo);
+ if (tableInfo != null) {
+ cleanDatabase(tableInfo);
+ tableInfo = null;
+ }
}
@Test
diff --git
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduTableSourceITCase.java
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduTableSourceITCase.java
index 87c8828..c0758f4 100644
---
a/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduTableSourceITCase.java
+++
b/flink-connector-kudu/src/test/java/org/apache/flink/connector/kudu/table/KuduTableSourceITCase.java
@@ -42,7 +42,7 @@ public class KuduTableSourceITCase extends KuduTestBase {
@BeforeEach
void init() {
- tableInfo = booksTableInfo("books", true);
+ tableInfo = uniqueBooksTableInfo(true);
setUpDatabase(tableInfo);
tableEnv = KuduTableTestUtils.createTableEnvInBatchMode();
catalog = new KuduCatalog(getMasterAddress());
@@ -61,13 +61,13 @@ public class KuduTableSourceITCase extends KuduTestBase {
@Test
void testFullBatchScan() throws Exception {
CloseableIterator<Row> it =
- tableEnv.executeSql("select * from books order by
id").collect();
+ tableEnv.executeSql("select * from " + tableInfo.getName() + "
order by id")
+ .collect();
List<Row> results = new ArrayList<>();
it.forEachRemaining(results::add);
assertEquals(5, results.size());
assertEquals(
"+I[1001, Java for dummies, Tan Ah Teck, 11.11, 11]",
results.get(0).toString());
- tableEnv.executeSql("DROP TABLE books");
}
@Test
@@ -75,13 +75,14 @@ public class KuduTableSourceITCase extends KuduTestBase {
// (price > 30 and price < 40)
CloseableIterator<Row> it =
tableEnv.executeSql(
- "SELECT title FROM books WHERE id IN (1003,
1004) and "
+ "SELECT title FROM "
+ + tableInfo.getName()
+ + " WHERE id IN (1003, 1004) and "
+ "quantity < 40")
.collect();
List<Row> results = new ArrayList<>();
it.forEachRemaining(results::add);
assertEquals(1, results.size());
assertEquals("+I[More Java for more dummies]",
results.get(0).toString());
- tableEnv.executeSql("DROP TABLE books");
}
}
