Flink: Bump DynamicWriteResultSerializer version after format change

[wangyum]

Summary

#14810 changed the serialized format of DynamicWriteResultSerializer by adding a specId field, but did not bump getVersion() from 1. This breaks the SimpleVersionedSerializer versioning contract and makes version 1 ambiguous:

• Iceberg 1.10.x wrote the old WriteTarget layout (tableName, branch, schemaId, specId, upsertMode, equalityFields) tagged as version 1
• Iceberg 1.11.0 (buggy) wrote the new TableKey + specId layout also tagged as version 1 (the version was not bumped)

This causes checkpoint/savepoint restore failures:

Upgrade path	Result
1.10.x → 1.11.0	Fails: StreamCorruptedException: invalid stream header: 00000000
1.11.0 → fixed version (apply this patch)	Fails: OutOfMemoryError (huge equalityFields count from misaligned bytes)

The sibling DynamicCommittableSerializer was correctly versioned in the same commit (bumped to V2 with deserializeV1/deserializeV2). This was simply missed.

Fix

1. Bump getVersion() to 2 — new checkpoints are tagged V2, eliminating future ambiguity
2. deserializeV1WriteTarget — reads the old 1.10.x WriteTarget format
3. deserializeV2 — reads the current TableKey + specId format (same body as the old deserialize(1,...))
4. isV1NewFormat — sniffs V1 bytes to distinguish 1.10.x from 1.11.0, then delegates to the appropriate deserializer

Heuristic: distinguishing the two V1 layouts

After tableName + branch (common to both layouts), the method skips 4 bytes and checks if the next 2 are 0xACED (Java serialization magic):

• New format (1.11.0): the 4 skipped bytes are specId, the next 2 bytes are the start of the Java-serialized WriteResult stream (0xACED) → correctly identified as new format, delegates to deserializeV2
• Old format (1.10.x): the 4 skipped bytes are schemaId, the next 2 bytes are the upper 2 bytes of specId. Since spec IDs are sequentially assigned starting from 0, they are always < 65536, so the upper 2 bytes in big-endian are 0x0000 → correctly identified as old format, uses deserializeV1WriteTarget

A false positive would require specId >= 0xACED0000 (~2.9 billion), which is impossible in practice.

Upgrade paths after fix

From → To	Path
1.10.x → fixed	V1 → heuristic detects old format → deserializeV1WriteTarget
1.11.0 → fixed	V1 → heuristic detects new format → deserializeV2
fixed → fixed	V2 → deserializeV2

Testing

• testRoundtrip — verifies V2 serialize → deserialize roundtrip (existing, uses getVersion() which now returns 2)
• testUnsupportedVersion — verifies unknown versions throw (existing, unchanged)
• testDeserializeV1Format — constructs old V1-format bytes (WriteTarget layout with specId=3) and verifies correct deserialization
• testBuggyV1BytesRestoredWithFixedDeserializer — constructs 1.11.0 V1-format bytes (TableKey + specId, tagged version 1) and verifies the fixed deserializer correctly restores them
• testOldV1FormatWithZeroSpecIdNotMisdetected — edge case: old V1 format with specId=0 must not be misidentified as new format by the heuristic

All tests pass on iceberg-flink-1.20, iceberg-flink-2.0, and iceberg-flink-2.1.

Closes #17314

---

AI Disclosure

• Model: GLM 5.2
• Platform/Tool: opencode
• Human Oversight: fully reviewed
• Prompt Summary: Fix DynamicWriteResultSerializer version not bumped after format change; add heuristic to distinguish 1.10.x and 1.11.0 V1 byte layouts for backward-compatible deserialization.

[wangyum]

cc @aiborodin @pvary

[pvary]

Could you please help me what is the user facing issue and which Iceberg versions are affected? Also what will happen with the users if we fix this now?

[wangyum]

The user-facing issue is:

• 1.10.x writes a checkpoint/savepoint using the old WriteTarget format, tagged version 1
• 1.11.0 reads that checkpoint, sees version == 1, but reads it using the new TableKey + specId format (because the version wasn't bumped). fails with StreamCorruptedException: invalid stream header: 00000000

[pvary]

The user-facing issue is:

• 1.10.x writes a checkpoint/savepoint using the old WriteTarget format, tagged version 1
• 1.11.0 reads that checkpoint, sees version == 1, but reads it using the new TableKey + specId format (because the version wasn't bumped). fails with StreamCorruptedException: invalid stream header: 00000000

If we fix this now, then what will happen when a user runs a job with 1.11.0, and upgrades to the fixed version?

[wangyum]

The user-facing issue is:

• 1.10.x writes a checkpoint/savepoint using the old WriteTarget format, tagged version 1
• 1.11.0 reads that checkpoint, sees version == 1, but reads it using the new TableKey + specId format (because the version wasn't bumped). fails with StreamCorruptedException: invalid stream header: 00000000

If we fix this now, then what will happen when a user runs a job with 1.11.0, and upgrades to the fixed version?

Both 1.10 and 1.11 works if apply this patch.

From → To	Path
1.10.x → fixed	V1 → heuristic detects old format → deserializeV1WriteTarget
1.11.0 → fixed	V1 → heuristic detects new format → deserializeV2
fixed → fixed	V2 → deserializeV2

[pvary]

I'm afraid, there was a mistake with one of our bugfixes. @mxm: Could you please review?

[aiborodin]

@wangyum DynamicWriteResult is not stored in checkpoints/savepoints. It is used purely for runtime communication between DynamicWriter and DynamicWriteResultAggregator and stored in-memory. Only DynamicCommittable is persisted in checkpoints. How does this issue occur?

[mxm]

I'm also struggling to understand how the described issue could have occurred. DynamicWriteResultSerializer is used between DynamicWriter and DynamicWriteResultAggregator to serialize / deserialize DynamicWriteResult.

DynamicWriter flushes WriteResults when the checkpoint barrier arrives. It is stateless. DynamicWriteResultAggregator emits DynamicCommittables which are derived from DynamicWriteResult, but use their own serializer which has proper versioning. It also does not checkpoint any state.

The only plausible scenario where the serializer could pose an issue is with unaligned checkpoints where we checkpoint the in-flight data. We generally do not recommend using unaligned checkpoints because it changes fundamental stream assumptions, e.g. in a channel, where element A is sent before element B, A is processed first. That said, Flink version upgrades should generally not be performed through checkpoints but through savepoints which are aligned by nature.

Maybe you can give more context @wangyum?