public Optional<Result> execute( List<RowMetaAndData> rows ) throws KettleException {
    if ( rows.isEmpty() || stopped ) {
      return Optional.empty();
    }

    Trans subtrans = this.createSubtrans();
    running.add( subtrans );
    parentTrans.addActiveSubTransformation( subTransName, subtrans );

    // Pass parameter values
    passParametersToTrans( subtrans, rows.get( 0 ) );

    Result result = new Result();
    result.setRows( rows );
    subtrans.setPreviousResult( result );

    subtrans.prepareExecution( this.parentTrans.getArguments() );
    List<RowMetaAndData> rowMetaAndData = new ArrayList<>();
    subtrans.getSteps().stream()
      .filter( c -> c.step.getStepname().equalsIgnoreCase( subStep ) )
      .findFirst()
      .ifPresent( c -> c.step.addRowListener( new RowAdapter() {
        @Override public void rowWrittenEvent( RowMetaInterface rowMeta, Object[] row ) {
          rowMetaAndData.add( new RowMetaAndData( rowMeta, row ) );
        }
      } ) );
    subtrans.startThreads();

    subtrans.waitUntilFinished();
    updateStatuses( subtrans );
    running.remove( subtrans );

    Result subtransResult = subtrans.getResult();
    subtransResult.setRows( rowMetaAndData  );
    releaseBufferPermits( rows.size() );
    return Optional.of( subtransResult );
  }
@Test
  public void blockAndUnblockTwice() throws KettleException, InterruptedException, ExecutionException,
    TimeoutException {
    final ExecutorService executorService = Executors.newFixedThreadPool( 1 );

    TransMeta parentMeta =
      new TransMeta( this.getClass().getResource( "subtrans-executor-parent.ktr" ).getPath(), new Variables() );
    TransMeta subMeta =
      new TransMeta( this.getClass().getResource( "subtrans-executor-sub.ktr" ).getPath(), new Variables() );
    LoggingObjectInterface loggingObject = new LoggingObject( "anything" );
    Trans parentTrans = new Trans( parentMeta, loggingObject );
    SubtransExecutor subtransExecutor =
      new SubtransExecutor( "subtransname", parentTrans, subMeta, true, new TransExecutorParameters(), "", 1 );

    RowMetaInterface rowMeta = parentMeta.getStepFields( "Data Grid" );
    List<RowMetaAndData> rows = Arrays.asList(
      new RowMetaAndData( rowMeta, "Pentaho", 1L ),
      new RowMetaAndData( rowMeta, "Pentaho", 2L ),
      new RowMetaAndData( rowMeta, "Pentaho", 3L ),
      new RowMetaAndData( rowMeta, "Pentaho", 4L ) );

    CompletableFuture<Boolean> acquireThreadRunning = new CompletableFuture();

    Future future = executorService.submit( () -> safeAcquirePermits( subtransExecutor, 8, acquireThreadRunning ) );

    //Acquire the permits before releasing them by waiting for the acquire thread to spin up.
    assertTrue( acquireThreadRunning.get( 5, TimeUnit.SECONDS ) );

    //Release 2*4=8 permits by calling execute twice with 4 rows
    subtransExecutor.execute( rows );
    subtransExecutor.execute( rows );

    final boolean timedOut = !safeWaitForCompletion( future, 5 );
    assertFalse( timedOut );
  }