You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "nhuantho (via GitHub)" <gi...@apache.org> on 2026/01/25 18:54:56 UTC
[PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
nhuantho opened a new pull request, #2477:
URL: https://github.com/apache/fluss/pull/2477
### Purpose
Linked issue: close https://github.com/apache/fluss/issues/2471
<!-- What is the purpose of the change -->
### Brief change log
* Overloading the `collectRowsWithTimeout` method allows passing `maxWaitTime`.
* Add a test for covering scan.startup.mode='latest'.
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### 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'`
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3890381232
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)
```
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "nhuantho (via GitHub)" <gi...@apache.org>.
nhuantho commented on code in PR #2477:
URL: https://github.com/apache/fluss/pull/2477#discussion_r2745513725
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -377,6 +377,25 @@ public void testChangelogWithScanStartupMode() throws Exception {
.containsExactlyInAnyOrder(
"+I[+I, 3, 1970-01-01T00:00:00.200Z, 4, v4]",
"+I[+I, 4, 1970-01-01T00:00:00.200Z, 5, v5]");
+
+ // 3. Test scan.startup.mode='latest' - should only read new records after subscription
+ String optionsLatest = " /*+ OPTIONS('scan.startup.mode' = 'latest') */";
+ String queryLatest = "SELECT id FROM startup_mode_test$changelog" + optionsLatest;
+ CloseableIterator<Row> rowIterLatest = tEnv.executeSql(queryLatest).collect();
+ List<String> latestResults = new ArrayList<>();
+ 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));
Review Comment:
This suggestion is not correct:
<img width="686" height="340" alt="image" src="https://github.com/user-attachments/assets/7b2ac81e-7c9c-47cd-82d6-d54bf9b2c670" />
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "nhuantho (via GitHub)" <gi...@apache.org>.
nhuantho commented on code in PR #2477:
URL: https://github.com/apache/fluss/pull/2477#discussion_r2762067605
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -377,6 +377,25 @@ public void testChangelogWithScanStartupMode() throws Exception {
.containsExactlyInAnyOrder(
"+I[+I, 3, 1970-01-01T00:00:00.200Z, 4, v4]",
"+I[+I, 4, 1970-01-01T00:00:00.200Z, 5, v5]");
+
+ // 3. Test scan.startup.mode='latest' - should only read new records after subscription
+ String optionsLatest = " /*+ OPTIONS('scan.startup.mode' = 'latest') */";
+ String queryLatest = "SELECT id FROM startup_mode_test$changelog" + optionsLatest;
+ CloseableIterator<Row> rowIterLatest = tEnv.executeSql(queryLatest).collect();
+ List<String> latestResults = new ArrayList<>();
+ 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));
Review Comment:
@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++)
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "nhuantho (via GitHub)" <gi...@apache.org>.
nhuantho commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3872503629
@MehulBatra I resolved conflicts
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "nhuantho (via GitHub)" <gi...@apache.org>.
nhuantho commented on code in PR #2477:
URL: https://github.com/apache/fluss/pull/2477#discussion_r2762067605
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -377,6 +377,25 @@ public void testChangelogWithScanStartupMode() throws Exception {
.containsExactlyInAnyOrder(
"+I[+I, 3, 1970-01-01T00:00:00.200Z, 4, v4]",
"+I[+I, 4, 1970-01-01T00:00:00.200Z, 5, v5]");
+
+ // 3. Test scan.startup.mode='latest' - should only read new records after subscription
+ String optionsLatest = " /*+ OPTIONS('scan.startup.mode' = 'latest') */";
+ String queryLatest = "SELECT id FROM startup_mode_test$changelog" + optionsLatest;
+ CloseableIterator<Row> rowIterLatest = tEnv.executeSql(queryLatest).collect();
+ List<String> latestResults = new ArrayList<>();
+ 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));
Review Comment:
@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++)`
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3979598742
Hi @nhuantho I am already working on the shared base class for the IT, you can help me in review!
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3977415201
@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.
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2477:
URL: https://github.com/apache/fluss/pull/2477#discussion_r2778783386
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -377,6 +377,25 @@ public void testChangelogWithScanStartupMode() throws Exception {
.containsExactlyInAnyOrder(
"+I[+I, 3, 1970-01-01T00:00:00.200Z, 4, v4]",
"+I[+I, 4, 1970-01-01T00:00:00.200Z, 5, v5]");
+
+ // 3. Test scan.startup.mode='latest' - should only read new records after subscription
+ String optionsLatest = " /*+ OPTIONS('scan.startup.mode' = 'latest') */";
+ String queryLatest = "SELECT id FROM startup_mode_test$changelog" + optionsLatest;
+ CloseableIterator<Row> rowIterLatest = tEnv.executeSql(queryLatest).collect();
+ List<String> latestResults = new ArrayList<>();
+ 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));
Review Comment:
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:
1. **Pre-populate** the source table with some existing records.
2. Launch a job:
```sql
INSERT INTO sink SELECT * FROM source /*+ OPTIONS('scan.startup.mode' = 'latest') */;
```
3. **Stop the job with a savepoint** (e.g., using `FlinkOperator#stopWithSavepoint`, as shown in [this example](https://github.com/apache/fluss/blob/main/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceFailOverITCase.java#L243)).
→ At this point, the “latest” offset has been **determined, initialized, and checkpointed** into the state.
4. **Restart the job from the savepoint**. (see example in `FlinkTableSourceFailOverITCase#initTableEnvironment`)
5. **Write new records** into the source table.
6. **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.
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3977238052
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
One concern @wuchong as we have moved to maintaining MiniClusterWithClientResource with before each instead of abstarcttestbase could this hamper the CI runtimes.
Also can we move waitForCheckpoint, getFileBasedCheckpointsConfig to a base class and share it instead of writing individually for both?
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "nhuantho (via GitHub)" <gi...@apache.org>.
nhuantho commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3822887681
@MehulBatra, can you please review the pull request again?
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2477:
URL: https://github.com/apache/fluss/pull/2477#discussion_r2740382297
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -377,6 +377,25 @@ public void testChangelogWithScanStartupMode() throws Exception {
.containsExactlyInAnyOrder(
"+I[+I, 3, 1970-01-01T00:00:00.200Z, 4, v4]",
"+I[+I, 4, 1970-01-01T00:00:00.200Z, 5, v5]");
+
+ // 3. Test scan.startup.mode='latest' - should only read new records after subscription
+ String optionsLatest = " /*+ OPTIONS('scan.startup.mode' = 'latest') */";
+ String queryLatest = "SELECT id FROM startup_mode_test$changelog" + optionsLatest;
+ CloseableIterator<Row> rowIterLatest = tEnv.executeSql(queryLatest).collect();
+ List<String> latestResults = new ArrayList<>();
+ 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));
Review Comment:
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.
```suggestion
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));
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -377,6 +377,25 @@ public void testChangelogWithScanStartupMode() throws Exception {
.containsExactlyInAnyOrder(
"+I[+I, 3, 1970-01-01T00:00:00.200Z, 4, v4]",
"+I[+I, 4, 1970-01-01T00:00:00.200Z, 5, v5]");
+
+ // 3. Test scan.startup.mode='latest' - should only read new records after subscription
+ String optionsLatest = " /*+ OPTIONS('scan.startup.mode' = 'latest') */";
+ String queryLatest = "SELECT id FROM startup_mode_test$changelog" + optionsLatest;
+ CloseableIterator<Row> rowIterLatest = tEnv.executeSql(queryLatest).collect();
+ List<String> latestResults = new ArrayList<>();
+ 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));
+ if (!latestResults.isEmpty()) {
+ break;
+ }
+ }
+ int id = Integer.parseInt(latestResults.getFirst().replaceAll(".*\\[(\\d+)]", "$1"));
Review Comment:
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.
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3871758848
@nhuantho can you resolve the conflicts? I will run the workflow and review
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #2477:
URL: https://github.com/apache/fluss/pull/2477
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "nhuantho (via GitHub)" <gi...@apache.org>.
nhuantho commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3977249572
@MehulBatra if this idea is approved, I can create another issue and implement it.
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3979661736
> @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.
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2477:
URL: https://github.com/apache/fluss/pull/2477#discussion_r2787907431
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -378,6 +411,66 @@ public void testChangelogWithScanStartupMode() throws Exception {
.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
Review Comment:
It will be more clear to extract this test into a separate test method.
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -378,6 +411,66 @@ public void testChangelogWithScanStartupMode() throws Exception {
.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
+ // Create a new source table with 1 bucket for consistent log_offset numbers
+ tEnv.executeSql(
+ "CREATE TABLE startup_mode_test_mode_latest ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ('bucket.num' = '1')");
+ tablePath = TablePath.of(DEFAULT_DB, "startup_mode_test_mode_latest");
+ // Pre-populate the new source table with some existing records.
+ TableResult insertResult =
+ tEnv.executeSql(
+ "INSERT INTO "
+ + tablePath.getTableName()
+ + " SELECT * FROM startup_mode_test");
+ // Stop the job with save point
+ String savepointPath =
+ insertResult
+ .getJobClient()
+ .get()
+ .stopWithSavepoint(
+ false,
+ savepointDir.getAbsolutePath(),
+ SavepointFormatType.CANONICAL)
+ .get();
+ // Init env with savepoint Path
+ tEnv = initTableEnvironment(savepointPath);
+ // Recreate a table
+ tEnv.executeSql(
+ "CREATE TABLE startup_mode_test_mode_latest ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ('bucket.num' = '1')");
+ // Write a new record into the source table
+ tablePath = TablePath.of(DEFAULT_DB, "startup_mode_test_mode_latest");
+ CLOCK.advanceTime(Duration.ofMillis(100));
+ writeRows(conn, tablePath, Arrays.asList(row(6, "v6")), false);
+
+ // Recreate a table
+ tEnv.executeSql(
+ "CREATE TABLE "
Review Comment:
we don't need to re-create the table
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -378,6 +411,66 @@ public void testChangelogWithScanStartupMode() throws Exception {
.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
+ // Create a new source table with 1 bucket for consistent log_offset numbers
+ tEnv.executeSql(
+ "CREATE TABLE startup_mode_test_mode_latest ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ('bucket.num' = '1')");
+ tablePath = TablePath.of(DEFAULT_DB, "startup_mode_test_mode_latest");
+ // Pre-populate the new source table with some existing records.
+ TableResult insertResult =
+ tEnv.executeSql(
+ "INSERT INTO "
+ + tablePath.getTableName()
+ + " SELECT * FROM startup_mode_test");
+ // Stop the job with save point
+ String savepointPath =
+ insertResult
+ .getJobClient()
+ .get()
+ .stopWithSavepoint(
+ false,
+ savepointDir.getAbsolutePath(),
+ SavepointFormatType.CANONICAL)
+ .get();
+ // Init env with savepoint Path
+ tEnv = initTableEnvironment(savepointPath);
+ // Recreate a table
+ tEnv.executeSql(
+ "CREATE TABLE startup_mode_test_mode_latest ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ('bucket.num' = '1')");
+ // Write a new record into the source table
+ tablePath = TablePath.of(DEFAULT_DB, "startup_mode_test_mode_latest");
+ CLOCK.advanceTime(Duration.ofMillis(100));
+ writeRows(conn, tablePath, Arrays.asList(row(6, "v6")), false);
+
+ // Recreate a table
+ tEnv.executeSql(
+ "CREATE TABLE "
+ + tablePath.getTableName()
+ + " ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ('bucket.num' = '1')");
+
+ 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();
Review Comment:
rerun the query of insertResult
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -378,6 +411,66 @@ public void testChangelogWithScanStartupMode() throws Exception {
.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
+ // Create a new source table with 1 bucket for consistent log_offset numbers
+ tEnv.executeSql(
+ "CREATE TABLE startup_mode_test_mode_latest ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ('bucket.num' = '1')");
+ tablePath = TablePath.of(DEFAULT_DB, "startup_mode_test_mode_latest");
+ // Pre-populate the new source table with some existing records.
+ TableResult insertResult =
+ tEnv.executeSql(
+ "INSERT INTO "
+ + tablePath.getTableName()
+ + " SELECT * FROM startup_mode_test");
Review Comment:
Run the query with `startup_mode_test$changelog` with latest startup mode. So we can materilaize the latest offsets into state.
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -378,6 +411,66 @@ public void testChangelogWithScanStartupMode() throws Exception {
.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
+ // Create a new source table with 1 bucket for consistent log_offset numbers
+ tEnv.executeSql(
+ "CREATE TABLE startup_mode_test_mode_latest ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ('bucket.num' = '1')");
+ tablePath = TablePath.of(DEFAULT_DB, "startup_mode_test_mode_latest");
+ // Pre-populate the new source table with some existing records.
+ TableResult insertResult =
+ tEnv.executeSql(
+ "INSERT INTO "
+ + tablePath.getTableName()
+ + " SELECT * FROM startup_mode_test");
+ // Stop the job with save point
+ String savepointPath =
+ insertResult
+ .getJobClient()
+ .get()
+ .stopWithSavepoint(
+ false,
+ savepointDir.getAbsolutePath(),
+ SavepointFormatType.CANONICAL)
+ .get();
+ // Init env with savepoint Path
+ tEnv = initTableEnvironment(savepointPath);
+ // Recreate a table
+ tEnv.executeSql(
+ "CREATE TABLE startup_mode_test_mode_latest ("
Review Comment:
we don't need to re-create the table, the table is already there.
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on code in PR #2477:
URL: https://github.com/apache/fluss/pull/2477#discussion_r2749181448
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -377,6 +377,25 @@ public void testChangelogWithScanStartupMode() throws Exception {
.containsExactlyInAnyOrder(
"+I[+I, 3, 1970-01-01T00:00:00.200Z, 4, v4]",
"+I[+I, 4, 1970-01-01T00:00:00.200Z, 5, v5]");
+
+ // 3. Test scan.startup.mode='latest' - should only read new records after subscription
+ String optionsLatest = " /*+ OPTIONS('scan.startup.mode' = 'latest') */";
+ String queryLatest = "SELECT id FROM startup_mode_test$changelog" + optionsLatest;
+ CloseableIterator<Row> rowIterLatest = tEnv.executeSql(queryLatest).collect();
+ List<String> latestResults = new ArrayList<>();
+ 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));
Review Comment:
Can you please ellaborate your concern?
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "MehulBatra (via GitHub)" <gi...@apache.org>.
MehulBatra commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3827851939
> @MehulBatra, can you please review the pull request again?
Thank you for your PR I will try to review it within this week!
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "nhuantho (via GitHub)" <gi...@apache.org>.
nhuantho commented on code in PR #2477:
URL: https://github.com/apache/fluss/pull/2477#discussion_r2779086900
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -377,6 +377,25 @@ public void testChangelogWithScanStartupMode() throws Exception {
.containsExactlyInAnyOrder(
"+I[+I, 3, 1970-01-01T00:00:00.200Z, 4, v4]",
"+I[+I, 4, 1970-01-01T00:00:00.200Z, 5, v5]");
+
+ // 3. Test scan.startup.mode='latest' - should only read new records after subscription
+ String optionsLatest = " /*+ OPTIONS('scan.startup.mode' = 'latest') */";
+ String queryLatest = "SELECT id FROM startup_mode_test$changelog" + optionsLatest;
+ CloseableIterator<Row> rowIterLatest = tEnv.executeSql(queryLatest).collect();
+ List<String> latestResults = new ArrayList<>();
+ 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));
Review Comment:
Thanks for the suggestion! I’ll follow this approach and submit a PR accordingly.
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "nhuantho (via GitHub)" <gi...@apache.org>.
nhuantho commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3979647571
@MehulBatra sure! If there’s anything I can help with, feel free to tag me.
--
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
Re: [PR] [flink] add a test for ccovering the latest mode in scan.startup.mode configuration tests [fluss]
Posted by "nhuantho (via GitHub)" <gi...@apache.org>.
nhuantho commented on PR #2477:
URL: https://github.com/apache/fluss/pull/2477#issuecomment-3980623765
> @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
<img width="641" height="552" alt="image" src="https://github.com/user-attachments/assets/d76ccf44-8dd0-496a-b869-dc9dbce593e2" />
* Create MiniCluster in beforeAll
<img width="628" height="278" alt="image (4)" src="https://github.com/user-attachments/assets/91a358f7-e682-4a42-8ef3-769925664f1d" />
* Second
* Create MiniCluster in beforeEach
<img width="1164" height="278" alt="image" src="https://github.com/user-attachments/assets/fb879166-bc48-47fd-89e5-b70a61fe85d7" />
* Create MiniCluster in beforeAll
<img width="1184" height="287" alt="image" src="https://github.com/user-attachments/assets/6036a1c8-b844-4141-8f3d-ce303df02a6a" />
* If it is correct, I can create a pr.
--
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