apache / apache/beam

FileIO errors on unbounded input with nondefault trigger

Open
#19,286 1 comment 0 reactions 0 assignees View on GitHub
files improvement io java P3
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

`In a pipeline with unbounded input, if a user defines a custom trigger and does not specify a specific non-zero withNumShards, they may see an IllegalArgumentException at runtime due to incompatible windows.`

  For example, consider this compound trigger:

`Window.into(new GlobalWindows())`
`  .triggering(Repeatedly.forever(AfterFirst.of(`
`    AfterPane.elementCountAtLeast(10000),`
`    AfterProcessingTime.pastFirstElementInPane()`
`              .plusDelayOf(Duration.standardMinutes(10)))))`

`  .discardingFiredPanes()`

 
Using that windowing without specifying sharding yields:
 
`Inputs to Flatten had incompatible triggers:``Repeatedly.forever(AfterFirst.of(AfterPane.elementCountAtLeast(10000), AfterProcessingTime.pastFirstElementInPane().plusDelayOf(1 minute))),``Repeatedly.forever(AfterFirst.of(AfterPane.elementCountAtLeast(1), AfterSynchronizedProcessingTime.pastFirstElementInPane()))`
 
Without explicit sharding, WriteFiles creates both a sharded and unsharded collection; the first goes through one GroupByKey while the other goes through 2. These two collections are then flattened together and they have incompatible triggers due to the double-grouped collection using a continuation trigger.
 
If the user instead specifies numShards, then a different code path is followed that avoids this incompatibility.
 
It looks like WriteFiles may need to be implemented differently to avoid combining collections with potentially incompatible triggers.

Imported from Jira [BEAM-6399](https://issues.apache.org/jira/browse/BEAM-6399). Original Jira may contain additional context.
Reported by: jeff.klukas@gmail.com.

Contributor guide

Open the contributing guide

Research direction

Start by tracing the WriteFiles code path used for unbounded input without an explicit non-zero withNumShards, especially where sharded and unsharded collections are combined. Reproduce the custom-trigger example and determine what behavior would avoid incompatible triggers; done means the pipeline no longer raises IllegalArgumentException in that case.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
stream-processing
Issue type
Bug
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.