[BEAM-12867] Either Create or DirectRunner fails to produce all elements to the following transform

• Type: Bug
• Status: Open
• Priority: P1
• Resolution: Unresolved
• Component/s: runner-direct
• Assignee: Unassigned
• Reporter: Daniel Collins
• Created: 09/Sep/21 20:52
• Updated: 04/Jun/22 22:30

Description

The following pipeline fails to print out all numbers 1 to 100 when run on DirectRunner in streaming mode.

This was identified implementing org.apache.beam.sdk.io.gcp.pubsublite.ReadWriteIT, which uses a workaround for the bug in either Create or DirectRunner:

private static final int COUNT = 100;

private static AtomicInteger CREATED_COUNT = new AtomicInteger();

public static void run(Pipeline pipeline) {
  PCollection<Integer> indexes = pipeline.apply( "createIndexes",
      Create.of(IntStream.range(0, COUNT).boxed().collect(Collectors.toList())));

indexes.apply(
 "createMessages",
 MapElements.via(
     new SimpleFunction<Integer, Integer>(
         index -> {
           System.err.println("Created message index " + createdCount.incrementAndGet());
           return index;
         }) {}));

  pipeline.run().waitUntilFinish();  // Never terminates
}