This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new d90a9563e [lake/hudi] Add Flink union read test base for Hudi (#3522)
d90a9563e is described below
commit d90a9563e3e67b9b0734d79025e5090474c53229
Author: fhan <[email protected]>
AuthorDate: Thu Jun 25 17:18:42 2026 +0800
[lake/hudi] Add Flink union read test base for Hudi (#3522)
---
.../lake/hudi/flink/FlinkUnionReadTestBase.java | 97 ++++++++++++++++++++++
.../hudi/testutils/FlinkHudiTieringTestBase.java | 34 ++++++++
2 files changed, 131 insertions(+)
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/flink/FlinkUnionReadTestBase.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/flink/FlinkUnionReadTestBase.java
new file mode 100644
index 000000000..243c65290
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/flink/FlinkUnionReadTestBase.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.lake.hudi.flink;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.lake.hudi.testutils.FlinkHudiTieringTestBase;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import javax.annotation.Nullable;
+
+import static org.apache.fluss.flink.FlinkConnectorOptions.BOOTSTRAP_SERVERS;
+
+/** Base class for Hudi union read tests. */
+public class FlinkUnionReadTestBase extends FlinkHudiTieringTestBase {
+
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder()
+ .setClusterConf(initConfig())
+ .setNumOfTabletServers(3)
+ .build();
+
+ protected static final int DEFAULT_BUCKET_NUM = 1;
+ StreamTableEnvironment batchTEnv;
+ StreamTableEnvironment streamTEnv;
+
+ @BeforeAll
+ protected static void beforeAll() {
+
FlinkHudiTieringTestBase.beforeAll(FLUSS_CLUSTER_EXTENSION.getClientConfig());
+ }
+
+ @BeforeEach
+ public void beforeEach() {
+ super.beforeEach();
+ buildBatchTEnv();
+ buildStreamTEnv();
+ }
+
+ @Override
+ protected FlussClusterExtension getFlussClusterExtension() {
+ return FLUSS_CLUSTER_EXTENSION;
+ }
+
+ protected StreamTableEnvironment buildStreamTEnv(@Nullable String
savepointPath) {
+ Configuration conf = new Configuration();
+ if (savepointPath != null) {
+ conf.setString("execution.savepoint.path", savepointPath);
+ execEnv.configure(conf);
+ }
+ String bootstrapServers = String.join(",",
clientConf.get(ConfigOptions.BOOTSTRAP_SERVERS));
+ streamTEnv = StreamTableEnvironment.create(execEnv,
EnvironmentSettings.inStreamingMode());
+ streamTEnv.executeSql(
+ String.format(
+ "create catalog %s with ('type' = 'fluss', '%s' =
'%s')",
+ CATALOG_NAME, BOOTSTRAP_SERVERS.key(),
bootstrapServers));
+ streamTEnv.executeSql("use catalog " + CATALOG_NAME);
+ streamTEnv.executeSql("use " + DEFAULT_DB);
+ return streamTEnv;
+ }
+
+ private void buildStreamTEnv() {
+ buildStreamTEnv(null);
+ }
+
+ public void buildBatchTEnv() {
+ String bootstrapServers = String.join(",",
clientConf.get(ConfigOptions.BOOTSTRAP_SERVERS));
+ batchTEnv = StreamTableEnvironment.create(execEnv,
EnvironmentSettings.inBatchMode());
+ batchTEnv.executeSql(
+ String.format(
+ "create catalog %s with ('type' = 'fluss', '%s' =
'%s')",
+ CATALOG_NAME, BOOTSTRAP_SERVERS.key(),
bootstrapServers));
+ batchTEnv.executeSql("use catalog " + CATALOG_NAME);
+ batchTEnv.executeSql("use " + DEFAULT_DB);
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java
index f678c3766..40de1ed42 100644
---
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/testutils/FlinkHudiTieringTestBase.java
@@ -72,10 +72,12 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -321,6 +323,14 @@ public abstract class FlinkHudiTieringTestBase {
return getFlussClusterExtension().waitAndGetLeaderReplica(tableBucket);
}
+ protected void assertReplicaStatus(Map<TableBucket, Long>
expectedLogEndOffset) {
+ for (Map.Entry<TableBucket, Long> expectedLogEndOffsetEntry :
+ expectedLogEndOffset.entrySet()) {
+ assertReplicaStatus(
+ expectedLogEndOffsetEntry.getKey(),
expectedLogEndOffsetEntry.getValue());
+ }
+ }
+
protected void assertReplicaStatus(TableBucket tb, long
expectedLogEndOffset) {
retry(
Duration.ofMinutes(1),
@@ -332,6 +342,30 @@ public abstract class FlinkHudiTieringTestBase {
});
}
+ protected void waitUntilBucketSynced(
+ TablePath tablePath, long tableId, int bucketCount, boolean
isPartitioned) {
+ Set<TableBucket> tableBuckets = new HashSet<>();
+ if (isPartitioned) {
+ Map<Long, String> partitionById = waitUntilPartitions(tablePath);
+ for (Long partitionId : partitionById.keySet()) {
+ for (int i = 0; i < bucketCount; i++) {
+ tableBuckets.add(new TableBucket(tableId, partitionId, i));
+ }
+ }
+ } else {
+ for (int i = 0; i < bucketCount; i++) {
+ tableBuckets.add(new TableBucket(tableId, i));
+ }
+ }
+ waitUntilBucketsSynced(tableBuckets);
+ }
+
+ protected void waitUntilBucketsSynced(Set<TableBucket> tableBuckets) {
+ for (TableBucket tableBucket : tableBuckets) {
+ waitUntilBucketSynced(tableBucket);
+ }
+ }
+
protected void waitUntilBucketSynced(TableBucket tb) {
waitUntil(
() ->
getLeaderReplica(tb).getLogTablet().getLakeTableSnapshotId() >= 0,
