protected void acceptRows( List<T> rows ) {
    try {
      acceptingRowsSemaphore.acquire();
      waitForSubscribers();
      for ( T row : rows ) {
        streamStep.getSubtransExecutor().acquireBufferPermit();
        streamStep.incrementLinesInput();
        publishProcessor.onNext( row );
      }
    } catch ( InterruptedException e ) {
      logChannel
        .logError( getString( PKG, "BlockingQueueStream.AcceptRowsInterrupt", Arrays.toString( rows.toArray() ) ) );
      Thread.currentThread().interrupt();
    } finally {
      acceptingRowsSemaphore.release();
    }
  }
@Test
  @SuppressWarnings ( "unchecked" )
  public void errorLoggedIfInterruptedInAcceptRows() throws InterruptedException {
    streamSource.acceptingRowsSemaphore = semaphore;
    streamSource.logChannel = logChannel;
    doThrow( new InterruptedException( "interrupt" ) )
      .when( semaphore ).acquire();
    streamSource.acceptRows( singletonList( "new row" ) );
    verify( logChannel ).logError( any() );
    verify( semaphore ).release();
  }