You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "kaori-seasons (via GitHub)" <gi...@apache.org> on 2025/07/02 18:42:40 UTC
Re: [I] [test] testReadKvTableWithEarliestAndTimestampScanStartupMode failed [fluss]
kaori-seasons commented on issue #1194:
URL: https://github.com/apache/fluss/issues/1194#issuecomment-3028972546
看起来， 这个错误发生在 Flink118TableSourceITCase.testReadKvTableWithEarliestAndTimestampScanStartupMode 测试中，具体的失败点是在等待表分区创建时超时。
## 错误原因
- 分区创建超时：测试在第551行调用 waitUntilPartitions 方法等待1个表分区被创建 FlinkTableSourceITCase.java:551
- 等待机制：waitUntilPartitions 方法内部使用 waitValue 来等待分区创建完成，当超时后会抛出 AssertionError FlinkTestBase.java:230
- 超时控制：CommonTestUtils.waitUtil 方法实现了超时逻辑，当条件在指定时间内未满足时会失败 CommonTestUtils.java:81
## 重构方案
基于对测试失败原因的分析，我建议从以下几个方面重构这个测试：
1. 增强分区等待逻辑
当前的测试在第551行直接调用 waitUntilPartitions 方法 FlinkTableSourceITCase.java:551 ，但缺乏足够的容错机制。
```
@ParameterizedTest
@MethodSource("readKvTableScanStartupModeArgs")
void testReadKvTableWithEarliestAndTimestampScanStartupMode(String mode, boolean isPartitioned)
throws Exception {
long timestamp = CLOCK.milliseconds();
String tableName = mode + "_test_" + (isPartitioned ? "partitioned" : "non_partitioned");
TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
String partitionName = null;
if (!isPartitioned) {
tEnv.executeSql(
String.format(
"create table %s (a int not null primary key not enforced, b varchar)",
tableName));
} else {
tEnv.executeSql(
String.format(
"create table %s (a int not null, b varchar, c varchar, primary key (a, c) NOT ENFORCED) partitioned by (c) "
+ "with ("
+ " 'table.auto-partition.enabled' = 'true',"
+ " 'table.auto-partition.time-unit' = 'year',"
+ " 'table.auto-partition.num-precreate' = '1')",
tableName));
// 增强的分区等待逻辑，包含重试和详细日志
Map<Long, String> partitionNameById = waitForPartitionsWithRetry(tablePath, 1, 3);
partitionName = partitionNameById.values().iterator().next();
}
......................................
}
/**
* 带重试机制的分区等待方法
*/
private Map<Long, String> waitForPartitionsWithRetry(TablePath tablePath, int expectedCount, int maxRetries)
throws Exception {
Exception lastException = null;
for (int attempt = 1; attempt <= maxRetries; attempt++) {
try {
LOG.info("Waiting for {} partitions for table {} (attempt {}/{})",
expectedCount, tablePath, attempt, maxRetries);
Map<Long, String> partitions = waitUntilPartitions(
FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(),
tablePath,
expectedCount);
LOG.info("Successfully found {} partitions for table {}: {}",
partitions.size(), tablePath, partitions);
return partitions;
} catch (Exception e) {
lastException = e;
LOG.warn("Attempt {}/{} failed to wait for partitions: {}",
attempt, maxRetries, e.getMessage());
if (attempt < maxRetries) {
// 在重试前等待一段时间，让系统有时间创建分区
Thread.sleep(Duration.ofSeconds(2).toMillis());
}
}
}
throw new AssertionError(
String.format("Failed to create %d partitions for table %s after %d attempts",
expectedCount, tablePath, maxRetries),
lastException);
}
```
2. 优化超时配置
当前的 waitUntilPartitions 方法使用1分钟超时 FlinkTestBase.java:230 。应该创建一个专门的方法来处理测试环境中的超时：
```
/**
* 为测试环境优化的分区等待方法，使用更长的超时时间
*/
public static Map<Long, String> waitUntilPartitionsForTest(
ZooKeeperClient zooKeeperClient, TablePath tablePath, int expectPartitions) {
return waitValue(
() -> {
Map<Long, String> gotPartitions =
zooKeeperClient.getPartitionIdAndNames(tablePath);
LOG.debug("Current partitions for {}: {} (expecting {})",
tablePath, gotPartitions.size(), expectPartitions);
return expectPartitions == gotPartitions.size()
? Optional.of(gotPartitions)
: Optional.empty();
},
Duration.ofMinutes(3), // 增加到3分钟超时
String.format("expect %d table partition has not been created for table %s",
expectPartitions, tablePath));
}
```
3. 添加环境检查
在测试开始前添加环境健康检查：
```
@BeforeEach
void setUp() throws Exception {
// 现有的设置代码...
// 添加环境健康检查
verifyTestEnvironment();
}
private void verifyTestEnvironment() throws Exception {
// 检查ZooKeeper连接
assertThat(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient().isConnected())
.as("ZooKeeper should be connected")
.isTrue();
// 检查集群状态
try (Admin admin = conn.getAdmin()) {
// 简单的健康检查 - 尝试列出数据库
admin.listDatabases().get(Duration.ofSeconds(10).toMillis(), TimeUnit.MILLISECONDS);
}
LOG.info("Test environment verification passed");
}
```
4. 改进错误处理和日志
参考现有测试中的模式 FlinkTableSourceITCase.java:528-608 ，添加更详细的日志和错误处理：
```
private void logTestProgress(String phase, Object... args) {
LOG.info("[{}] {}",
Thread.currentThread().getName(),
String.format(phase, args));
}
// 在测试的关键点添加日志
logTestProgress("Starting test with mode=%s, isPartitioned=%s", mode, isPartitioned);
logTestProgress("Table created: %s", tableName);
logTestProgress("Writing first batch of rows: %d", rows1.size());
```
## 重构后的测试
```
@ParameterizedTest
@MethodSource("readKvTableScanStartupModeArgs")
@Timeout(value = 5, unit = TimeUnit.MINUTES) // 添加测试级别超时
void testReadKvTableWithEarliestAndTimestampScanStartupMode(String mode, boolean isPartitioned)
throws Exception {
logTestProgress("Starting test with mode=%s, isPartitioned=%s", mode, isPartitioned);
long timestamp = CLOCK.milliseconds();
String tableName = mode + "_test_" + (isPartitioned ? "partitioned" : "non_partitioned");
TablePath tablePath = TablePath.of(DEFAULT_DB, tableName);
String partitionName = null;
try {
if (!isPartitioned) {
tEnv.executeSql(
String.format(
"create table %s (a int not null primary key not enforced, b varchar)",
tableName));
logTestProgress("Created non-partitioned table: %s", tableName);
} else {
tEnv.executeSql(
String.format(
"create table %s (a int not null, b varchar, c varchar, primary key (a, c) NOT ENFORCED) partitioned by (c) "
+ "with ("
+ " 'table.auto-partition.enabled' = 'true',"
+ " 'table.auto-partition.time-unit' = 'year',"
+ " 'table.auto-partition.num-precreate' = '1')",
tableName));
logTestProgress("Created partitioned table: %s", tableName);
Map<Long, String> partitionNameById = waitForPartitionsWithRetry(tablePath, 1, 3);
partitionName = partitionNameById.values().iterator().next();
logTestProgress("Found partition: %s", partitionName);
}
// 其余测试逻辑保持与原始实现相同...
List<InternalRow> rows1 = Arrays.asList(
rowWithPartition(new Object[] {1, "v1"}, partitionName),
rowWithPartition(new Object[] {2, "v2"}, partitionName),
rowWithPartition(new Object[] {3, "v3"}, partitionName),
rowWithPartition(new Object[] {3, "v33"}, partitionName));
logTestProgress("Writing first batch of %d rows", rows1.size());
writeRows(conn, tablePath, rows1, false);
if (partitionName == null) {
waitUtilAllBucketFinishSnapshot(admin, tablePath);
} else {
waitUtilAllBucketFinishSnapshot(admin, tablePath, Collections.singleton(partitionName));
}
logTestProgress("First batch snapshot completed");
CLOCK.advanceTime(Duration.ofMillis(100));
List<InternalRow> rows2 = Arrays.asList(
rowWithPartition(new Object[] {1, "v11"}, partitionName),
rowWithPartition(new Object[] {2, "v22"}, partitionName),
rowWithPartition(new Object[] {4, "v4"}, partitionName));
logTestProgress("Writing second batch of %d rows", rows2.size());
writeRows(conn, tablePath, rows2, false);
CLOCK.advanceTime(Duration.ofMillis(100));
String options = String.format(
" /*+ OPTIONS('scan.startup.mode' = '%s', 'scan.startup.timestamp' = '%s') */",
mode, timestamp);
String query = "select a, b from " + tableName + options;
List<String> expected = Arrays.asList(
"+I[1, v1]", "+I[2, v2]", "+I[3, v3]", "-U[3, v3]", "+U[3, v33]",
"-U[1, v1]", "+U[1, v11]", "-U[2, v2]", "+U[2, v22]", "+I[4, v4]");
logTestProgress("Executing query and collecting results");
try (CloseableIterator<Row> rowIter = tEnv.executeSql(query).collect()) {
int expectRecords = 10;
List<String> actual = new ArrayList<>(expectRecords);
for (int i = 0; i < expectRecords; i++) {
Row r = rowIter.next();
String row = r.toString();
actual.add(row);
}
assertThat(actual).containsExactlyElementsOf(expected);
logTestProgress("Test completed successfully");
}
} catch (Exception e) {
LOG.error("Test failed for mode={}, isPartitioned={}", mode, isPartitioned, e);
throw e;
}
}
```
--
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-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org