Implementation:ArroyoSystems Arroyo Sink V2 Migration
| Knowledge Sources | |
|---|---|
| Domains | Streaming, Connectors, File_Systems |
| Last Updated | 2026-02-08 08:00 GMT |
Overview
Provides migration functions to convert V1 filesystem sink recovery state and pre-commit data into the V2 checkpoint format, enabling seamless upgrades from the V1 to V2 sink operator.
Description
This module contains two primary migration functions:
migrate_recovery_v1_to_v2 converts a FileSystemDataRecovery (V1 checkpoint) into a FilesCheckpointV2. For each active file in the V1 state, it maps the V1 checkpoint variants to V2 InProgressFile entries:
- Empty - Becomes an empty InProgressFileState::New entry.
- MultiPartNotCreated - Flattens all queued parts and trailing bytes into a single data buffer with InProgressFileState::New state.
- MultiPartInFlight / MultiPartWriterClosed - Extracts completed part content IDs into InProgressFileState::MultipartStarted with trailing bytes as the data buffer. Returns an error if any parts are still in-progress (which should not occur during a normal checkpoint).
- MultiPartWriterUploadCompleted - Returns an error as this state should not appear in recovery data (it belongs in pre-commit data).
migrate_precommit_v1_to_v2 converts a FileToFinish (V1 pre-commit) into a FileToCommit (V2 pre-commit), mapping the multipart upload ID, completed parts, and total size into a FileToCommitType::Multipart variant.
Both functions preserve Iceberg metadata through the migration.
Usage
Called during the V2 sink operator's on_start phase when it detects V1-format recovery state in the state backend. This enables zero-downtime migration from V1 to V2 filesystem sink operators.
Code Reference
Source Location
- Repository: ArroyoSystems_Arroyo
- File: crates/arroyo-connectors/src/filesystem/sink/v2/migration.rs
- Lines: 1-359
Signature
pub fn migrate_recovery_v1_to_v2(
v1: FileSystemDataRecovery,
) -> DataflowResult<FilesCheckpointV2>;
pub fn migrate_precommit_v1_to_v2(
v1: FileToFinish,
) -> FileToCommit;
Import
use arroyo_connectors::filesystem::sink::v2::migration::{
migrate_recovery_v1_to_v2, migrate_precommit_v1_to_v2,
};
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| v1 (recovery) | FileSystemDataRecovery | Yes | V1 recovery state with next_file_index, active_files, and delta_version |
| v1 (precommit) | FileToFinish | Yes | V1 pre-commit data with filename, multipart_upload_id, completed_parts, and size |
Outputs
| Name | Type | Description |
|---|---|---|
| FilesCheckpointV2 | FilesCheckpointV2 | V2 checkpoint state with open_files, file_index, and delta_version |
| FileToCommit | FileToCommit | V2 pre-commit data with path, FileToCommitType::Multipart, and optional Iceberg metadata |
Usage Examples
// During V2 sink initialization, migrate V1 recovery state
let v1_recovery = FileSystemDataRecovery {
next_file_index: 5,
active_files: vec![...],
delta_version: 10,
};
let v2_checkpoint = migrate_recovery_v1_to_v2(v1_recovery)?;
// Migrate pre-commit data
let v1_precommit = FileToFinish { ... };
let v2_commit = migrate_precommit_v1_to_v2(v1_precommit);