[1.11.x] Spark: Test streaming recovery across replacement metadata

[timothyw553]

Problem

An Iceberg Spark streaming checkpoint stores a snapshot ID and a position inside that snapshot. If table metadata is replaced with an independent snapshot history, the old snapshot ID is no longer on the current lineage and the stream cannot restart.

For example:

Old history:          O1 -> O2 -> O3 -> O4
                           ^
                           checkpoint

Replacement history: N1 -> K -> A1 -> A2

Restarting against the replacement history fails because O2 cannot be resolved.

What this draft proves

This is a test-only proof of concept, not a production recovery command or service.

The tests construct a separate, read-only recovery table:

O1 -> O2 -> O3 -> O4 -> K* -> A1
                           ^
                       REPLACE

The recovery table keeps the old snapshots and manifest lists unchanged. K* keeps the final replacement snapshot ID and v3 manifests, but points to O4 and uses the REPLACE operation. Spark therefore finishes the partially consumed old snapshot, reads later old appends, skips K*, and reaches the real post-replacement snapshot A1.

After Spark commits A1, the same checkpoint restarts against the original live table. Because A1 exists in both histories, streaming continues from the live descendants without replaying earlier rows.

The tests cover both an in-process catalog and an HTTP REST catalog with catalog caching enabled. They also verify that:

• a normal restart first fails with the missing-snapshot error;
• old manifest lists remain byte-for-byte unchanged;
• v3 row IDs remain unique and contiguous;
• the bridge emits no rows;
• a live source write can commit while recovery is active;
• the handoff reads every row exactly once across another restart.

The REST fixture serves metadata that is already registered in the test catalog. It does not test production REST registration, authentication, or credential vending.

Current assumptions

The fixture is intentionally limited to an append-only, unpartitioned table without delete files. A production implementation must fail safely unless the old terminal snapshot and final replacement snapshot describe the same physical table state. It must also handle metadata retention, schema and partition evolution, delete files, encryption, authentication, and temporary recovery-table lifecycle.

The test demonstrates a handoff between two table identifiers with the same source provider, schema, and source position. It does not establish a general Spark compatibility contract for arbitrary source changes.

Tests

./gradlew --offline -DsparkVersions=4.0 \
  :iceberg-spark:iceberg-spark-4.0_2.13:spotlessJavaCheck \
  :iceberg-spark:iceberg-spark-4.0_2.13:test \
  --tests org.apache.iceberg.spark.source.TestStreamingSnapshotRecovery \
  --no-daemon

Result: 2 tests passed and formatting passed.

AI assistance

AI assistance was used to scaffold and pressure-test this prototype. I reviewed the complete diff, corrected the snapshot-summary and row-ID validation, and ran the focused tests and formatting check. The production limitations and source-handoff compatibility noted above remain open design questions.