lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-4053] Flaky ITTestHoodieDataSource.testStreamWriteBatchReadOpti… (#5526)

Browse Source 
* [HUDI-4053] Flaky ITTestHoodieDataSource.testStreamWriteBatchReadOptimized

Co-authored-by: xicm <xicm@asiainfo.com>
This commit is contained in:
xicm
2022-05-09 16:35:50 +08:00
committed by GitHub
parent 4c70840275
commit 6b47ef6ed2
2 changed files with 21 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
View File

@@ -240,7 +240,15 @@ public class ITTestHoodieDataSource extends AbstractTestBase {
List<Row> rows = CollectionUtil.iterableToList( List<Row> rows = CollectionUtil.iterableToList(
() -> streamTableEnv.sqlQuery("select * from t1").execute().collect()); () -> streamTableEnv.sqlQuery("select * from t1").execute().collect());
assertRowsEquals(rows, TestData.DATA_SET_SOURCE_INSERT);
// the test is flaky based on whether the first compaction is pending when
// scheduling the 2nd compaction.
// see details in CompactionPlanOperator#scheduleCompaction.
if (rows.size() < TestData.DATA_SET_SOURCE_INSERT.size()) {
assertRowsEquals(rows, TestData.DATA_SET_SOURCE_INSERT_FIRST_COMMIT);
} else {
assertRowsEquals(rows, TestData.DATA_SET_SOURCE_INSERT);
}
} }
@Test @Test

12
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestData.java
View File

@@ -164,6 +164,18 @@ public class TestData {
TimestampData.fromEpochMillis(8000), StringData.fromString("par4")) TimestampData.fromEpochMillis(8000), StringData.fromString("par4"))
); );
// data set of test_source.data first commit.
public static List<RowData> DATA_SET_SOURCE_INSERT_FIRST_COMMIT = Arrays.asList(
insertRow(StringData.fromString("id1"), StringData.fromString("Danny"), 23,
TimestampData.fromEpochMillis(1000), StringData.fromString("par1")),
insertRow(StringData.fromString("id2"), StringData.fromString("Stephen"), 33,
TimestampData.fromEpochMillis(2000), StringData.fromString("par1")),
insertRow(StringData.fromString("id3"), StringData.fromString("Julian"), 53,
TimestampData.fromEpochMillis(3000), StringData.fromString("par2")),
insertRow(StringData.fromString("id4"), StringData.fromString("Fabian"), 31,
TimestampData.fromEpochMillis(4000), StringData.fromString("par2"))
);
// data set of test_source.data latest commit. // data set of test_source.data latest commit.
public static List<RowData> DATA_SET_SOURCE_INSERT_LATEST_COMMIT = Arrays.asList( public static List<RowData> DATA_SET_SOURCE_INSERT_LATEST_COMMIT = Arrays.asList(
insertRow(StringData.fromString("id5"), StringData.fromString("Sophia"), 18, insertRow(StringData.fromString("id5"), StringData.fromString("Sophia"), 18,