PDGGK opened a new pull request, #9262:
URL: https://github.com/apache/paimon/pull/9262
### Purpose
With `lookup.refresh.async = true`, a failed refresh is recorded in a field
that nothing ever reads:
```java
refreshExecutor.submit(
() -> {
try {
doRefresh();
} catch (Exception e) {
LOG.error("Refresh lookup table {} failed",
context.table.name(), e);
cachedException.set(e); // FullCacheLookupTable:316
}
});
```
`git grep cachedException` over the whole tree returns exactly three hits —
the declaration at `:97`, the initialisation at `:149`, and that `set`. There
is no `get`, `getAndSet` or `compareAndSet` anywhere, and `git log -S` shows it
has had no reader since it was introduced. The `catch` also means the submitted
task always completes normally, so the `refreshFuture.get()` at `:300` cannot
resurface it either.
### Why the snapshot is lost rather than retried
`doRefresh()` plans a snapshot and then reads it:
```java
List<Split> splits = reader.nextSplits(); // plans
snapshot N
try (RecordReaderIterator<InternalRow> batch = ...) {
refresh(batch); // applies
snapshot N
}
```
`nextSplits()` goes through `DataTableStreamScan.nextPlan()`, which advances
the cursor **before** handing the plan back:
```java
SnapshotReader.Plan plan = followUpScanner.scan(snapshot, snapshotReader);
currentWatermark = plan.watermark();
nextSnapshotId++; // :232
if (plan.splits().isEmpty()) { continue; }
return plan; // :236
```
So if the failure happens *after* planning — a read error while streaming
the rows, or an `IOException` from the local KV store inside `refreshRow` —
snapshot N is already consumed. `refresh(Iterator)` writes rows one at a time
with no transaction, so N is left half-applied, and nothing re-reads it.
The existing backlog guard does not help in that case, because the cursor
moved: `latestSnapshotId - nextSnapshotId` stays small, so `refresh()` keeps
taking the asynchronous branch and never reaches the synchronous `doRefresh()`
at `:302` whose exceptions propagate.
**Failures raised during planning are already handled and are not what this
changes.** `OutOfRangeException`, and the `ReopenException` that
`LookupDataTableScan.handleOverwriteSnapshot` throws after an `INSERT
OVERWRITE`, are both raised before `nextSnapshotId++`. The cursor is pinned,
the backlog grows past `lookup.refresh.async.pending-snapshot-count` (default
5), the next `refresh()` takes the synchronous branch, and the exception
propagates to `FileStoreLookupFunction.lookup()`'s `catch (OutOfRangeException
| ReopenException) { reopen(); }`. That path self-heals today and is unaffected
here.
### What the user sees
Only in async mode, and only for a post-planning failure: the rows of that
snapshot never enter the cache, the lookup join keeps returning the previous
value for those keys, the job stays healthy, and the only trace is one
`LOG.error`. In synchronous mode the identical failure propagates out of
`lookup()` and the job restarts and rebuilds the cache.
### What changes
Drain the recorded failure at the top of `refresh()` and rethrow it, so the
next refresh cycle reports what the previous one hit:
```java
Exception previousFailure = cachedException.getAndSet(null);
if (previousFailure != null) {
throw previousFailure;
}
```
This is the pattern `TableCommitImpl` already uses for its own asynchronous
executor:
```java
// TableCommitImpl:410-412
if (maintainError.get() != null) {
throw new RuntimeException(maintainError.get());
}
```
`getAndSet` rather than `get` so one failure is reported once rather than
blocking every later refresh.
### Blast radius
`refresh()` already declares `throws Exception`; no signature changes. When
no async refresh has failed — every ordinary cycle — `getAndSet` returns null
and the method continues exactly as before. In synchronous mode
(`lookup.refresh.async = false`, the default at `FlinkConnectorOptions:298`)
the field is never written, so the drain is a no-op there too.
Worth stating plainly: a job currently running with `lookup.refresh.async =
true` and quietly absorbing recurring refresh failures will start failing
visibly. That is the intent — it is what synchronous mode already does — but it
is a behaviour change for such a deployment.
### Test
`LookupTableTest#testRefreshRethrowsAFailureFromAnEarlierAsyncRefresh`
records a failure the way the executor's catch does, then asserts the next
`refresh()` throws that exact instance and that the field has been cleared.
Removing the drain fails it; `LookupTableTest` as a whole is 48 tests, 0
failures.
The field is private with no accessor, so the test reaches it reflectively —
the same approach several existing tests take (`IcebergCommitCallbackTest`,
`SortBufferWriteBufferOverflowTest`, `PrimaryKeyIndexWriteTest`). Happy to add
a `@VisibleForTesting` accessor instead if you would rather.
### API and Format
No change to any option, on-disk format or public signature.
--
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]
