You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2022/09/22 06:14:10 UTC
[flink-table-store] branch release-0.2 updated: [hotfix] Fix unstable FileStoreTableTestBase.testPartitionEmptyWriter
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.2
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/release-0.2 by this push:
new a7efc6ba [hotfix] Fix unstable FileStoreTableTestBase.testPartitionEmptyWriter
a7efc6ba is described below
commit a7efc6bafe5f1bfb25eb00de1715945c8d83f501
Author: JingsongLi <lz...@aliyun.com>
AuthorDate: Thu Sep 22 14:11:43 2022 +0800
[hotfix] Fix unstable FileStoreTableTestBase.testPartitionEmptyWriter
---
.../table/store/table/FileStoreTableTestBase.java | 40 +++++++++++++++-------
1 file changed, 27 insertions(+), 13 deletions(-)
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
index 19b7a836..27a06133 100644
--- a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/FileStoreTableTestBase.java
@@ -206,19 +206,33 @@ public abstract class FileStoreTableTestBase {
List<FileCommittable> commit4 = write.prepareCommit(false);
// trigger compaction, but not wait it.
- write.write(rowData(2, 20, 200L));
- List<FileCommittable> commit5 = write.prepareCommit(true);
- // wait compaction finish
- // commit5 should be a compaction commit
-
- write.write(rowData(1, 60, 600L));
- List<FileCommittable> commit6 = write.prepareCommit(true);
- // if remove writer too fast, will see old files, do another compaction
- // then will be conflicts
-
- commit.commit("4", commit4);
- commit.commit("5", commit5);
- commit.commit("6", commit6);
+ if (commit4.get(0).increment().compactBefore().isEmpty()) {
+ // commit4 is not a compaction commit
+ // do compaction commit5 and compaction commit6
+ write.write(rowData(2, 20, 200L));
+ List<FileCommittable> commit5 = write.prepareCommit(true);
+ // wait compaction finish
+ // commit5 should be a compaction commit
+
+ write.write(rowData(1, 60, 600L));
+ List<FileCommittable> commit6 = write.prepareCommit(true);
+ // if remove writer too fast, will see old files, do another compaction
+ // then will be conflicts
+
+ commit.commit("4", commit4);
+ commit.commit("5", commit5);
+ commit.commit("6", commit6);
+ } else {
+ // commit4 is a compaction commit
+ // do compaction commit5
+ write.write(rowData(2, 20, 200L));
+ List<FileCommittable> commit5 = write.prepareCommit(true);
+ // wait compaction finish
+ // commit5 should be a compaction commit
+
+ commit.commit("4", commit4);
+ commit.commit("5", commit5);
+ }
write.close();
}