[Bug] Native Paimon split assignment can race with final fetcher retirement #334

Native Paimon streaming reads can fail during split assignment when the final fetcher retires concurrently. This is a task failure, not a SQL result mismatch.

Observed failure

PR #333's Paimon 1.0.0 / Flink 1.18.1 upstream run executed 242 unchanged tests with zero assertion failures and one error:

• Test: PrimaryKeyFileStoreTableITCase#testStandAloneLookupJobRandom
• CI log
• The captured job graph shows Source: native-paimon-source feeding the collect sink.

java.util.NoSuchElementException
  at ConcurrentHashMap$ValueIterator.next
  at SingleThreadFetcherManager.getRunningFetcher
  at SingleThreadFetcherManager.addSplits
  at SourceReaderBase.addSplits
  at SourceOperator.handleAddSplitsEvent

The source job failed with NoRestartBackoffTimeStrategy before the result checker could finish. The producer job had already succeeded.

Root cause and deterministic reproducer

The canonical released Flink 1.18.1 and 2.2.1 managers both select the running fetcher with:

return fetchers.isEmpty() ? null : fetchers.values().iterator().next();

The I/O-thread shutdown callback can remove a fetcher independently of mailbox-thread split assignment:

1. The mailbox observes one fetcher, so isEmpty() returns false.
2. The I/O thread removes that last fetcher.
3. The mailbox creates an empty iterator and calls next().

The native Paimon source currently inherits that manager. A controlled map that retires its last entry immediately before values() returns reproduces the exact exception against the actual released manager, without sleeps or random retries. The SQL test is intermittent; passing it once does not disprove the race.

Proposed fix

Use the released source reader's manager extension point for the native Paimon source. Select through a single concurrent-map iterator using hasNext() followed by next(). That iterator retains an observed entry; an empty iterator returns null and lets the existing Flink lifecycle create the replacement fetcher.

Keep split handling, progress events, emitted offsets, checkpoint serialization, queue capacity and shutdown in the released implementation. The Flink 1.18 adapter must continue supplying the same configured handover queue to both manager and reader. No third-party artifacts or SDK sources need modification.

Validation / acceptance

• A deterministic regression must show the released manager throws and the native adapter returns null for the same retirement boundary, and preserve the active-fetcher case.
• Run that regression on both supported Flink lines.
• Run the original lookup SQL test and complete Paimon upstream suite with unchanged assertions and no test retries.
• Retain source/watermark/recovery parity coverage.

The adapter passes 33 focused tests on Paimon 1.0.0 / Flink 1.18.1 / JDK 17, including the deterministic control, source SQL and watermark tests. All 242 unchanged Paimon 1.0 upstream tests pass with zero failures, errors or skips, without test retries. The original failing lookup case also passes in a separate single invocation. Paimon 2.0.0 source/lifecycle/watermark/recovery tests pass on both supported Flink lines with JDK 11 (41 tests per line), including eight checkpoint restoration combinations. Fix commit bee09987 is included in PR #333; the new hosted CI matrix remains pending.