This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new eec5e09090 [flink] Surface a failed asynchronous lookup refresh
instead of dropping it (#9262)
eec5e09090 is described below
commit eec5e09090d64890e2a882ab555b24d06ec51359
Author: ZIHAN DAI <[email protected]>
AuthorDate: Thu Aug 20 12:22:28 2026 +1000
[flink] Surface a failed asynchronous lookup refresh instead of dropping it
(#9262)
---
.../paimon/flink/lookup/FullCacheLookupTable.java | 9 +++++
.../paimon/flink/lookup/LookupTableTest.java | 41 ++++++++++++++++++++++
2 files changed, 50 insertions(+)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
index 08cae53350..76ba9dc346 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
@@ -259,6 +259,15 @@ public abstract class FullCacheLookupTable implements
LookupTable {
@Override
public void refresh() throws Exception {
+ // Surface a failure from a previous asynchronous refresh. Without
this the field is
+ // write-only and the failure is lost: the scan cursor has already
moved past the
+ // snapshot whose rows failed to apply, so nothing retries it and the
cache keeps
+ // serving what it held before. Same shape as
TableCommitImpl.maintain().
+ Exception previousFailure = cachedException.getAndSet(null);
+ if (previousFailure != null) {
+ throw previousFailure;
+ }
+
if (refreshExecutor == null) {
doRefresh();
return;
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
index f3b291b976..3f628c022b 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
@@ -57,6 +57,7 @@ import org.junit.jupiter.api.io.TempDir;
import org.testcontainers.shaded.com.google.common.collect.ImmutableList;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
@@ -69,6 +70,7 @@ import java.util.Random;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
@@ -129,6 +131,45 @@ public class LookupTableTest extends TableTestBase {
return (FileStoreTable) catalog.getTable(identifier);
}
+ /**
+ * An asynchronous refresh records its failure in a field; the next
refresh has to surface it.
+ * The scan cursor has already advanced past the snapshot whose rows
failed to apply, so if the
+ * failure is dropped nothing ever retries it and the cache serves stale
rows with the job still
+ * healthy.
+ */
+ @TestTemplate
+ public void testRefreshRethrowsAFailureFromAnEarlierAsyncRefresh() throws
Exception {
+ FileStoreTable storeTable = createTable(singletonList("f0"), new
Options());
+ FullCacheLookupTable.Context context =
+ new FullCacheLookupTable.Context(
+ storeTable,
+ new int[] {0, 1, 2},
+ null,
+ null,
+ tempDir.toFile(),
+ singletonList("f0"),
+ null);
+ table = FullCacheLookupTable.create(context, 0);
+ table.open();
+
+ Exception failure = new IOException("refresh failed while applying a
snapshot");
+ AtomicReference<Exception> recorded = cachedExceptionOf(table);
+ recorded.set(failure);
+
+ assertThatThrownBy(() -> table.refresh()).isSameAs(failure);
+
+ // Drained, so the same failure does not block every later refresh.
+ assertThat(recorded.get()).isNull();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static AtomicReference<Exception>
cachedExceptionOf(FullCacheLookupTable table)
+ throws Exception {
+ Field field =
FullCacheLookupTable.class.getDeclaredField("cachedException");
+ field.setAccessible(true);
+ return (AtomicReference<Exception>) field.get(table);
+ }
+
@TestTemplate
public void testPkTable() throws Exception {
FileStoreTable storeTable = createTable(singletonList("f0"), new
Options());
