apache / apache/beam

[Bug]: Default value in a global side input doesn't work with DirectRunner

Open
#24,868 0 comments 0 reactions 0 assignees View on GitHub
bug java P2
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.