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

nhuantho · 2026-01-25T18:54:53Z

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 · 2026-01-29T07:51:55Z

+                break;
+            }
+        }
+        int id = Integer.parseInt(latestResults.getFirst().replaceAll(".*\\[(\\d+)]", "$1"));


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 · 2026-01-29T07:51:55Z

+        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));


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));

This suggestion is not correct:

Can you please ellaborate your concern?

@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++)

I’ve thought about this again, and a more elegant approach might be to leverage Flink’s savepoint mechanism to explicitly force the initialization of the start offset. Here’s the refined idea:

Pre-populate the source table with some existing records.

Launch a job:
INSERT INTO sink SELECT * FROM source /*+ OPTIONS('scan.startup.mode' = 'latest') */;

Stop the job with a savepoint (e.g., using FlinkOperator#stopWithSavepoint, as shown in this example).
→ At this point, the “latest” offset has been determined, initialized, and checkpointed into the state.

Restart the job from the savepoint. (see example in FlinkTableSourceFailOverITCase#initTableEnvironment)

Write new records into the source table.

Read from the sink table — the output should contain only the records written in step 5, confirming that the job correctly started reading from the offset captured at savepoint time.

Thanks for the suggestion! I’ll follow this approach and submit a PR accordingly.

nhuantho · 2026-01-30T10:03:31Z

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

MehulBatra · 2026-01-31T08:09:50Z

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

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

MehulBatra · 2026-02-09T13:37:09Z

@nhuantho can you resolve the conflicts? I will run the workflow and review

nhuantho · 2026-02-09T15:41:59Z

@MehulBatra I resolved conflicts

wuchong

Thanks @nhuantho , I left some comments.

wuchong · 2026-02-10T13:16:42Z

                .containsExactly(
                        "+I[insert, 3, 1970-01-01T00:00:00.200Z, 4, v4]",
                        "+I[insert, 4, 1970-01-01T00:00:00.200Z, 5, v5]");
+        // 3. Test scan.startup.mode='latest' - should only read new records after subscription


It will be more clear to extract this test into a separate test method.

wuchong · 2026-02-10T13:17:02Z

+        tEnv = initTableEnvironment(savepointPath);
+        // Recreate a table
+        tEnv.executeSql(
+                "CREATE TABLE startup_mode_test_mode_latest ("


we don't need to re-create the table, the table is already there.

wuchong · 2026-02-10T13:17:20Z

+
+        // Recreate a table
+        tEnv.executeSql(
+                "CREATE TABLE "


we don't need to re-create the table

wuchong · 2026-02-10T13:18:02Z

+                tEnv.executeSql(
+                        "INSERT INTO "
+                                + tablePath.getTableName()
+                                + " SELECT * FROM startup_mode_test");


Run the query with startup_mode_test$changelog with latest startup mode. So we can materilaize the latest offsets into state.

wuchong · 2026-02-10T13:18:22Z

+
+        String optionsLatest = " /*+ OPTIONS('scan.startup.mode' = 'latest') */";
+        String queryLatest =
+                "SELECT _change_type, id, name FROM "
+                        + tablePath.getTableName()
+                        + "$changelog"
+                        + optionsLatest;
+        CloseableIterator<Row> rowIterLatest = tEnv.executeSql(queryLatest).collect();


rerun the query of insertResult

wuchong · 2026-02-12T11:32:21Z

I tried to run locally, but it fails with following exception, not sure what's the reason

java.util.concurrent.ExecutionException: org.apache.flink.runtime.dispatcher.UnavailableDispatcherOperationException: Unable to get JobMasterGateway for initializing job. The requested operation is not available while the JobManager is initializing.

	at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
	at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2005)
	at org.apache.fluss.flink.source.ChangelogVirtualTableITCase.testChangelogWithLatestScanStartupMode(ChangelogVirtualTableITCase.java:444)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:727)
	at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
	at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:156)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:147)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:86)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(InterceptingExecutableInvoker.java:103)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.lambda$invoke$0(InterceptingExecutableInvoker.java:93)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:92)
	at org.junit.jupiter.engine.execution.InterceptingExecutableInvoker.invoke(InterceptingExecutableInvoker.java:86)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$7(TestMethodTestDescriptor.java:217)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:213)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:138)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:68)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:151)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
	at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
	at org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService$ExclusiveTask.compute(ForkJoinPoolHierarchicalTestExecutorService.java:185)
	at org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService.invokeAll(ForkJoinPoolHierarchicalTestExecutorService.java:129)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
	at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
	at org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService$ExclusiveTask.compute(ForkJoinPoolHierarchicalTestExecutorService.java:185)
	at org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService.invokeAll(ForkJoinPoolHierarchicalTestExecutorService.java:129)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
	at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
	at org.junit.platform.engine.support.hierarchical.ForkJoinPoolHierarchicalTestExecutorService$ExclusiveTask.compute(ForkJoinPoolHierarchicalTestExecutorService.java:185)
	at java.base/java.util.concurrent.RecursiveAction.exec(RecursiveAction.java:189)
	at java.base/java.util.concurrent.ForkJoinTask.doExec$$$capture(ForkJoinTask.java:290)
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java)
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)
Caused by: org.apache.flink.runtime.dispatcher.UnavailableDispatcherOperationException: Unable to get JobMasterGateway for initializing job. The requested operation is not available while the JobManager is initializing.
	at org.apache.flink.runtime.dispatcher.Dispatcher.getJobMasterGateway(Dispatcher.java:1503)
	at org.apache.flink.runtime.dispatcher.Dispatcher.performOperationOnJobMasterGateway(Dispatcher.java:1513)
	at org.apache.flink.runtime.dispatcher.Dispatcher.stopWithSavepointAndGetLocation(Dispatcher.java:1077)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:318)

