[Bug]: Default value in a global side input doesn't work with DirectRunner
- Dominant language
- Java
- Stars
- 8.7k
- Forks
- 4.7k
- Avg merge
- 2d 2h
- Merged PRs (30d)
- 205
Description
### What happened?
When using a side input like the one documented [here](https://beam.apache.org/documentation/patterns/side-inputs/#slowly-updating-global-window-side-inputs), using `org.apache.beam.sdk.transforms.View.AsSingleton#withDefaultValue` to configure a default value doesn't seem to work.
- Is a default value supported when using the global window?
Example:
```java
public class SideInputDefaultTest {
@Test
public void test() {
final DirectOptions options = PipelineOptionsFactory.as(DirectOptions.class);
options.setBlockOnRun(true);
final Pipeline pipeline = Pipeline.create(options);
final Duration sideInputWindowSize = Duration.standardSeconds(10);
final PCollectionView sideInput =
pipeline.apply(GenerateSequence.from(1).withRate(1, sideInputWindowSize))
.apply(Window.into(FixedWindows.of(sideInputWindowSize)))
.apply(Combine.globally(Max.ofLongs()).withoutDefaults())
.apply(ParDo.of(new AlwaysNull()))
.apply(Window.into(new GlobalWindows())
.triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1)))
.discardingFiredPanes())
.apply(View.asSingleton().withDefaultValue("0"));
final Instant start = Instant.now();
final Duration windowSize = Duration.standardSeconds(5);
pipeline.apply(TestStream.create(StringUtf8Coder.of())
.addElements(TimestampedValue.of("a", start))
.addElements(TimestampedValue.of("b", start.plus(windowSize)))
.advanceWatermarkToInfinity())
.apply(Window.into(FixedWindows.of(windowSize)))
.apply(ParDo.of(new ConcatFn(sideInput)).withSideInputs(sideInput));
pipeline.run();
}
private static class ConcatFn extends DoFn {
private final PCollectionView sideInput;
public ConcatFn(final PCollectionView sideInput) {
this.sideInput = sideInput;
}
@ProcessElement
public void process(final ProcessContext context, final OutputReceiver receiver) {
receiver.output(context.element() + " " + context.sideInput(sideInput));
}
}
private static class AlwaysNull extends DoFn {
@ProcessElement
public void process() {
// do nothing to force the default
}
}
}
```
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [X] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
Contributor guide
Research direction
Start with the DirectRunner handling of View.AsSingleton withDefaultValue in the Java SDK, using the SideInputDefaultTest example as the reproduction. Verify the behavior with the global-window side input when no value is produced; done means the configured default value is returned and the regression test passes.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100