[flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests

[nhuantho]

Purpose

Linked issue: close #2471

Brief change log

• Overloading the collectRowsWithTimeout method allows passing maxWaitTime.
• Add a test for covering scan.startup.mode='latest'.

Tests

• fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/testutils/FlinkRowAssertionsUtils.java
  • Overloading the collectRowsWithTimeout method allows passing maxWaitTime.
• fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java
  • testChangelogWithScanStartupMode -> add logic for testing scan.startup.mode='latest'

[Copilot]

Pull request overview

This PR adds test coverage for the scan.startup.mode='latest' configuration option in Flink's changelog virtual table integration tests. The change ensures that the latest mode correctly reads only new records after subscription, filling a gap in the existing test suite.

Changes:

• Added an overloaded collectRowsWithTimeout method that accepts a custom maxWaitTime parameter
• Extended testChangelogWithScanStartupMode to include a test case for scan.startup.mode='latest'

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/testutils/FlinkRowAssertionsUtils.java	Adds overloaded method to support custom timeout duration when collecting rows
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java	Implements test logic for verifying latest scan startup mode behavior

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The regex pattern assumes a specific row format. If latestResults.getFirst() doesn't match the expected pattern, parseInt will throw a NumberFormatException. Consider adding validation or using a more robust parsing approach to handle unexpected formats gracefully.

[Copilot]

The hardcoded 5-second timeout per attempt combined with 10 attempts could result in up to 50 seconds of waiting. Consider reducing the timeout or number of attempts, or explaining why this duration is necessary to avoid unnecessarily long test execution times.

Suggested change

	for (int attempt = 0; attempt < 10; attempt++) {
	// Write a new record (with id larger than 5)
	int rowId = 6 + attempt;
	writeRows(conn, tablePath, Arrays.asList(row(rowId, "v" + rowId)), false);
	// Try to fetch one record with a 5-second timeout
	latestResults = collectRowsWithTimeout(rowIterLatest, 1, Duration.ofSeconds(5));
	for (int attempt = 0; attempt < 5; attempt++) {
	// Write a new record (with id larger than 5)
	int rowId = 6 + attempt;
	writeRows(conn, tablePath, Arrays.asList(row(rowId, "v" + rowId)), false);
	// Try to fetch one record with a 1-second timeout
	latestResults = collectRowsWithTimeout(rowIterLatest, 1, Duration.ofSeconds(1));

[nhuantho]

This suggestion is not correct:

[MehulBatra]

Can you please ellaborate your concern?

[nhuantho]

@MehulBatra, from my side, Copilot was suggested incorrectly:

• Proposal:
  • Fetch timeout: 5s -> Copilot: latestResults = collectRowsWithTimeout(rowIterLatest, 1, Duration.ofSeconds(1));
  • Repeat step 1 for 10 times -> Copilot: for (int attempt = 0; attempt < 5; attempt++)

[nhuantho]

@MehulBatra, can you please review the pull request again?

[MehulBatra]

@MehulBatra, can you please review the pull request again?

Thank you for your PR I will try to review it within this week!