wuchong commented on code in PR #2467:
URL: https://github.com/apache/fluss/pull/2467#discussion_r2724324458
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -355,12 +370,19 @@ public void testChangelogWithScanStartupMode() throws
Exception {
"SELECT _change_type, id, name FROM
startup_mode_test$changelog" + optionsLatest;
CloseableIterator<Row> rowIterLatest =
tEnv.executeSql(queryLatest).collect();
- // Write new data after subscribing with 'latest'
- CLOCK.advanceTime(Duration.ofMillis(100));
- writeRows(conn, tablePath, Arrays.asList(row(6, "v6")), false);
+ // Write multiple records with delays to ensure capture after 'latest'
subscription is
+ // ready.
+ for (int i = 6; i <= 8; i++) {
+ Thread.sleep(1000);
+ CLOCK.advanceTime(Duration.ofMillis(100));
+ writeRows(conn, tablePath, Arrays.asList(row(i, "v" + i)), false);
+ }
+
+ // Collect results - we expect at least 1 record (the ones written
after subscription)
List<String> latestResults = collectRowsWithTimeout(rowIterLatest, 1,
true);
assertThat(latestResults).hasSize(1);
- assertThat(latestResults.get(0)).isEqualTo("+I[+I, 6, v6]");
+ // The record should be one of the records we wrote (v6 to v10)
Review Comment:
Personally, I’m not in favor of adding 3 × 4 seconds (3 more seconds for
each test and 4 flink versions matrix) of sleep to tests just to stabilize
flaky ones. Simply inserting sleeps doesn’t actually solve the underlying
issue—the enumerator might still be delayed due to CPU load during testing.
The root problem is that we currently lack a clear signal indicating that a
`SELECT` query has successfully started. Given this, I suggest removing this
test for now, especially since we also don’t cover the `latest` mode in our
`scan.startup.mode` configuration tests.
Alternatively, if we still want to cover this case, we can do like this:
1. Use `collectRowsWithTimeout` to fetch one record with a 5-second timeout.
2. If no record is received within the timeout, write a new record (with id
larger than 5) and repeat step 1 for 10 times.
3. Once a record is fetched, assert that it matches id larger than 5.
This strategy avoids excessive waiting in most cases while providing a
clearer success condition.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
