Remove the Fluss Nexmark rung's bounded-scan workarounds (blocked on upstream scan.bounded.mode)

[jordepic]

What

The Nexmark Fluss rung carries special measurement machinery no other source rung needs, and it should be removed:

• the count-to-Nth-row cancel with a bounded-generator calibration (flussTargetRows),
• the far-future watermark sentinel row that closes the final event-time windows,
• the traced table + poison pair + marker sink for queries whose changelog cardinality is non-deterministic (q4/q9) or zero (q21),
• the skip for q12 — a proctime window flushes early only at end-of-input, which no data row can fake, so the query can't be measured on this rung at all.

The end state: the Fluss rung measures exactly like the Kafka rung — both engines read bounded, the job ends by itself, the measurement is end-to-end time — and all 23 runnable queries measure uniformly, q12 included.

Blocker: an upstream Fluss change

All of that machinery exists for one reason: the Fluss Flink connector has no bounded read for log tables — no scan.bounded.mode, no stopping-offset surface (bounded reads exist only via the lake source in batch mode and the pushed-LIMIT RPC). So the fix is an upstream contribution to apache/fluss, not local work: Kafka-shaped scan.bounded.mode = 'latest-offset' (and plausibly timestamp) for log-table scans.

Every ingredient already exists upstream:

• OffsetsInitializer.latest() already resolves per-bucket end offsets (it backs scan.startup.mode = 'latest') — a bounded read resolves the same offsets at enumerator start and stamps them as stopping offsets instead.
• LogSplit already carries a stoppingOffset (used by the hybrid snapshot→log handoff), and the reader's finished-split machinery already terminates on it.
• The enumerator already diffs/assigns per-bucket splits; boundedness is one more field at split construction plus Boundedness.BOUNDED on the source.

Our side needs nothing new: the native Fluss source reuses Fluss's enumerator verbatim (a bounded assignment flows through unchanged), and the native split reader already implements the stopping-offset path (immediate-finish for empty ranges, batch slicing at the stop, finished-split reporting) — currently exercised only by tests.

When the upstream change lands in a Fluss release

• Re-point the benchmark's Fluss rung at the Kafka rung's bounded method and delete the machinery above, including the q12 skip reason.
• The stock baseline needs the released connector version — do not ship a patched fluss-flink jar (unlike the fluss-rs cargo pin, a Maven fork is not a self-contained build).