title: GroupIntoBatches not flushing batches, waiting for everything to be processed before outputting
body: <p>I have a pipeline reading from a bounded data source. After some processing transforms, I get a PCollection of String, and I want to group these Strings in batches of 100 elements to send them to an API.</p>
  <p>According to its documentation, <code>GroupIntoBatches</code>, : &quot;Elements are buffered until there are batchSize elements buffered, at which point they are output to the output PCollection.&quot;</p>
  <p>My input collection is very large, and yet my GroupIntoBatches does not output anything until every input element is received :
  <a href="https://i.stack.imgur.com/Vub3n.png" rel="nofollow noreferrer">input / output throughput graphs from Dataflow Runner</a></p>
  <p>At runtime, I can see that nothing is outputted by the <code>GroupAll</code> subtask of the <code>GroupIntoBatches</code>, even though it clearly has received more than 100 elements for one key - only 3 keys and over 100k elements - .</p>
  <p><strong>So the question is :</strong> How can I force the <code>GroupIntoBatches</code> to flush batches as soon as they are full - or every N seconds otherwise ? Do I need to set a window - processing-time window maybe - ? Is it unrelated ?</p>
  <p>I know that <code>GroupIntoBatches</code> relies on KV, so I generated random keys - between 0 and N - where N=3. I also tried to add a window on processing time with triggering every 5 seconds to force it to flush the batches, but it does not change anything, all batches are only outputted at the end of the full processing</p>
  <p>Code sample - with an attempt to use global window - :</p>
  <pre><code>PCollection&lt;KV&lt;Integer, String&gt;&gt; bod = stringCollec.apply(&quot;Add Keys&quot;, ParDo.of(new AssignRandomKeys(3)));
  bod.apply(
     Window.&lt;KV&lt;Integer, String&gt;&gt;into(new GlobalWindows())
   .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardSeconds(1)))).withAllowedLateness(Duration.ZERO).discardingFiredPanes())
  .apply(GroupIntoBatches.ofSize(BATCH_SIZE))
  .apply(&quot;extract&quot;, ParDo.of(new DoFn&lt;KV&lt;Integer, Iterable&lt;String&gt;&gt;, String&gt;() {
        @ProcessElement
        public void process(ProcessContext c) {
         //send one full batch to API
      }));
  </code></pre>
  <p>and my attempt at using processing-time with FixedWindows:</p>
  <pre><code>PCollection&lt;KV&lt;Integer, String&gt;&gt; bod = stringCollec.
  apply(&quot;addTime&quot;,WithTimestamps.of(us -&gt; new Instant()))
  .apply(&quot;Add Keys&quot;, ParDo.of(new AssignRandomKeys(3)));
  bod.apply(
     Window.&lt;KV&lt;Integer, String&gt;&gt;into(FixedWindows.of(Duration.standardSeconds(4)))
   .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardSeconds(1)))).withAllowedLateness(Duration.ZERO).discardingFiredPanes())
  .apply(GroupIntoBatches.ofSize(BATCH_SIZE))
  .apply(&quot;extract&quot;, ParDo.of(new DoFn&lt;KV&lt;Integer, Iterable&lt;String&gt;&gt;, String&gt;() {
        @ProcessElement
        public void process(ProcessContext c) {
         //send one full batch to API
      }));
  </code></pre>