…l table (apache#2477)

…ete checkpoint (apache#2743)

wuchong

I investigated the failing test case and traced it to an issue #2743. I submitted a commit to address the root cause and the test should pass now.

cc @MehulBatra could you help to review it when you are free, because it is related to a bug of $changelog and $binlog table.

…yAfterFailover

…l table (#2477)

MehulBatra

LGTM! just left one small comment

Thank you @wuchong & @nhuantho for the PR looks great to me.
Verified Latest mode doesn't read historical data for changelog and binlog

Latest mode state persists in savepoints
Restart from savepoint continues correctly
No duplicate processing after restart

Can we move waitForCheckpoint, getFileBasedCheckpointsConfig to a base class and share it instead of writing individually for both?

One concern @wuchong as we have moved to maintaining MiniClusterWithClientResource with before each instead of abstarcttestbase could this hamper the CI runtimes.

nhuantho · 2026-02-28T14:37:14Z

@MehulBatra if this idea is approved, I can create another issue and implement it.

wuchong · 2026-02-28T16:53:27Z

@MehulBatra, I agree with extracting common test methods.

I also share your concern regarding increased test execution time and support reusing the Flink mini-cluster as an optimization. Let's track this improvement in a follow-up PR. However, we cannot simply extend AbstractTestBase in this case, as we need to override specific checkpoint configurations.

MehulBatra · 2026-03-01T09:49:58Z

Hi @nhuantho I am already working on the shared base class for the IT, you can help me in review!

nhuantho · 2026-03-01T10:12:32Z

@MehulBatra sure! If there’s anything I can help with, feel free to tag me.

MehulBatra · 2026-03-01T10:14:56Z

@MehulBatra, I agree with extracting common test methods.

I also share your concern regarding increased test execution time and support reusing the Flink mini-cluster as an optimization. Let's track this improvement in a follow-up PR. However, we cannot simply extend AbstractTestBase in this case, as we need to override specific checkpoint configurations.

Sure @wuchong , we can discuss this more in the upcoming pr and improvements related to this, will kleep you posted on same.

nhuantho · 2026-03-01T17:55:18Z

@MehulBatra, I agree with extracting common test methods.

I also share your concern regarding increased test execution time and support reusing the Flink mini-cluster as an optimization. Let's track this improvement in a follow-up PR. However, we cannot simply extend AbstractTestBase in this case, as we need to override specific checkpoint configurations.

Hi @wuchong, @MehulBatra I have a question, Can we move MiniClusterWithClientResource to beforeAll, instead of beforeEach. From my side, it can reduce time to creating MiniCluster.
I tested with:

First
Create MiniCluster in beforeEach

Create MiniCluster in beforeAll

Second
Create MiniCluster in beforeEach

Create MiniCluster in beforeAll

If it is correct, I can create a pr.

…l table (apache#2477)

MehulBatra requested a review from Copilot January 29, 2026 07:51

Copilot AI reviewed Jan 29, 2026

View reviewed changes

luoyuxia force-pushed the main branch from 0e2d71f to 74ee888 Compare February 6, 2026 12:32

nhuantho force-pushed the flink-test/issue-2471 branch from 57489ce to 6fb31ee Compare February 9, 2026 15:40

nhuantho requested review from MehulBatra and wuchong February 9, 2026 15:42

wuchong reviewed Feb 10, 2026

View reviewed changes

nhuantho force-pushed the flink-test/issue-2471 branch 3 times, most recently from 98c7a4e to b74adcc Compare February 11, 2026 19:27

nhuantho requested a review from wuchong February 11, 2026 19:28

nhuantho force-pushed the flink-test/issue-2471 branch from 10766c1 to 2a6d9cb Compare February 15, 2026 16:11

nhuantho and others added 2 commits February 28, 2026 01:16

[flink] Add IT test for latest scan startup mode for changelog virtua…

7aaea33

…l table (apache#2477)

[flink] Fix $changelog and $binlog virtual tables fail (NPE) to compl…

6b06d6f

…ete checkpoint (apache#2743)

wuchong force-pushed the flink-test/issue-2471 branch from b66dcd8 to 6b06d6f Compare February 27, 2026 17:17

wuchong approved these changes Feb 27, 2026

View reviewed changes

wuchong linked an issue Feb 27, 2026 that may be closed by this pull request

[flink] $changelog and $binlog virtual tables fail to complete checkpoint (NullPointerException) #2743

Closed

2 tasks

wuchong force-pushed the flink-test/issue-2471 branch 2 times, most recently from 347e0be to 995104f Compare February 28, 2026 03:26

[hotfix] Fix unstable test KvReplicaRestoreITCase.testRowCountRecover…

7c76581

…yAfterFailover

wuchong force-pushed the flink-test/issue-2471 branch from 995104f to 7c76581 Compare February 28, 2026 04:32

wuchong merged commit fbdb7fa into apache:main Feb 28, 2026
14 of 16 checks passed

wuchong pushed a commit that referenced this pull request Feb 28, 2026

[flink] Add IT test for latest scan startup mode for changelog virtua…

ea1f60b

…l table (#2477)

nhuantho deleted the flink-test/issue-2471 branch February 28, 2026 08:04

wuchong pushed a commit that referenced this pull request Feb 28, 2026

[flink] Add IT test for latest scan startup mode for changelog virtua…

1cb95bc

…l table (#2477)

MehulBatra reviewed Feb 28, 2026

View reviewed changes

MehulBatra mentioned this pull request Mar 1, 2026

[flink] Rewrite waitForCheckpoint, getFileBasedCheckpointsConfig to a shared base class #2758

Open

2 tasks

nhuantho mentioned this pull request Mar 2, 2026

[flink] Extract shared checkpoint/savepoint utilities to base class #2761

Open

Ugbot pushed a commit to Ugbot/fluss that referenced this pull request Apr 26, 2026

[flink] Add IT test for latest scan startup mode for changelog virtua…

09ebcf7

…l table (apache#2477)

Conversation

nhuantho commented Jan 25, 2026

Purpose

Brief change log

Tests

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nhuantho Feb 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nhuantho commented Jan 30, 2026

Uh oh!

MehulBatra commented Jan 31, 2026

Uh oh!

MehulBatra commented Feb 9, 2026

Uh oh!

nhuantho commented Feb 9, 2026

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wuchong commented Feb 12, 2026

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

MehulBatra left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nhuantho commented Feb 28, 2026

Uh oh!

wuchong commented Feb 28, 2026

Uh oh!

MehulBatra commented Mar 1, 2026

Uh oh!

nhuantho commented Mar 1, 2026

Uh oh!

MehulBatra commented Mar 1, 2026

Uh oh!

nhuantho commented Mar 1, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

nhuantho Feb 4, 2026 •

edited

Loading

MehulBatra left a comment •

edited

Loading