Nice bring-up of the Iceberg snapshot read path to Delta parity — the retry-around-execute_df + parallel-parse-via-JobQueue split is clean, and hoisting both helpers into adapterlib is the right call now that two connectors need them (the crate-graph note in the commit message is on point). The parser speedup table (1.72× / 2.71× / 4.42× / 6.19× at 2/4/8/16 workers) is convincing, and the manual chmod 000 retry trace matches the code path. Approve. A few non-blocking things worth a look before you merge:

1. max_retries doc overstates the scope. In feldera-types/src/transport/iceberg.rs (also mirrored into the docs page and OpenAPI) the field is described as covering "opening the table or reading a Parquet file", but in iceberg/src/input.rs the retry loop only wraps execute_df — open_table() at line 462 is called exactly once with no wrapper. That is defensible (opening the table is fail-fast and returns a ControllerError, not a stream failure), but the doc should say so. Either extend the retry to open_table too or narrow the wording to "reading the snapshot data-frame".

2. Backoff sleep is not cancellable. With max_retries defaulting to u32::MAX and a 32s cap, a stuck object store can leave the reader in sleep(backoff_delay).await while the controller is trying to shut the pipeline down. Wrap the sleep in a select! with receiver.changed() (or check for PipelineState::Terminated) so shutdown does not stall for up to 32s per attempt. Delta has the same shape, so if you fix it here, worth doing there too.

3. OpenAPI minimum for num_parsers. The generated schema emits "minimum": 0, but num_parsers == 0 is rejected at construction (correctly). Consider #[schema(minimum = 1)] or equivalent so the API doc matches runtime validation — same for the recommended upper bound if you want to advertise it.

4. Double fork in the parser closure. In execute_df_inner (input.rs ~1030), worker_builder calls input_stream.fork() once per worker, and then the inner FnMut calls input_stream.fork() again on every job. If parse_record_batch drains state via take_all() per call, the inner fork looks redundant. Might be defensive, but worth a comment saying so — or drop it if it truly is redundant.

5. Plural mismatch in the retry log. "error retrieving {descr} after 1 attempts" — cosmetic, but easy to fix ("attempt" for retry_count == 1).

6. Changelog checkbox is unchecked in the PR body. A one-line entry under Unreleased for the new num_parsers / max_retries options would help downstream users notice.

None of these block. Once #1 and #2 are addressed I would ship it.

Re-review at tip 278a65cc.

Additive delta since my last review addresses the nits I raised plus mihaibudiu's:

• max_retries doc scope narrowed to "reading the table snapshot" (was "opening the table or reading a Parquet file", which oversold coverage since open_table() is called once outside the retry loop) — mirrored in the config doc, docs page, and OpenAPI description.
• #[schema(minimum = 1)] added on num_parsers so the generated OpenAPI schema matches the runtime num_parsers == 0 rejection.
• Double-fork in the JobQueue worker closure documented with a comment explaining the outer-worker vs per-job fork rationale.
• "after N attempts" -> "after N attempt(s)" for the single-attempt case.
• Parse error message narrowed to "records read from the Iceberg table" (drops the Parquet-only assumption; Iceberg supports Avro/ORC too — good catch by mihaibudiu).

On the cancellable sleep(backoff_delay) question: reasonable to defer. worker_task at input.rs:295 already wraps worker_task_inner in select! against receiver.wait_for(Terminated), so a stuck sleep inside execute_df is cancelled at the outer level on shutdown. Latency to react is bounded by the outer termination cancelling the whole future tree, not by the sleep itself — the exponential backoff cap of ~32s isn't observable to the controller. I'd still consider making it explicit some day (the pattern-of-two makes reasoning about shutdown a hair harder), but it's not a blocker.

Approving. Nice work turning the review around quickly.

Nice restructure. The refactor is clean, the retry loop now returns healthy on success and unhealthy on terminal failure, and the transaction commit fires exactly once per successful execute_df regardless of retries. Two small notes inline, neither blocking.

LGTM — metrics addition is clean. IcebergPhase enum with repr(u64) + Atomic is a nice pattern, Ordering::Relaxed is correct for independent gauges/counters, tests assert on the new metrics, and docs table entries match the code descriptions exactly.

LGTM. Metrics wiring is clean, ordering is Relaxed throughout (correct for counters/gauges), docs and tests updated.

Re-review at 854b7a1 — one new commit [connectors] retry Iceberg table opens.

Directly addresses the very first nit I raised on the initial review ("extend retry to open_table"). Implementation is careful:

• is_retryable_open_error matches on lower-cased full rate-limit codes/messages (throttlingexception, rate exceeded, slowdown, too many requests, requestlimitexceeded, provisionedthroughputexceeded). No stems, so prose like "not throttled" or "throttling is disabled" won't match.
• open_table_with_retries reuses the existing calculate_backoff_delay + max_retries() accessor for behavior parity with the mid-stream retry loop.
• Permanent errors (missing table, bad credentials, wrong region) fail fast — critical because max_retries defaults to unlimited; otherwise a typo in the table name would loop forever.
• New unit test retryable_open_error_matches_only_throttling covers a real Glue ServiceError payload and four "must not retry" strings including the tricky false-positive prose.

Signed-off, no AI trailer.

APPROVE.