[BEAM-8577] FileSystems may have not be initialized during ResourceId deserialization

• Type: Bug
• Status: Triage Needed
• Priority: P2
• Resolution: Fixed
• Affects Version/s: 2.16.0
• Fix Version/s: 2.19.0
• Component/s: runner-flink
• Assignee: David Morávek
• Reporter: David Morávek
• Created: 07/Nov/19 14:27
• Updated: 13/Apr/23 11:00
• Resolved: 06/Jan/20 18:41

Description

• FileSystems use static registration using FileSystems#setDefaultPipelineOptions method.
• #setDefaultPipelineOptions is called either when deserializaing SerializablePipelineOptions or during opening of various beam operators.
• FileIO#matchAll is expanded using Reshuffle.viaRandomKey().
• Reshuffle is implemented using .rebalance, that doesn't have a "RichFunction" lifecycle, so we need to find another way to register FileSystems, as the deserialization may happen before other "rich operators" get executed on particular task manager.

This results in random pipeline fails as the task assignment is not deterministic.

We can workaround this, by registering FileSystems during coder deserialization.

Issue Links

• relates to: BEAM-2712 SerializablePipelineOptions should not call FileSystems.setDefaultPipelineOptions.
• supercedes: BEAM-8303 Filesystems not properly registered using FileIO.write() on FlinkRunner
• links to: GitHub Pull Request #10027