@Override public Iterable<Result> buffer( Flowable<I> flowable ) {
    Flowable<List<I>> buffer = millis > 0
      ? batchSize > 0 ? flowable.buffer( millis, MILLISECONDS, Schedulers.io(), batchSize, ArrayList::new, true )
      : flowable.buffer( millis, MILLISECONDS )
      : flowable.buffer( batchSize );
    return buffer
      .parallel( parallelism, rxBatchCount )
      .runOn( sharedStreamingBatchPoolSize > 0 ? Schedulers.from( sharedStreamingBatchPool ) : Schedulers.io(),
        rxBatchCount )
      .filter( list -> !list.isEmpty() )
      .map( this.bufferFilter ) // apply any filtering for data that should no longer be processed
      .filter( list -> !list.isEmpty() ) // ensure at least one record is left before sending to subtrans
      .map( this::sendBufferToSubtrans )
      .filter( Optional::isPresent )
      .map( Optional::get )
      .sequential()
      .doOnNext( this::failOnError )
      .doOnNext( postProcessor )
      .map( Map.Entry::getValue )
      .blockingIterable();
  }
@Test
  public void abortedSubtransThrowsAnError() throws KettleException {
    Result result1 = new Result();
    result1.setNrErrors( 1 );
    when( subtransExecutor.execute( any()  ) ).thenReturn( Optional.of( result1 ) );
    when( subtransExecutor.getPrefetchCount() ).thenReturn( 10 );
    RowMetaInterface rowMeta = new RowMeta();
    rowMeta.addValueMeta( new ValueMetaString( "field" ) );
    FixedTimeStreamWindow<List> window =
      new FixedTimeStreamWindow<>( subtransExecutor, rowMeta, 0, 2, 1 );
    try {
      window.buffer( Flowable.fromIterable( singletonList( asList( "v1", "v2" ) ) ) ).forEach( result -> { } );
    } catch ( Exception e ) {
      assertEquals(
        getString( BaseStreamStep.class, "FixedTimeStreamWindow.SubtransFailed" ), e.getCause().getMessage().trim() );
    }
  }